from pathlib import Path
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_hk = 252
rf_annual_hk = 0.04
rf_daily_hk = (1.0 + rf_annual_hk) ** (1.0 / annualization_hk) - 1.0
w_max_hk = 0.10
data_path = Path("../data")
if not (data_path / "hkex_close_volume.parquet").exists():
data_path = Path("data")
panel = load_yfinance_panel(
data_path / "hkex_close_volume.parquet",
fields=("close", "volume"),
source="hkex_close_volume",
start="2016-01-01",
)
close_hk, volume_hk = clean_close_volume_panels(
panel["close"], panel["volume"], start="2016-01-01", min_price=1.0
)
ret_hk = prices_to_returns_panel(close_hk).replace([np.inf, -np.inf], np.nan).dropna(how="all")
monthly_candidates_hk = make_rebalance_dates(close_hk.index, freq="M", min_history_days=504)
sampled_dates_hk = pd.DatetimeIndex(monthly_candidates_hk[::3])
universe_hk = build_liquid_universe_by_date(
close=close_hk,
volume=volume_hk,
rebalance_dates=sampled_dates_hk,
top_n=100,
liquidity_lookback=252,
min_listing_days=504,
min_obs=220,
min_price=1.0,
)
dates_hk = pd.DatetimeIndex([dt for dt in sampled_dates_hk if dt in universe_hk])
grid_hk = run_walkforward_grid(
returns=ret_hk,
close=close_hk,
volume=volume_hk,
rebalance_dates=dates_hk,
universe_by_date=universe_hk,
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_hk,
min_weight=0.0,
trading_cost_bps=10,
turnover_penalty_bps=10,
fallback="equal",
rf_daily=rf_daily_hk,
)
w_equal_hk = grid_hk.weights["Equal Weight"].copy()
w_minvar_hk = grid_hk.weights["MinVar"].copy()
w_hrp_hk = hrp_weight_frame(
grid_hk.cache,
grid_hk.metadata["rebalance_dates"],
cov_model="LedoitWolf",
linkage_method="average",
w_min=0.0,
w_max=w_max_hk,
)
first_hk = dates_hk[0]
window_first_hk = ret_hk.loc[:first_hk, universe_hk[first_hk]["tickers"]].tail(252).dropna(axis=0, how="any")
corr_first_hk = shrink_corr(window_first_hk)
dist_first_hk = corr_distance(corr_first_hk)
u_first_hk = pseudo_observations(window_first_hk)
nu_first_hk = select_t_copula_nu(u_first_hk, nu_grid=(4, 7, 15), max_pairs=40)
tail_first_hk = student_t_tail_dependence(kendall_to_t_copula_corr(u_first_hk), nu_first_hk)
tail_dist_first_hk = dependence_to_distance(tail_first_hk)
first_networks_hk = {
"Correlation Dense": dense_network(corr_first_hk, distance=dist_first_hk),
"Correlation PMFG": pmfg_network(corr_first_hk, distance=dist_first_hk),
"Correlation MST": mst_network(corr_first_hk, distance=dist_first_hk),
"Copula Dense": dense_network(tail_first_hk, distance=tail_dist_first_hk),
"Copula PMFG": pmfg_network(tail_first_hk, distance=tail_dist_first_hk),
"Copula MST": mst_network(tail_first_hk, distance=tail_dist_first_hk),
}
store_hk, comparison_rows_hk = {}, {}
for dt in dates_hk:
window = ret_hk.loc[:dt, universe_hk[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_hk.setdefault(key, {})[dt] = central_peripheral_weights(
selection_score, returns=window, n_stocks=20, max_weight=w_max_hk
)
store_hk[dt] = {
"window": window,
"corr": corr,
"corr_dist": corr_dist,
"tail": tail,
"tail_dist": tail_dist,
"nu": nu,
"graphs": graphs,
"centralities": centralities,
}
model_dates_hk = (
pd.DatetimeIndex(sorted(store_hk))
.intersection(w_equal_hk.index)
.intersection(w_minvar_hk.index)
.intersection(w_hrp_hk.index)
)
comparison_weights_hk = {
key: pd.DataFrame.from_dict(rows, orient="index").fillna(0.0)
for key, rows in comparison_rows_hk.items()
}
comparison_cols_hk = sorted(set().union(*(set(frame.columns) for frame in comparison_weights_hk.values())))
comparison_weights_hk = {
name: frame.reindex(index=model_dates_hk, columns=comparison_cols_hk).fillna(0.0)
for name, frame in comparison_weights_hk.items()
}
comparison_results_hk = run_many_weights_backtests(
comparison_weights_hk,
returns=ret_hk.reindex(columns=comparison_cols_hk).fillna(0.0),
cost_bps=10,
w_min=0.0,
w_max=w_max_hk,
long_only=True,
weight_timing="next_close",
)
comparison_table_hk = []
for name, result in comparison_results_hk.items():
direction, dependence, network_type, measure = name.split("|")
metrics = performance_metrics(result.net_returns, result.net_values, rf_daily=rf_daily_hk, annualization=annualization_hk)
comparison_table_hk.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_hk, result.weights).mean()),
}
)
comparison_hk = pd.DataFrame(comparison_table_hk)
comparison_hk["sharpe"] = comparison_hk["Sharpe"]
comparison_hk["max_drawdown"] = comparison_hk["Max Drawdown"]
def parse_hk_choice(name):
direction, dependence, network_type, measure = name.split("|")
return direction, dependence, network_type, measure
def rolling_hk_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_hk,
annualization=annualization_hk,
)
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_hk_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_hk,
annualization=annualization_hk,
)
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_hk = model_dates_hk[min(8, len(model_dates_hk) - 1)]
central_fixed_hk = "central|copula|pmfg|combined"
peripheral_train_hk = train_hk_choice(comparison_results_hk, "peripheral", selector_train_end_hk)
def stability_gated_hk_choice(results, dt, direction, train_selection, threshold=0.50):
if pd.Timestamp(dt) <= selector_train_end_hk:
return f"{direction}|copula|pmfg|combined", {"Sharpe": np.nan, "source": "fallback"}
rolling_name, rolling_info = rolling_hk_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_hk = []
central_rows_hk, peripheral_rows_hk, diversifier_rows_hk = {}, {}, {}
for dt in model_dates_hk:
central_choice = central_fixed_hk
central_info = {"Sharpe": np.nan, "source": "fixed"}
peripheral_choice, peripheral_info = stability_gated_hk_choice(
comparison_results_hk, dt, "peripheral", peripheral_train_hk
)
network_choice, network_info = stability_gated_hk_choice(
comparison_results_hk, dt, "peripheral", peripheral_train_hk
)
selection_records_hk.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_hk[dt]
_, dependence, network_type, measure = parse_hk_choice(central_choice)
centrality = record["centralities"][(dependence, network_type)]
central_score = scale_01(centrality[measure])
central_rows_hk[dt] = central_peripheral_weights(
central_score, returns=record["window"], side="central", n_stocks=20, max_weight=w_max_hk
)
_, dependence, network_type, measure = parse_hk_choice(peripheral_choice)
centrality = record["centralities"][(dependence, network_type)]
periphery = 1.0 - scale_01(centrality[measure])
peripheral_rows_hk[dt] = central_peripheral_weights(
periphery, returns=record["window"], side="peripheral", n_stocks=20, max_weight=w_max_hk
)
_, dependence, network_type, measure = parse_hk_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_hk[dt] = network_diversifier_weights(
scores, returns=record["window"], n_stocks=25, max_weight=w_max_hk
)
selection_hk = pd.DataFrame(selection_records_hk).set_index("date")
w_central_hk = pd.DataFrame.from_dict(central_rows_hk, orient="index").fillna(0.0)
w_peripheral_hk = pd.DataFrame.from_dict(peripheral_rows_hk, orient="index").fillna(0.0)
w_diversifier_hk = pd.DataFrame.from_dict(diversifier_rows_hk, orient="index").fillna(0.0)
common_index_hk = (
w_diversifier_hk.index.intersection(w_equal_hk.index).intersection(w_minvar_hk.index).intersection(w_hrp_hk.index)
)
columns_hk = sorted(
set(w_equal_hk.columns)
| set(w_minvar_hk.columns)
| set(w_hrp_hk.columns)
| set(w_central_hk.columns)
| set(w_peripheral_hk.columns)
| set(w_diversifier_hk.columns)
)
weights_hk = {
"Equal Weight": w_equal_hk.reindex(index=common_index_hk, columns=columns_hk).fillna(0.0),
"MinVar": w_minvar_hk.reindex(index=common_index_hk, columns=columns_hk).fillna(0.0),
"HRP": w_hrp_hk.reindex(index=common_index_hk, columns=columns_hk).fillna(0.0),
"Most Central 20": w_central_hk.reindex(index=common_index_hk, columns=columns_hk).fillna(0.0),
"Most Peripheral 20": w_peripheral_hk.reindex(index=common_index_hk, columns=columns_hk).fillna(0.0),
"Network Diversifier": w_diversifier_hk.reindex(index=common_index_hk, columns=columns_hk).fillna(0.0),
}
results_hk = run_many_weights_backtests(
weights_hk,
returns=ret_hk.reindex(columns=columns_hk).fillna(0.0),
cost_bps=10,
w_min=0.0,
w_max=w_max_hk,
long_only=True,
weight_timing="next_close",
rf_daily=rf_daily_hk,
)
summary_hk = build_strategy_summary(results_hk, rf_daily=rf_daily_hk, annualization=annualization_hk)
display(summary_hk[["CAGR", "Vol", "Sharpe", "Max Drawdown", "Calmar", "Turnover", "Cost Drag", "Effective N"]].round(4))
display(selection_hk.tail(8))
display(
pd.concat(
{
"Most Central 20": selection_hk["central_choice"].value_counts(),
"Most Peripheral 20": selection_hk["peripheral_choice"].value_counts(),
"Network Diversifier": selection_hk["network_choice"].value_counts(),
},
axis=1,
).fillna(0).astype(int)
)
latest_dt_hk = common_index_hk[-1]
latest_hk = store_hk[latest_dt_hk]
latest_network_choice_hk = selection_hk.loc[latest_dt_hk, "network_choice"]
_, latest_dependence_hk, latest_network_type_hk, latest_measure_hk = parse_hk_choice(latest_network_choice_hk)
centrality_hk = latest_hk["centralities"][(latest_dependence_hk, latest_network_type_hk)]
central_score_hk = scale_01(centrality_hk[latest_measure_hk])
periphery_hk = 1.0 - central_score_hk
network_scores_hk = network_score(periphery_hk, returns=latest_hk["window"])
score_frame_hk = pd.concat(
{
"centrality": central_score_hk,
"periphery": periphery_hk,
"network score": network_scores_hk,
"weight": weights_hk["Network Diversifier"].loc[latest_dt_hk],
},
axis=1,
).sort_values("network score", ascending=False).head(25)
weight_frame_hk = pd.concat(
{
"Most Central 20": weights_hk["Most Central 20"].loc[latest_dt_hk],
"Most Peripheral 20": weights_hk["Most Peripheral 20"].loc[latest_dt_hk],
"Network Diversifier": weights_hk["Network Diversifier"].loc[latest_dt_hk],
},
axis=1,
).fillna(0.0)
nav_hk = pd.concat({name: result.net_values for name, result in results_hk.items()}, axis=1)
drawdowns_hk = nav_hk / nav_hk.cummax() - 1.0
fig, axes = plt.subplots(2, 4, figsize=(22, 9))
axes = axes.ravel()
network_graph(
axes[0],
latest_hk["graphs"][("correlation", "pmfg")],
title="HKEX Latest Correlation PMFG",
)
network_graph(
axes[1],
latest_hk["graphs"][("copula", "pmfg")],
title="HKEX Latest Copula PMFG",
)
score_heatmap(axes[2], score_frame_hk, title="HKEX Diversifier Scores")
weight_heatmap(axes[3], weight_frame_hk, title="HKEX Latest Network Weights")
nav_hk.plot(ax=axes[4], title="HKEX Costed NAV")
drawdowns_hk.plot(ax=axes[5], title="HKEX Drawdowns")
performance_heatmap(axes[6], comparison_hk, value="sharpe", title="HKEX Centrality Sharpe")
performance_heatmap(axes[7], comparison_hk, 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_hk = risk_report(
objects={name: result.net_returns for name, result in results_hk.items()},
rf_daily=rf_daily_hk,
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")