1#!/usr/bin/env python32"""Tier C summary: mean losses per basket/model on the common sample (test period), DM vs CCC, MCS over models."""3from __future__ import annotations45import json6import sys7from pathlib import Path89import numpy as np10import pandas as pd11import yaml1213ROOT = Path(__file__).resolve().parents[1]14sys.path.insert(0, str(ROOT / "src"))15from wp46.inference import dm_test, mcs # noqa: E4021617PROT = yaml.safe_load(open(ROOT / "configs" / "forecast_protocol.yaml"))181920def main(period: str = "test") -> None:21 d = ROOT / "results" / "metrics" / "tier_c"22 per = PROT["periods"][period]23 out = {}24 rows = []25 STUDY = yaml.safe_load(open(ROOT / "configs" / "study.yaml"))26 for b in STUDY["tier_c_baskets"]:27 frames = {p.stem[len(b) + 1:]: pd.read_parquet(p) for p in d.glob(f"{b}_*.parquet")}28 if not frames:29 continue30 mats = {}31 for loss in ("mv_qlike", "frobenius", "pf_qlike"):32 M = pd.DataFrame({m: f.set_index(pd.to_datetime(f["origin"]))[loss] for m, f in frames.items() if loss in f})33 M = M[(M.index >= per["start"]) & (M.index <= per["end"])].dropna()34 mats[loss] = M35 bench = "ccc_garch11" if "ccc_garch11" in M else M.columns[0]36 for m in M.columns:37 dm = dm_test(M[m].values, M[bench].values, 1)38 rows.append(dict(basket=b, loss=loss, model=m, n=len(M), mean=float(M[m].mean()), ratio=float(M[m].mean() / M[bench].mean()), dm_stat=dm["stat"], dm_p=dm["p"]))39 try:40 mc = mcs(M, size=0.10, reps=2000, block=22)41 for r in mc.itertuples():42 rows[-len(M.columns):]43 out[f"{b}|{loss}|mcs"] = {r.model: {"p": float(r.pvalue), "in": bool(r.included)} for r in mc.itertuples()}44 except Exception as e: # noqa: BLE00145 out[f"{b}|{loss}|mcs"] = {"error": repr(e)[:200]}46 status = {m: f["status"].value_counts().to_dict() for m, f in frames.items()}47 mineig = {m: float(f["min_eig"].min()) for m, f in frames.items() if "min_eig" in f}48 out[b] = {"status": status, "min_eig": mineig, "fit_time_median": {m: float(f.loc[f["fit_time_s"] > 0, "fit_time_s"].median()) for m, f in frames.items() if "fit_time_s" in f},49 "basket": STUDY["tier_c_baskets"][b]}50 T = pd.DataFrame(rows)51 o = ROOT / "results" / "metrics" / f"analysis_tier_c_{period}"; o.mkdir(parents=True, exist_ok=True)52 T.to_csv(o / "tier_c_summary.csv", index=False); (o / "tier_c_extra.json").write_text(json.dumps(out, indent=1))53 (ROOT / "results" / "tables").mkdir(exist_ok=True); T.to_csv(ROOT / "results" / "tables" / f"tier_c_{period}.csv", index=False)54 print(T[T.loss == "mv_qlike"].round(4).to_string(index=False))555657if __name__ == "__main__":58 main(sys.argv[1] if len(sys.argv) > 1 else "test")59