1#!/usr/bin/env python32"""Stage 1 of the evaluation: forecasts -> per-origin losses for every (spec, h, proxy, loss), one Parquet per instrument.3usage: compute_losses.py --run main [--symbols SPY,...] [--workers 8]4Reads results/forecasts/<run>/<symbol>/*.parquet (not .fits), panels from the cache; writes results/metrics/losses_<run>/<symbol>.parquet5and results/metrics/losses_<run>/fits_<symbol>.parquet (concatenated refit records)."""6from __future__ import annotations78import argparse9import glob10import sys11from concurrent.futures import ProcessPoolExecutor, as_completed12from pathlib import Path1314import pandas as pd15import yaml1617ROOT = Path(__file__).resolve().parents[1]18sys.path.insert(0, str(ROOT / "src"))19from wp46.preprocessing import load_panel # noqa: E40220from wp46.scoring.evaluate import pack_wide, score_forecasts_wide # noqa: E4022122PROT = yaml.safe_load(open(ROOT / "configs" / "forecast_protocol.yaml"))232425def one(run: str, sym: str) -> dict:26 d = ROOT / "results" / "forecasts" / run / sym27 files = [f for f in sorted(glob.glob(str(d / "*.parquet"))) if not f.endswith(".fits.parquet")]28 if not files:29 return {"symbol": sym, "n_specs": 0}30 F = pd.concat([pd.read_parquet(f) for f in files], ignore_index=True)31 gf = [pd.read_parquet(f.replace(".parquet", ".fits.parquet")) for f in files if Path(f.replace(".parquet", ".fits.parquet")).exists()]32 G = pd.concat(gf, ignore_index=True) if gf else pd.DataFrame()33 panel = load_panel(sym)34 proxies = [PROT["primary_proxy"]] + list(PROT["secondary_proxies"])35 losses = [PROT["primary_loss"]] + list(PROT["secondary_losses"])36 res = score_forecasts_wide(F, panel, proxies=proxies, losses=losses, horizons=tuple(PROT["horizons"]), window=PROT["estimation"]["window"],37 primary_proxy=PROT["primary_proxy"], primary_loss=PROT["primary_loss"])38 W, T, S = pack_wide(res, sym)39 out = ROOT / "results" / "metrics" / f"losses_{run}"40 out.mkdir(parents=True, exist_ok=True)41 W.to_parquet(out / f"{sym}.parquet", index=False)42 T.to_parquet(out / f"targets_{sym}.parquet", index=False)43 S.to_parquet(out / f"status_{sym}.parquet", index=False)44 G.to_parquet(out / f"fits_{sym}.parquet", index=False)45 return {"symbol": sym, "n_specs": F["spec_id"].nunique(), "n_rows": int(len(W)), "n_cols": int(W.shape[1])}464748def main() -> None:49 ap = argparse.ArgumentParser()50 ap.add_argument("--run", default="main")51 ap.add_argument("--symbols", default="")52 ap.add_argument("--workers", type=int, default=6)53 a = ap.parse_args()54 syms = a.symbols.split(",") if a.symbols else sorted(p.name for p in (ROOT / "results" / "forecasts" / a.run).iterdir() if p.is_dir())55 with ProcessPoolExecutor(a.workers) as ex:56 futs = {ex.submit(one, a.run, s): s for s in syms}57 for f in as_completed(futs):58 print(f.result(), flush=True)596061if __name__ == "__main__":62 main()63