import warnings
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from quantfinlab.dataio.panel import load_yfinance_panel, prices_to_returns_panel
from quantfinlab.portfolio import covariance, expected_returns
from quantfinlab.portfolio.universe import (
build_liquid_universe_by_date,
clean_close_volume_panels,
make_rebalance_dates,
)
from quantfinlab.portfolio.walkforward import run_walkforward_grid
from quantfinlab.portfolio.hrp import hrp_weight_frame
from quantfinlab.backtest.portfolio import run_many_weights_backtests
from quantfinlab.portfolio.selection import build_strategy_summary, performance_metrics
from quantfinlab.reports.risk_report import risk_report
from quantfinlab.portfolio.network import (central_peripheral_weights, centrality_table, corr_distance,
dense_network, dependence_to_distance, kendall_to_t_copula_corr,
mst_network, network_diversifier_weights, network_score,
pairwise_corr_for_weights, pmfg_network, pseudo_observations,
scale_01, select_t_copula_nu, shrink_corr,
student_t_tail_dependence)
from quantfinlab.plotting.portfolio import (corr_heatmap, distance_heatmap, network_graph,
performance_heatmap, score_heatmap, tail_heatmap,
weight_heatmap)
annualization = 252
rf_annual = 0.04
rf_daily = (1.0 + rf_annual) ** (1.0 / annualization) - 1.0
w_max = 0.10
panel = load_yfinance_panel("../data/hkex_close_volume.parquet", fields=("close", "volume"),
source="hkex_close_volume", start="2016-01-01")
close, volume = clean_close_volume_panels(
panel["close"], panel["volume"], start="2016-01-01", min_price=1.0
)
ret = prices_to_returns_panel(close).replace([np.inf, -np.inf], np.nan).dropna(how="all")
monthly_candidates = make_rebalance_dates(close.index, freq="M", min_history_days=504)
sampled_dates = pd.DatetimeIndex(monthly_candidates[::3])
universe = build_liquid_universe_by_date(close=close, volume=volume, rebalance_dates=sampled_dates,
top_n=100, liquidity_lookback=252, min_listing_days=504,
min_obs=220, min_price=1.0)
dates = pd.DatetimeIndex([dt for dt in sampled_dates if dt in universe])
grid = run_walkforward_grid(
returns=ret,
close=close,
volume=volume,
rebalance_dates=dates,
universe_by_date=universe,
mu_models={"Momentum": expected_returns.momentum_mu},
cov_models={"LedoitWolf": covariance.ledoit_wolf_covariance},
strategy_specs=[
{"name": "Equal Weight", "optimizer": "EW"},
{"name": "MinVar", "optimizer": "MinVar", "cov_model": "LedoitWolf"},
],
cov_lookback=252,
mu_lookback=252,
min_cov_observations=220,
min_mu_observations=220,
max_weight=w_max,
min_weight=0.0,
trading_cost_bps=10,
turnover_penalty_bps=10,
fallback="equal",
rf_daily=rf_daily,
)
w_equal = grid.weights["Equal Weight"].copy()
w_minvar = grid.weights["MinVar"].copy()
w_hrp = hrp_weight_frame(grid.cache, grid.metadata["rebalance_dates"], cov_model="LedoitWolf",
linkage_method="average", w_min=0.0, w_max=w_max)
first = dates[0]
window_first = ret.loc[:first, universe[first]["tickers"]].tail(252).dropna(axis=0, how="any")
corr_first = shrink_corr(window_first)
dist_first = corr_distance(corr_first)
u_first = pseudo_observations(window_first)
nu_first = select_t_copula_nu(u_first, nu_grid=(4, 7, 15), max_pairs=40)
tail_first = student_t_tail_dependence(kendall_to_t_copula_corr(u_first), nu_first)
tail_dist_first = dependence_to_distance(tail_first)
first_networks = {"Correlation Dense": dense_network(corr_first, distance=dist_first),
"Correlation PMFG": pmfg_network(corr_first, distance=dist_first),
"Correlation MST": mst_network(corr_first, distance=dist_first),
"Copula Dense": dense_network(tail_first, distance=tail_dist_first),
"Copula PMFG": pmfg_network(tail_first, distance=tail_dist_first),
"Copula MST": mst_network(tail_first, distance=tail_dist_first)}
store, comparison_rows = {}, {}
for dt in dates:
window = ret.loc[:dt, universe[dt]["tickers"]].tail(252).dropna(axis=0, how="any")
if window.shape[0] < 220 or window.shape[1] < 50:
continue
corr = shrink_corr(window)
corr_dist = corr_distance(corr)
pseudo = pseudo_observations(window)
nu = select_t_copula_nu(pseudo, nu_grid=(4, 7, 15), max_pairs=40)
tail = student_t_tail_dependence(kendall_to_t_copula_corr(pseudo), nu)
tail_dist = dependence_to_distance(tail)
graphs = {("correlation", "pmfg"): pmfg_network(corr, distance=corr_dist),
("correlation", "mst"): mst_network(corr, distance=corr_dist),
("copula", "pmfg"): pmfg_network(tail, distance=tail_dist),
("copula", "mst"): mst_network(tail, distance=tail_dist)}
centralities = {key: centrality_table(graph) for key, graph in graphs.items()}
for (dependence, network_type), centrality in centralities.items():
for measure in ["degree", "eigenvector", "betweenness", "closeness", "combined"]:
central = scale_01(centrality[measure])
for direction, selection_score in {"central": central, "peripheral": 1.0 - central}.items():
key = f"{direction}|{dependence}|{network_type}|{measure}"
comparison_rows.setdefault(key, {})[dt] = central_peripheral_weights(
selection_score, returns=window, n_stocks=20, max_weight=w_max
)
store[dt] = {"window": window, "corr": corr, "corr_dist": corr_dist,
"tail": tail, "tail_dist": tail_dist, "nu": nu,
"graphs": graphs, "centralities": centralities}
model_dates = (
pd.DatetimeIndex(sorted(store))
.intersection(w_equal.index)
.intersection(w_minvar.index)
.intersection(w_hrp.index)
)
comparison_weights = {
key: pd.DataFrame.from_dict(rows, orient="index").fillna(0.0)
for key, rows in comparison_rows.items()
}
comparison_cols = sorted(set().union(*(set(frame.columns) for frame in comparison_weights.values())))
comparison_weights = {
name: frame.reindex(index=model_dates, columns=comparison_cols).fillna(0.0)
for name, frame in comparison_weights.items()
}
comparison_results = run_many_weights_backtests(comparison_weights,
returns=ret.reindex(columns=comparison_cols).fillna(0.0), cost_bps=10,
w_min=0.0, w_max=w_max, long_only=True,
weight_timing="next_close")
comparison_table = []
for name, result in comparison_results.items():
direction, dependence, network_type, measure = name.split("|")
metrics = performance_metrics(result.net_returns, result.net_values, rf_daily=rf_daily, annualization=annualization)
comparison_table.append(
{"direction": direction, "dependence": dependence, "network": network_type,
"centrality": measure, "Sharpe": metrics["Sharpe"], "Max Drawdown": metrics["Max Drawdown"],
"Calmar": metrics["Calmar"], "turnover": float(result.turnover.mean()),
"avg_pairwise_corr": float(pairwise_corr_for_weights(ret, result.weights).mean())}
)
comparison = pd.DataFrame(comparison_table)
comparison["sharpe"] = comparison["Sharpe"]
comparison["max_drawdown"] = comparison["Max Drawdown"]
def parse_choice(name):
direction, dependence, network_type, measure = name.split("|")
return direction, dependence, network_type, measure
def rolling_choice(results, dt, direction, lookback_years=2, min_obs=252):
fallback = f"{direction}|copula|pmfg|combined"
start = pd.Timestamp(dt) - pd.DateOffset(years=lookback_years)
rows = []
for name, result in results.items():
if not name.startswith(f"{direction}|"):
continue
returns_window = result.net_returns.loc[
(result.net_returns.index >= start) & (result.net_returns.index < dt)
].dropna()
if len(returns_window) < min_obs:
continue
nav_window = (1.0 + returns_window).cumprod()
metrics = performance_metrics(returns_window, nav_window, rf_daily=rf_daily,
annualization=annualization)
turnover_window = result.turnover.loc[
(result.turnover.index >= start) & (result.turnover.index < dt)
].dropna()
rows.append(
{"name": name, "Sharpe": metrics["Sharpe"], "Max Drawdown": metrics["Max Drawdown"],
"Calmar": metrics["Calmar"],
"turnover": float(turnover_window.mean()) if len(turnover_window) else np.nan}
)
if not rows:
return fallback, {"Sharpe": np.nan, "Max Drawdown": np.nan, "source": "fallback"}
table = pd.DataFrame(rows).replace([np.inf, -np.inf], np.nan).dropna(subset=["Sharpe"])
if table.empty:
return fallback, {"Sharpe": np.nan, "Max Drawdown": np.nan, "source": "fallback"}
best = table.sort_values(
["Sharpe", "Max Drawdown", "turnover"],
ascending=[False, False, True],
).iloc[0]
return best["name"], {"Sharpe": float(best["Sharpe"]), "Max Drawdown": float(best["Max Drawdown"]),
"source": "rolling"}
def train_choice(results, direction, train_end, min_obs=252):
fallback = f"{direction}|copula|pmfg|combined"
rows = []
for name, result in results.items():
if not name.startswith(f"{direction}|"):
continue
returns_window = result.net_returns.loc[result.net_returns.index < train_end].dropna()
if len(returns_window) < min_obs:
continue
nav_window = (1.0 + returns_window).cumprod()
metrics = performance_metrics(returns_window, nav_window, rf_daily=rf_daily,
annualization=annualization)
turnover_window = result.turnover.loc[result.turnover.index < train_end].dropna()
rows.append(
{"name": name, "Sharpe": metrics["Sharpe"], "Max Drawdown": metrics["Max Drawdown"],
"turnover": float(turnover_window.mean()) if len(turnover_window) else np.nan}
)
if not rows:
return fallback
table = pd.DataFrame(rows).replace([np.inf, -np.inf], np.nan).dropna(subset=["Sharpe"])
if table.empty:
return fallback
return table.sort_values(["Sharpe", "Max Drawdown", "turnover"], ascending=[False, False, True]).iloc[0]["name"]
selector_train_end = model_dates[min(8, len(model_dates) - 1)]
central_fixed = "central|copula|pmfg|combined"
peripheral_train = train_choice(comparison_results, "peripheral", selector_train_end)
def stability_gated_choice(results, dt, direction, train_selection, threshold=0.50):
if pd.Timestamp(dt) <= selector_train_end:
return f"{direction}|copula|pmfg|combined", {"Sharpe": np.nan, "source": "fallback"}
rolling_name, rolling_info = rolling_choice(results, dt, direction)
if not np.isfinite(rolling_info["Sharpe"]) or rolling_info["Sharpe"] < threshold:
return train_selection, {"Sharpe": rolling_info["Sharpe"], "source": "train-fixed"}
return rolling_name, rolling_info
selection_records = []
central_rows, peripheral_rows, diversifier_rows = {}, {}, {}
for dt in model_dates:
central_choice = central_fixed
central_info = {"Sharpe": np.nan, "source": "fixed"}
peripheral_choice, peripheral_info = stability_gated_choice(
comparison_results, dt, "peripheral", peripheral_train
)
network_choice, network_info = stability_gated_choice(
comparison_results, dt, "peripheral", peripheral_train
)
selection_records.append(
{"date": dt, "central_choice": central_choice, "central_sharpe": central_info["Sharpe"],
"central_source": central_info["source"], "peripheral_choice": peripheral_choice,
"peripheral_sharpe": peripheral_info["Sharpe"], "peripheral_source": peripheral_info["source"],
"network_choice": network_choice, "network_sharpe": network_info["Sharpe"],
"network_source": network_info["source"]}
)
record = store[dt]
_, dependence, network_type, measure = parse_choice(central_choice)
centrality = record["centralities"][(dependence, network_type)]
central_score = scale_01(centrality[measure])
central_rows[dt] = central_peripheral_weights(
central_score, returns=record["window"], side="central", n_stocks=20, max_weight=w_max
)
_, dependence, network_type, measure = parse_choice(peripheral_choice)
centrality = record["centralities"][(dependence, network_type)]
periphery = 1.0 - scale_01(centrality[measure])
peripheral_rows[dt] = central_peripheral_weights(
periphery, returns=record["window"], side="peripheral", n_stocks=20, max_weight=w_max
)
_, dependence, network_type, measure = parse_choice(network_choice)
centrality = record["centralities"][(dependence, network_type)]
periphery = 1.0 - scale_01(centrality[measure])
scores = network_score(periphery, returns=record["window"], momentum_window=126,
momentum_skip=0, volatility_window=126, drawdown_window=252)
diversifier_rows[dt] = network_diversifier_weights(
scores, returns=record["window"], n_stocks=25, max_weight=w_max
)
selection = pd.DataFrame(selection_records).set_index("date")
w_central = pd.DataFrame.from_dict(central_rows, orient="index").fillna(0.0)
w_peripheral = pd.DataFrame.from_dict(peripheral_rows, orient="index").fillna(0.0)
w_diversifier = pd.DataFrame.from_dict(diversifier_rows, orient="index").fillna(0.0)
common_index = (
w_diversifier.index.intersection(w_equal.index).intersection(w_minvar.index).intersection(w_hrp.index)
)
columns = sorted(
set(w_equal.columns)
| set(w_minvar.columns)
| set(w_hrp.columns)
| set(w_central.columns)
| set(w_peripheral.columns)
| set(w_diversifier.columns)
)
weights = {"Equal Weight": w_equal.reindex(index=common_index, columns=columns).fillna(0.0),
"MinVar": w_minvar.reindex(index=common_index, columns=columns).fillna(0.0),
"HRP": w_hrp.reindex(index=common_index, columns=columns).fillna(0.0),
"Most Central 20": w_central.reindex(index=common_index, columns=columns).fillna(0.0),
"Most Peripheral 20": w_peripheral.reindex(index=common_index, columns=columns).fillna(0.0),
"Network Diversifier": w_diversifier.reindex(index=common_index, columns=columns).fillna(0.0)}
results_hk = run_many_weights_backtests(weights, returns=ret.reindex(columns=columns).fillna(0.0), cost_bps=10,
w_min=0.0, w_max=w_max, long_only=True,
weight_timing="next_close", rf_daily=rf_daily)
summary = build_strategy_summary(results_hk, rf_daily=rf_daily, annualization=annualization)
display(summary[["CAGR", "Vol", "Sharpe", "Max Drawdown", "Calmar", "Turnover", "Cost Drag", "Effective N"]].round(4))
display(selection.tail(8))
display(
pd.concat(
{"Most Central 20": selection["central_choice"].value_counts(),
"Most Peripheral 20": selection["peripheral_choice"].value_counts(),
"Network Diversifier": selection["network_choice"].value_counts()},
axis=1,
).fillna(0).astype(int)
)
latest_dt = common_index[-1]
latest = store[latest_dt]
latest_network_choice = selection.loc[latest_dt, "network_choice"]
_, latest_dependence, latest_network_type, latest_measure = parse_choice(latest_network_choice)
centrality_hk = latest["centralities"][(latest_dependence, latest_network_type)]
central_score_hk = scale_01(centrality_hk[latest_measure])
periphery_hk = 1.0 - central_score_hk
network_scores = network_score(periphery_hk, returns=latest["window"])
score_frame = pd.concat(
{"centrality": central_score_hk, "periphery": periphery_hk, "network score": network_scores,
"weight": weights["Network Diversifier"].loc[latest_dt]},
axis=1,
).sort_values("network score", ascending=False).head(25)
weight_frame = pd.concat(
{"Most Central 20": weights["Most Central 20"].loc[latest_dt],
"Most Peripheral 20": weights["Most Peripheral 20"].loc[latest_dt],
"Network Diversifier": weights["Network Diversifier"].loc[latest_dt]},
axis=1,
).fillna(0.0)
nav = pd.concat({name: result.net_values for name, result in results_hk.items()}, axis=1)
drawdowns = nav / nav.cummax() - 1.0
fig, axes = plt.subplots(2, 4, figsize=(22, 9))
axes = axes.ravel()
network_graph(axes[0], latest["graphs"][("correlation", "pmfg")], title="HKEX Latest Correlation PMFG")
network_graph(axes[1], latest["graphs"][("copula", "pmfg")], title="HKEX Latest Copula PMFG")
score_heatmap(axes[2], score_frame, title="HKEX Diversifier Scores")
weight_heatmap(axes[3], weight_frame, title="HKEX Latest Network Weights")
nav.plot(ax=axes[4], title="HKEX Costed NAV")
drawdowns.plot(ax=axes[5], title="HKEX Drawdowns")
performance_heatmap(axes[6], comparison, value="sharpe", title="HKEX Centrality Sharpe")
performance_heatmap(axes[7], comparison, value="max_drawdown", title="HKEX Centrality Max Drawdown")
for ax in axes[4:6]:
ax.grid(True, alpha=0.25)
fig.tight_layout()
plt.show()
warnings.filterwarnings("ignore", message="obj.round has no effect with datetime, timedelta, or period dtypes.*", category=UserWarning)
risk = risk_report(
objects={name: result.net_returns for name, result in results_hk.items()},
rf_daily=rf_daily,
include={"performance_tables": False, "shape_tables": False, "drawdowns": False,
"drawdown_episodes": True, "var_es": True, "var_backtest": False,
"stress": True, "capm": False, "rolling_beta": False,
"correlation": False, "attribution": False, "exec_bullets": False},
var_settings={"alpha": 0.05, "methods": ["hist", "cf"], "lookback": 252},
stress_settings={
"windows": {"2018_q4": ("2018-10-01", "2018-12-31"), "2020_covid": ("2020-02-20", "2020-04-30"),
"2022_inflation": ("2022-01-03", "2022-10-31"), "2023_rebound": ("2023-01-01", "2023-12-31")},
"worst_only": False,
},
output={"display_tables": True, "show_figures": False, "round_tables": 4},
)
plt.close("all")