1#!/usr/bin/env python32"""Phase 2 — build the daily panels (returns + realized measures) for every ingested instrument.3Writes <cache>/panels/<symbol>.parquet and results/manifests/panels_manifest.json (row counts, date ranges, validity4rates, sha256; NO return values)."""5from __future__ import annotations67import argparse8import hashlib9import json10import sys11from concurrent.futures import ProcessPoolExecutor, as_completed12from pathlib import Path1314import yaml1516ROOT = Path(__file__).resolve().parents[1]17sys.path.insert(0, str(ROOT / "src"))18from wp46.preprocessing import build_daily_panel # noqa: E40219from wp46.preprocessing.panel import panel_path # noqa: E402202122def one(cls: str, sym: str) -> dict:23 df = build_daily_panel(cls, sym)24 p = panel_path(sym)25 p.parent.mkdir(parents=True, exist_ok=True)26 df.to_parquet(p, index=False)27 h = hashlib.sha256(p.read_bytes()).hexdigest()28 return {"asset_class": cls, "symbol": sym, "rows": int(len(df)), "first": str(df["session"].min().date()),29 "last": str(df["session"].max().date()), "valid_rm_rate": float(df["valid_rm"].mean()),30 "n_nan_r": int(df["r"].isna().sum()), "sha256": h, "path": str(p)}313233def main() -> None:34 ap = argparse.ArgumentParser()35 ap.add_argument("--config", default=str(ROOT / "configs" / "instruments.yaml"))36 ap.add_argument("--symbols", default="")37 ap.add_argument("--workers", type=int, default=6)38 a = ap.parse_args()39 cfg = yaml.safe_load(open(a.config))40 uni = cfg.get("universe") or cfg["candidates"]41 want = set(a.symbols.split(",")) if a.symbols else None42 jobs = [(c, s) for c, syms in uni.items() for s in syms if not want or s in want]43 mp = ROOT / "results" / "manifests" / "panels_manifest.json"44 man = json.loads(mp.read_text()) if mp.exists() else {}45 with ProcessPoolExecutor(a.workers) as ex:46 futs = {ex.submit(one, c, s): (c, s) for c, s in jobs}47 for f in as_completed(futs):48 c, s = futs[f]49 try:50 rec = f.result()51 man[s] = rec52 print(f"{c:8s} {s:8s} rows={rec['rows']} {rec['first']}..{rec['last']} valid_rm={rec['valid_rm_rate']:.3f}", flush=True)53 except Exception as e: # noqa: BLE00154 print(f"{c:8s} {s:8s} FAILED {e!r}", flush=True)55 man[s] = {"asset_class": c, "symbol": s, "error": repr(e)}56 mp.write_text(json.dumps(man, indent=1))575859if __name__ == "__main__":60 main()61