#!/usr/bin/env python3 """Build every machine-generated exhibit of the paper from results/: numbers.tex (macros), tables/*.tex, figures/*.pdf. usage: build_paper.py --run main --period test (then: cd paper && make) Never edits prose. Every macro name is digit-free (LaTeX).""" from __future__ import annotations import argparse import glob import json import re import shutil import sys from pathlib import Path import numpy as np import pandas as pd import yaml ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) from wp46.reporting import figures as F # noqa: E402 from wp46.reporting import tables as TB # noqa: E402 REG = yaml.safe_load(open(ROOT / "configs" / "model_registry.yaml"))["models"] PROT = yaml.safe_load(open(ROOT / "configs" / "forecast_protocol.yaml")) WORDS = {0: "Zero", 1: "One", 2: "Two", 3: "Three", 4: "Four", 5: "Five", 6: "Six", 7: "Seven", 8: "Eight", 9: "Nine"} def macro_name(s: str) -> str: s = re.sub(r"[^A-Za-z0-9]", " ", s).title().replace(" ", "") return "".join(WORDS[int(c)] if c.isdigit() else c for c in s) def fnum(x, nd=3): if x is None or (isinstance(x, float) and not np.isfinite(x)): return "--" if isinstance(x, (int, np.integer)): return f"{x:,}" return f"{x:,.{nd}f}" def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--run", default="main"); ap.add_argument("--period", default="test") a = ap.parse_args() adir = ROOT / "results" / "metrics" / f"analysis_{a.run}_{a.period}" per_symbol = {Path(f).stem: json.load(open(f)) for f in sorted(glob.glob(str(adir / "per_symbol" / "*.json")))} P = pd.read_csv(adir / "pooled.csv") specs_all = sorted({s for r in per_symbol.values() for s in r["horizons"]["1"]["specs"]}) labels = {s: F.spec_label(s) for s in specs_all} macros = {} # --- headline numbers macros["NInstruments"] = len(per_symbol) import subprocess try: col = subprocess.run([str(ROOT / ".venv" / "bin" / "python"), "-m", "pytest", "--collect-only", "-q", "tests"], capture_output=True, text=True, cwd=ROOT, timeout=300).stdout macros["NTests"] = sum(1 for l in col.splitlines() if "::" in l) except Exception: # noqa: BLE001 macros["NTests"] = "--" macros["NSpecs"] = len(specs_all) macros["NModels"] = len({s.split("__")[0] for s in specs_all}) macros["NOriginsTotal"] = sum(r["horizons"]["1"]["n_origins"] for r in per_symbol.values()) macros["NValidTargetsHOne"] = sum(r["horizons"]["1"]["n_valid_targets"] for r in per_symbol.values()) Pa = P[P.asset_class == "all"] for r in Pa.itertuples(): base = macro_name(f"{r.spec_id.split('__')[0]} {r.spec_id.split('__')[1]} h{r.h}") macros[f"Ratio{base}"] = fnum(r.mean_ratio) macros[f"CiLo{base}"] = fnum(1 + r.boot_ci_low, 3); macros[f"CiHi{base}"] = fnum(1 + r.boot_ci_high, 3) macros[f"BootP{base}"] = fnum(r.boot_p, 3); macros[f"Wins{base}"] = int(r.dm_wins_holm10); macros[f"Losses{base}"] = int(r.dm_losses_holm10) macros[f"McsIn{base}"] = int(r.mcs_inclusions) # best spec per horizon (pooled, Tier A / Tier B) for h in (1, 5, 22): for tier in ("A", "B"): sub = Pa[(Pa.h == h) & (Pa.tier == tier)].sort_values("mean_ratio") if len(sub): macros[f"BestSpecTier{tier}H{WORDS[h] if h < 10 else 'TwentyTwo'}"] = labels[sub.iloc[0].spec_id] macros[f"BestRatioTier{tier}H{WORDS[h] if h < 10 else 'TwentyTwo'}"] = fnum(sub.iloc[0].mean_ratio) # --- accounting table acc_rows = [] for s in specs_all: vals = [r["accounting"].get(s, {}) for r in per_symbol.values()] acc_rows.append(dict(spec_id=s, family=F.fam(s), failed=np.nanmean([v.get("failed", np.nan) for v in vals]), inadmissible=np.nanmean([v.get("inadmissible", np.nan) for v in vals]), coverage=np.nanmean([r["horizons"]["1"]["specs"].get(s, {}).get("coverage", np.nan) for r in per_symbol.values()]), fit_time_median=np.nanmedian([v.get("fit_time_median", np.nan) for v in vals]), fit_time_mean=np.nanmean([v.get("fit_time_mean", np.nan) for v in vals]), iter_mean=np.nanmean([v.get("iter_mean", np.nan) for v in vals]))) A = pd.DataFrame(acc_rows); A["failed_or_inadmissible"] = A["failed"].fillna(0) + A["inadmissible"].fillna(0) A.to_csv(ROOT / "results" / "tables" / f"accounting_{a.run}_{a.period}.csv", index=False) # --- tail risk table tr = [] for sym, r in per_symbol.items(): for s, v in r.get("tailrisk", {}).items(): for lvl, t in v.items(): if isinstance(t, dict) and "hit_rate" in t: tr.append(dict(symbol=sym, spec_id=s, family=F.fam(s), level=float(lvl), **t)) TR = pd.DataFrame(tr); TR.to_csv(ROOT / "results" / "tables" / f"tailrisk_{a.run}_{a.period}.csv", index=False) # representative specifications (Student-t where available) and family order, used by tables and figures rep0 = {} for sp in specs_all: mid, dist, _ = sp.split("__"); pref = {"t": 0, "normal": 1, "empirical": 2, "ged": 3, "skewt": 4}[dist] if mid not in rep0 or pref < rep0[mid][0]: rep0[mid] = (pref, sp) fam_order = ["baseline", "classical", "asymmetric", "power", "fractional", "component", "regime-switching", "score-driven", "realized"] # --- LaTeX tables tdir = ROOT / "paper" / "tables"; tdir.mkdir(exist_ok=True) tierA = [s for s in specs_all if F.fam(s) != "realized" and (s.split("__")[1] in ("normal", "t", "empirical"))] tierB = [s for s in specs_all if F.fam(s) == "realized"] TB.pooled_table(P, tierA, labels, tdir / "tab_pooled_tierA.tex") repA_main = sorted([v[1] for v in rep0.values() if F.fam(v[1]) != "realized" and v[1].split("__")[0] not in F.EXT_PARENT], key=lambda s: (fam_order.index(F.fam(s)), s)) TB.pooled_table(P, repA_main, labels, tdir / "tab_pooled_tierA_rep.tex") rep_all = sorted([v[1] for v in rep0.values()], key=lambda s: (fam_order.index(F.fam(s)), s)) TB.robust_table(P, rep_all, labels, tdir / "tab_pooled_robust.tex") TB.secondary_table(per_symbol, rep_all, labels, ["rv5ss_on|qlike|1", "rv5_on|qlike|1", "rv1_on|qlike|1", "bv5_on|qlike|1", "medrv_on|qlike|1", "rv5ss_scaled|qlike|1", "r2|qlike|1", "rv5ss_on|mse|1", "rv5ss_on|mae|1", "rv5ss_on|mse_sd|1"], tdir / "tab_secondary.tex") ext_specs = sorted([sp for sp in specs_all if sp.split("__")[0] in F.EXT_PARENT]) if ext_specs: TB.pooled_table(P, ext_specs, labels, tdir / "tab_pooled_extensions.tex") TB.pooled_table(P, tierB, labels, tdir / "tab_pooled_tierB.tex") TB.accounting_table(A, labels, tdir / "tab_accounting.tex") tc = ROOT / "results" / "metrics" / "analysis_tier_c_test" if (tc / "tier_c_summary.csv").exists(): TC = pd.read_csv(tc / "tier_c_summary.csv"); TCX = json.load(open(tc / "tier_c_extra.json")) TB.tier_c_table(TC, TCX, tdir / "tab_tier_c.tex") for b in TC.basket.unique(): for m in ("dcc11", "adcc11", "scalar_bekk11", "ewma_cov094", "rolling_cov252"): q = TC[(TC.basket == b) & (TC.model == m) & (TC.loss == "mv_qlike")] if len(q): macros[f"TierC{macro_name(b)}{macro_name(m)}Ratio"] = fnum(float(q.iloc[0].ratio)); macros[f"TierC{macro_name(b)}{macro_name(m)}DmP"] = fnum(float(q.iloc[0].dm_p)) cc = ROOT / "results" / "metrics" / "rugarch_crosscheck.json" if cc.exists(): TB.crosscheck_table(json.load(open(cc)), tdir / "tab_crosscheck.tex") # --- figures # representative specifications (Student-t where available) ordered by family: readable heatmaps (mandate §5-6) rep0 = {} for sp in specs_all: mid, dist, _ = sp.split("__"); pref = {"t": 0, "normal": 1, "empirical": 2, "ged": 3, "skewt": 4}[dist] if mid not in rep0 or pref < rep0[mid][0]: rep0[mid] = (pref, sp) fam_order = ["baseline", "classical", "asymmetric", "power", "fractional", "component", "regime-switching", "score-driven", "realized"] repA = sorted([v[1] for v in rep0.values() if F.fam(v[1]) != "realized" and v[1].split("__")[0] not in F.EXT_PARENT], key=lambda s: (fam_order.index(F.fam(s)), s)) repB = sorted([v[1] for v in rep0.values() if F.fam(v[1]) == "realized" and v[1].split("__")[0] not in F.EXT_PARENT], key=lambda s: s) F.reset_audit() # one audit row per figure, rewritten at every build F.ratio_heatmap(per_symbol, 1, repA, F.FIG / "fig_heatmap_h1_tierA.pdf", "Tier A (representative specifications), h = 1") F.ratio_heatmap(per_symbol, 22, repA, F.FIG / "fig_heatmap_h22_tierA.pdf", "Tier A (representative specifications), h = 22") F.ratio_heatmap(per_symbol, 1, repB + [s for s in specs_all if s.startswith("garch11__normal")], F.FIG / "fig_heatmap_h1_tierB.pdf", "Tier B (+ GARCH(1,1)-N reference), h = 1") F.ratio_heatmap(per_symbol, 1, [s for s in tierA if s.split("__")[0] not in F.EXT_PARENT], F.FIG / "fig_heatmap_h1_tierA_full.pdf", "Tier A, all innovation laws (pre-registered architectures), h = 1", vmin=0.7, vmax=1.3) F.mcs_matrix(per_symbol, 1, "A", F.FIG / "fig_mcs_h1_tierA.pdf", "90% MCS, Tier A, all specifications (pre-registered), h = 1") F.mcs_matrix(per_symbol, 22, "A", F.FIG / "fig_mcs_h22_tierA.pdf", "90% MCS, Tier A, all specifications (pre-registered), h = 22") F.mcs_matrix(per_symbol, 1, "B", F.FIG / "fig_mcs_h1_tierB.pdf", "90% MCS, Tier B (+ GARCH(1,1)-N), h = 1") F.mcs_matrix(per_symbol, 1, "A", F.FIG / "fig_mcsrep_h1_tierA.pdf", "90% MCS, Tier A, one specification per architecture (exploratory), h = 1", key="mcs_rep") F.mcs_matrix(per_symbol, 22, "A", F.FIG / "fig_mcsrep_h22_tierA.pdf", "90% MCS, Tier A, one specification per architecture (exploratory), h = 22", key="mcs_rep") F.mcs_matrix(per_symbol, 1, "B", F.FIG / "fig_mcsrep_h1_tierB.pdf", "90% MCS, Tier B, one specification per architecture (exploratory), h = 1", key="mcs_rep") topA = Pa[(Pa.h == 1) & (Pa.tier == "A") & (~Pa.spec_id.str.split("__").str[0].isin(list(F.EXT_PARENT)))].sort_values("mean_ratio").spec_id.head(3).tolist() topB = Pa[(Pa.h == 1) & (Pa.tier == "B") & (~Pa.spec_id.str.split("__").str[0].isin(list(F.EXT_PARENT)))].sort_values("mean_ratio").spec_id.head(3).tolist() F.horizon_profile(P, topA + topB + [s for s in specs_all if s.startswith("garch11__t__")], F.FIG / "fig_horizon.pdf") F.pareto(Pa[Pa.h == 1], A, F.FIG / "fig_pareto_time.pdf") F.pareto(Pa[Pa.h == 1], A, F.FIG / "fig_pareto_fail.pdf", xcol="failed_or_inadmissible", xlabel="Share of rolling refits failed or inadmissible (fraction, all instruments)") F.coverage_map(); F.volatility_series(["SPY", "CL", "EURUSD", "BTC"], Path.home() / "wp46-cache" / "panels", F.FIG / "fig_volseries.pdf") F.lineage() # representative specs (Student-t where available) for heterogeneity exhibits rep = {} for sp in specs_all: mid, dist, _ = sp.split("__"); pref = {"t": 0, "normal": 1, "empirical": 2, "ged": 3, "skewt": 4}[dist] if mid not in rep or pref < rep[mid][0]: rep[mid] = (pref, sp) rep_specs = sorted([v[1] for v in rep.values()], key=lambda s: (F.fam(s), s)) rep_main = [s for s in rep_specs if s.split("__")[0] not in F.EXT_PARENT] # pre-registered architectures only (extensions: fig_extensions_*) F.class_family_panels(P, rep_main, 1, F.FIG / "fig_class_h1.pdf"); F.class_family_panels(P, rep_main, 22, F.FIG / "fig_class_h22.pdf") F.failure_diagnostics(A, Pa[Pa.h == 1], rep_specs, F.FIG / "fig_failures.pdf") if len(TR): F.var_calibration(TR, F.FIG / "fig_var.pdf", rep_main + [s for s in specs_all if s.startswith("garch11__normal__")]) shutil.copyfile(F.FIG / "fig_var.pdf", F.FIG / "fig_var_1pct.pdf") # legacy name still included by sections/results_tail_body.tex (same two-panel figure) reg_specs = [s for s in rep_specs if s.split("__")[0] in ("gjr11", "egarch11", "figarch11", "cgarch11", "msgarch2", "har_rv", "realgarch11", "heavy11")] F.regime_profile(per_symbol, reg_specs, F.FIG / "fig_regimes_h1.pdf", 1) TB.regime_table(per_symbol, rep_specs, labels, ["vol_tercile=low", "vol_tercile=mid", "vol_tercile=high", "prev_sign=neg", "prev_sign=pos", "episode=covid", "episode=rate_shock_2022", "episode=tariff_shock_2025"], tdir / "tab_regimes_h1.tex", 1) TB.regime_table(per_symbol, rep_specs, labels, ["vol_tercile=low", "vol_tercile=mid", "vol_tercile=high", "prev_sign=neg", "prev_sign=pos", "episode=covid", "episode=rate_shock_2022", "episode=tariff_shock_2025"], tdir / "tab_regimes_h22.tex", 22) if len(TR): TB.tailrisk_table(TR, [s for s in rep_specs] + [s for s in specs_all if s.startswith("garch11__") and s not in rep_specs], labels, tdir / "tab_tailrisk.tex") if any(sp.split("__")[0] in F.EXT_PARENT for sp in specs_all): F.extension_heatmap(per_symbol, 1, F.FIG / "fig_extensions_h1.pdf"); F.extension_heatmap(per_symbol, 22, F.FIG / "fig_extensions_h22.pdf") # pooled by asset class for the representative specs P[P.spec_id.isin(rep_specs)].to_csv(ROOT / "results" / "tables" / f"pooled_by_class_{a.run}_{a.period}.csv", index=False) # --- numbers.tex lines = ["% generated by scripts/build_paper.py — do not edit"] for k, v in macros.items(): lines.append(f"\\newcommand{{\\{k}}}{{{v}}}") (ROOT / "paper" / "numbers.tex").write_text("\n".join(lines) + "\n") print("macros", len(macros), "tables", len(list(tdir.glob("*.tex"))), "figures", len(list(F.FIG.glob("*.pdf")))) if __name__ == "__main__": main()