Python 75.9%
TeX 24%
1#!/usr/bin/env python32"""3================================================================4Auteur : Simon-Pierre Boucher5Contact : contact@spboucher.ai6Projet : Prévision de volatilité réalisée multi-actifs7 (HAR-RV vs GARCH vs Machine Learning)8Fichier : 06_economic_value.py9Description : Étape 06 — Valeur économique : volatility timing10 (Sharpe net, sensibilité aux coûts), gains11 d'utilité mean-variance vs HAR, backtests VaR12 (Kupiec, Christoffersen).13Usage : python scripts/06_economic_value.py14================================================================15"""1617from __future__ import annotations1819import logging20import sys21from pathlib import Path2223import numpy as np24import pandas as pd2526sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))2728from wp12 import backtest as bt, config # noqa: E4022930logger = logging.getLogger("06_economic_value")3132OUT = config.path("reproduced")33CFG = config.load_config()["backtest"]3435# models carried into the economic-value exercise (h=1 forecasts)36EV_MODELS = [37 "HAR", "LogHAR", "HARQ", "SHAR", "HAR-CJ",38 "GARCH", "GJR", "EGARCH", "RealGARCH",39 "Ridge-X", "LASSO-X", "RF-X", "XGBoost-X", "LightGBM-X",40 "LSTM", "Transformer", "Pooled-LGBM", "Comb-Mean", "Comb-InvMSE",41]424344def _load() -> tuple[pd.DataFrame, dict[str, pd.Series], dict[str, str]]:45 """Merged h=1 forecasts + daily returns per ticker + class map."""46 merged = pd.read_parquet(config.path("processed") / "merged_forecasts.parquet")47 merged["date"] = pd.to_datetime(merged["date"])48 merged = merged[(merged["h"] == 1) & merged["model"].isin(EV_MODELS)]49 panel = pd.read_parquet(config.path("processed") / "panel.parquet")50 rets = {tk: d.sort_index()["ret_cc"] for tk, d in panel.groupby("ticker")}51 cls = panel.groupby("ticker")["cls"].first().to_dict()52 return merged, rets, cls535455def timing_tables(merged: pd.DataFrame, rets: dict, cls: dict) -> None:56 """Volatility-timing Sharpe (net) per model, plus cost sensitivity."""57 rows = []58 for (tk, m), sub in merged.groupby(["ticker", "model"], observed=True):59 pv = sub.set_index("date")["forecast"]60 ret = rets[tk]61 for tc in CFG["tc_grid_bps"]:62 led = bt.volatility_timing(63 ret, pv, vol_target_ann=CFG["vol_target_ann"],64 leverage_cap=CFG["leverage_cap"], tc_bps=float(tc))65 st = bt.perf_stats(led["net"])66 rows.append((tk, cls[tk], m, tc, st["ann_ret"], st["ann_vol"],67 st["sharpe"], st["max_dd"],68 float(led["weight"].diff().abs().mean())))69 tab = pd.DataFrame(rows, columns=[70 "ticker", "cls", "model", "tc_bps", "ann_ret", "ann_vol",71 "sharpe", "max_dd", "turnover"])72 tab.to_csv(OUT / "timing_by_ticker.csv", index=False)73 base = tab[tab["tc_bps"] == CFG["tc_bps"]]74 summary = base.groupby(["cls", "model"], observed=True)["sharpe"].mean().reset_index()75 overall = base.groupby("model", observed=True)[["sharpe", "ann_ret", "ann_vol",76 "turnover"]].mean().reset_index()77 overall["cls"] = "all"78 summary.to_csv(OUT / "timing_by_class.csv", index=False)79 overall.to_csv(OUT / "timing_overall.csv", index=False)80 logger.info("timing tables written")818283def utility_tables(merged: pd.DataFrame, rets: dict, cls: dict) -> None:84 """Annualized utility gain (bps) of each model over HAR."""85 gamma = CFG["risk_aversion"]86 rows = []87 for tk, sub in merged.groupby("ticker", observed=True):88 ret = rets[tk]89 nets = {}90 for m, s in sub.groupby("model", observed=True):91 pv = s.set_index("date")["forecast"]92 led = bt.volatility_timing(93 ret, pv, vol_target_ann=CFG["vol_target_ann"],94 leverage_cap=CFG["leverage_cap"], tc_bps=float(CFG["tc_bps"]))95 nets[m] = led["net"]96 if "HAR" not in nets:97 continue98 for m, net in nets.items():99 if m == "HAR":100 continue101 gain = bt.utility_gain(net, nets["HAR"], risk_aversion=gamma)102 rows.append((tk, cls[tk], m, gain * 1e4))103 tab = pd.DataFrame(rows, columns=["ticker", "cls", "model", "utility_gain_bps"])104 tab.to_csv(OUT / "utility_by_ticker.csv", index=False)105 tab.groupby("model", observed=True)["utility_gain_bps"].agg(106 ["mean", "median"]).reset_index().to_csv(OUT / "utility_summary.csv", index=False)107 logger.info("utility tables written")108109110def var_tables(merged: pd.DataFrame, rets: dict, cls: dict) -> None:111 """VaR 1% / 5% backtests; RV forecasts scaled to return variance with a112 trailing 250-day factor (no look-ahead)."""113 rows = []114 for (tk, m), sub in merged.groupby(["ticker", "model"], observed=True):115 pv = sub.set_index("date")["forecast"]116 ret = rets[tk]117 # trailing conversion factor: return variance per unit of RV118 rv_al = pv.reindex(ret.index)119 c = ((ret**2).rolling(250).mean() / rv_al.rolling(250).mean()).clip(0.2, 5.0)120 pv_ret = (rv_al * c).dropna()121 for level in CFG["var_levels"]:122 res = bt.var_backtest(ret, pv_ret, float(level))123 rows.append((tk, cls[tk], m, level, res["n"], res["viol_rate"],124 res["kupiec_p"], res["christoffersen_p"]))125 tab = pd.DataFrame(rows, columns=[126 "ticker", "cls", "model", "level", "n", "viol_rate",127 "kupiec_p", "christoffersen_p"])128 tab.to_csv(OUT / "var_by_ticker.csv", index=False)129 summary = tab.groupby(["model", "level"], observed=True).apply(130 lambda d: pd.Series({131 "mean_viol_rate": d["viol_rate"].mean(),132 "pct_pass_kupiec": float((d["kupiec_p"] > 0.05).mean()),133 "pct_pass_cc": float((d["christoffersen_p"] > 0.05).mean()),134 }), include_groups=False).reset_index()135 summary.to_csv(OUT / "var_summary.csv", index=False)136 logger.info("VaR tables written")137138139def main() -> None:140 """Run all economic-value blocks."""141 config.setup_logging()142 merged, rets, cls = _load()143 timing_tables(merged, rets, cls)144 utility_tables(merged, rets, cls)145 var_tables(merged, rets, cls)146 logger.info("economic value complete")147148149if __name__ == "__main__":150 main()151