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 : 05_robustness.py9Description : Étape 05 — Robustesse : sous-périodes, sensibilité10 à la fréquence d'échantillonnage et à la fenêtre11 d'estimation, transférabilité cross-asset (pooled12 leave-one-out) et fluctuation test Giacomini-Rossi.13Usage : python scripts/05_robustness.py [--part all|subperiods|freq|window|loo|gr]14================================================================15"""1617from __future__ import annotations1819import argparse20import logging21import sys22from concurrent.futures import ProcessPoolExecutor, as_completed23from pathlib import Path2425import numpy as np26import pandas as pd2728sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))2930from wp12 import config, evaluation as ev, robustness as rb # noqa: E40231from wp12.models_econ import horizon_target # noqa: E4023233logger = logging.getLogger("05_robustness")3435OUT = config.path("reproduced")36HORIZONS = tuple(config.load_config()["evaluation"]["horizons"])37WINDOW = int(config.load_config()["evaluation"]["estimation_window_days"])38REFIT = int(config.load_config()["evaluation"]["reestimation_freq_days"])3940# representative asset per class for the expensive experiments41REPRESENTATIVE = {"equity": "NVDA", "fx": "GBPUSD", "crypto": "ETH", "futures": "GC"}42# models re-run in the sensitivity experiments43SENS_MODELS = ["HAR", "LogHAR", "LightGBM"]444546def _merged() -> pd.DataFrame:47 df = pd.read_parquet(config.path("processed") / "merged_forecasts.parquet")48 df["date"] = pd.to_datetime(df["date"])49 return df505152def part_subperiods() -> None:53 """Mean QLIKE per model within each sub-period of the config."""54 sp = {k: tuple(v) for k, v in config.load_config()["robustness"]["subperiods"].items()}55 merged = _merged()56 tab = rb.loss_by_period(merged, sp)57 tab.to_csv(OUT / "subperiod_losses.csv", index=False)58 logger.info("subperiod table written (%d rows)", len(tab))596061def _sens_one(job: tuple[str, str, str, int]) -> pd.DataFrame:62 """Worker: re-run one (ticker, model, measure, window) variant."""63 from wp12 import models_econ as me, models_ml as ml6465 ticker, model, measure, window = job66 df = pd.read_parquet(config.path("processed") / f"rv_{ticker}.parquet")67 if model in ("HAR", "LogHAR"):68 fc = me.rolling_har_forecasts(69 df, model, horizons=HORIZONS, measure=measure,70 window=window, refit_every=REFIT)71 else: # LightGBM extended72 panel = pd.read_parquet(config.path("processed") / "panel.parquet")73 frames = ml.ml_features(panel, measure=measure)74 fc = ml.rolling_ml_forecasts(75 frames[ticker], "LightGBM", feature_set="extended",76 horizons=HORIZONS, window=window, refit_every=REFIT)77 fc["model"] = "LightGBM"78 fc["ticker"] = ticker79 # attach actuals in the SAME measure80 rv = df[measure]81 parts = []82 for h in HORIZONS:83 t = horizon_target(rv, h).rename("actual").reset_index()84 t["h"] = h85 parts.append(t)86 tgt = pd.concat(parts, ignore_index=True)87 fc["date"] = pd.to_datetime(fc["date"])88 tgt["date"] = pd.to_datetime(tgt["date"])89 out = fc.merge(tgt, on=["date", "h"]).dropna(subset=["actual", "forecast"])90 return out[out["actual"] > 0]919293def _run_jobs(jobs: list[tuple], workers: int) -> list[pd.DataFrame]:94 res = []95 with ProcessPoolExecutor(max_workers=workers) as pool:96 futs = {pool.submit(_sens_one, j): j for j in jobs}97 for fut in as_completed(futs):98 j = futs[fut]99 try:100 res.append(fut.result())101 logger.info("done %s", j)102 except Exception as exc: # noqa: BLE001103 logger.error("FAILED %s: %s", j, exc)104 return res105106107def part_frequency(workers: int) -> None:108 """Sensitivity to the RV sampling scheme (1-min, 5-min ss, kernel)."""109 tickers = [i["ticker"] for i in config.universe()]110 freqs = {"rv1min": "rv1", "rv5min_ss": "rv5ss", "rkernel": "rk"}111 results = {}112 for label, measure in freqs.items():113 jobs = [(tk, m, measure, WINDOW) for tk in tickers for m in SENS_MODELS]114 frames = _run_jobs(jobs, workers)115 results[label] = pd.concat(frames, ignore_index=True)116 tab = rb.sensitivity_table(results, "frequency")117 tab.to_csv(OUT / "frequency_sensitivity.csv", index=False)118 logger.info("frequency sensitivity written")119120121def part_window(workers: int) -> None:122 """Sensitivity to the estimation window (500 / 1000 / 2000 days)."""123 tickers = [i["ticker"] for i in config.universe()]124 results = {}125 for w in config.load_config()["robustness"]["window_sizes"]:126 jobs = [(tk, m, "rv5ss", int(w)) for tk in tickers for m in SENS_MODELS]127 frames = _run_jobs(jobs, workers)128 results[str(w)] = pd.concat(frames, ignore_index=True)129 tab = rb.sensitivity_table(results, "window")130 tab.to_csv(OUT / "window_sensitivity.csv", index=False)131 logger.info("window sensitivity written")132133134def part_loo() -> None:135 """Cross-asset transferability: pooled model with one class representative136 excluded from training, predicted out-of-pool."""137 from wp12 import models_ml as ml138139 panel = pd.read_parquet(config.path("processed") / "panel.parquet")140 frames = ml.ml_features(panel)141 parts = []142 for cls, tk in REPRESENTATIVE.items():143 fc = ml.rolling_pooled_lgbm(144 frames, horizons=HORIZONS, window=WINDOW, refit_every=REFIT, exclude=tk)145 fc["cls"] = cls146 parts.append(fc)147 loo = pd.concat(parts, ignore_index=True)148 loo.to_parquet(config.path("processed") / "forecasts" / "pooled_loo.parquet",149 index=False)150 # compare in-pool vs out-of-pool QLIKE on the same dates151 merged = _merged()152 rows = []153 for cls, tk in REPRESENTATIVE.items():154 sub_loo = loo[loo["ticker"] == tk].copy()155 sub_loo["date"] = pd.to_datetime(sub_loo["date"])156 base = merged[(merged["ticker"] == tk)157 & (merged["model"].isin(["Pooled-LGBM", "HAR", "LightGBM-X"]))]158 act = base.drop_duplicates(["date", "h"])[["date", "h", "actual"]]159 sub = sub_loo.merge(act, on=["date", "h"]).dropna(subset=["actual"])160 for h in HORIZONS:161 d = sub[sub["h"] == h]162 q_loo = float(np.mean(ev.qlike(d["actual"].to_numpy(), d["forecast"].to_numpy())))163 for m in ["Pooled-LGBM", "HAR", "LightGBM-X"]:164 b = base[(base["h"] == h) & (base["model"] == m)]165 b = b.merge(d[["date"]], on="date")166 q = float(np.mean(ev.qlike(b["actual"].to_numpy(), b["forecast"].to_numpy())))167 rows.append((cls, tk, h, m, q))168 rows.append((cls, tk, h, "Pooled-LGBM-LOO", q_loo))169 tab = pd.DataFrame(rows, columns=["cls", "ticker", "h", "model", "qlike"])170 tab.to_csv(OUT / "transferability.csv", index=False)171 logger.info("transferability written")172173174def part_gr() -> None:175 """Giacomini-Rossi fluctuation paths: best ML vs HAR, h=1, per class rep."""176 merged = _merged()177 paths = []178 for cls, tk in REPRESENTATIVE.items():179 sub = merged[(merged["ticker"] == tk) & (merged["h"] == 1)]180 wide = sub.pivot_table(index="date", columns="model", values="forecast")181 actual = sub.drop_duplicates("date").set_index("date")["actual"]182 for m in ["LightGBM-X", "Pooled-LGBM"]:183 if m not in wide.columns:184 continue185 both = wide[[m, "HAR"]].dropna()186 la = pd.Series(ev.qlike(actual[both.index].to_numpy(), both[m].to_numpy()),187 index=both.index)188 lb = pd.Series(ev.qlike(actual[both.index].to_numpy(), both["HAR"].to_numpy()),189 index=both.index)190 path = rb.giacomini_rossi_fluctuation(la, lb, h=1, mu=0.3)191 path["cls"] = cls192 path["ticker"] = tk193 path["model"] = m194 paths.append(path)195 pd.concat(paths, ignore_index=True).to_csv(OUT / "gr_fluctuation.csv", index=False)196 logger.info("GR fluctuation paths written")197198199def main() -> None:200 """Run the requested robustness parts."""201 ap = argparse.ArgumentParser()202 ap.add_argument("--part", default="all",203 choices=["all", "subperiods", "freq", "window", "loo", "gr"])204 ap.add_argument("--workers", type=int, default=8)205 args = ap.parse_args()206 config.setup_logging()207 if args.part in ("all", "subperiods"):208 part_subperiods()209 if args.part in ("all", "freq"):210 part_frequency(args.workers)211 if args.part in ("all", "window"):212 part_window(args.workers)213 if args.part in ("all", "loo"):214 part_loo()215 if args.part in ("all", "gr"):216 part_gr()217218219if __name__ == "__main__":220 main()221