1#!/usr/bin/env python32"""Phase 2 pilot: GARCH(1,1), EGARCH, GJR, APARCH (normal, t) + HAR-RV on a few instruments over the VALIDATION3period only (2012-01-03 .. 2015-12-31), window 1000, refit every 5 days. Writes results/metrics/pilot/*.4Reports: status counts, fit/forecast timings, QLIKE on the common sample, leakage check (future returns perturbed)."""5from __future__ import annotations67import argparse8import json9import sys10import time11from pathlib import Path1213import numpy as np14import pandas as pd1516ROOT = Path(__file__).resolve().parents[1]17sys.path.insert(0, str(ROOT / "src"))18from wp46.estimation import JobSpec, run_job # noqa: E40219from wp46.inference import dm_test # noqa: E40220from wp46.preprocessing import load_panel # noqa: E40221from wp46.scoring.evaluate import common_sample_mask, score_forecasts # noqa: E4022223SPECS = [("garch11", "normal", "A"), ("garch11", "t", "A"), ("egarch11", "t", "A"), ("gjr11", "t", "A"), ("aparch11", "t", "A"),24 ("nagarch11", "t", "A"), ("histvar22", "empirical", "A"), ("har_rv", "empirical", "B"), ("realgarch11", "normal", "B")]252627def main() -> None:28 ap = argparse.ArgumentParser()29 ap.add_argument("--symbols", default="SPY,AAPL,CL,EURUSD")30 ap.add_argument("--start", default="2012-01-03")31 ap.add_argument("--end", default="2015-12-31")32 a = ap.parse_args()33 out = ROOT / "results" / "metrics" / "pilot"34 out.mkdir(parents=True, exist_ok=True)35 report = {}36 for sym in a.symbols.split(","):37 panel = load_panel(sym)38 Fs, Gs = [], []39 for mid, dist, tier in SPECS:40 t0 = time.perf_counter()41 F, G = run_job(panel, JobSpec(sym, mid, dist, tier, oos_start=a.start, oos_end=a.end))42 Fs.append(F); Gs.append(G)43 print(f"{sym:7s} {mid:12s} {dist:9s} origins={len(F)} {F['status'].value_counts().to_dict()} fits={G['status'].value_counts().to_dict()} "44 f"fit_s={G['time_s'].mean():.3f} fc_ms={1000*F['time_fc_s'].mean():.1f} wall={time.perf_counter()-t0:.1f}s", flush=True)45 F = pd.concat(Fs, ignore_index=True); G = pd.concat(Gs, ignore_index=True)46 L = score_forecasts(F, panel, proxies=("rv5ss_on", "r2"), losses=("qlike", "mse"))47 specs = sorted(F["spec_id"].unique())48 cm = common_sample_mask(L, specs)49 Lc = L[cm]50 tab = Lc.groupby(["proxy", "loss", "h", "spec_id"])["value"].mean().unstack("spec_id")51 bench = [s for s in specs if s.startswith("garch11__normal")][0]52 ratio = tab.div(tab[bench], axis=0)53 # leakage check: perturb returns AND realized measures after the last origin -> every forecast of the whole54 # job (same refit schedule) is unchanged55 p2 = panel.copy(); last = F["origin"].max(); m = p2["session"] > last56 p2.loc[m, "r"] *= 3.0; p2.loc[m, "rv_5m_ss"] *= 9.057 leak_ok = True58 for mid, dist, tier in [("gjr11", "t", "A"), ("har_rv", "empirical", "B")]:59 F1 = F[F.spec_id.str.startswith(f"{mid}__{dist}")].sort_values("origin")60 F2, _ = run_job(p2, JobSpec(sym, mid, dist, tier, oos_start=a.start, oos_end=a.end))61 leak_ok &= bool(np.allclose(F1["h1"].values, F2.sort_values("origin")["h1"].values, equal_nan=True))62 dm = {s: dm_test(Lc[(Lc.spec_id == s) & (Lc.h == 1) & (Lc.proxy == "rv5ss_on") & (Lc.loss == "qlike")].sort_values("origin")["value"].values,63 Lc[(Lc.spec_id == bench) & (Lc.h == 1) & (Lc.proxy == "rv5ss_on") & (Lc.loss == "qlike")].sort_values("origin")["value"].values, 1)64 for s in specs if s != bench}65 report[sym] = {"n_origins": int(F.groupby("spec_id").size().max()), "common_sample_n_h1": int(cm[(L.h == 1) & (L.proxy == "rv5ss_on") & (L.loss == "qlike")].sum()),66 "status": F.groupby(["spec_id", "status"]).size().unstack(fill_value=0).to_dict(), "fit_status": G.groupby(["spec_id", "status"]).size().unstack(fill_value=0).to_dict(),67 "mean_loss_common": tab.round(4).reset_index().to_dict(orient="records"), "ratio_to_garch11_normal": ratio.round(4).reset_index().to_dict(orient="records"),68 "dm_vs_garch11_normal_h1_qlike": {k: {kk: (round(vv, 4) if isinstance(vv, float) else vv) for kk, vv in v.items()} for k, v in dm.items()},69 "leakage_check_passed": leak_ok, "timing": G.groupby("spec_id")["time_s"].agg(["mean", "median", "max"]).round(4).to_dict()}70 L.to_parquet(out / f"losses_{sym}.parquet", index=False)71 F.to_parquet(out / f"forecasts_{sym}.parquet", index=False)72 print(sym, "QLIKE ratio to GARCH(1,1)-N, proxy rv5ss_on:"); print(ratio.loc[("rv5ss_on", "qlike")].round(3).to_string())73 print("leakage check:", leak_ok)74 (out / "pilot_report.json").write_text(json.dumps(report, indent=1, default=str))757677if __name__ == "__main__":78 main()79