#!/usr/bin/env python3
"""Phase 2 — build the daily panels (returns + realized measures) for every ingested instrument.
Writes <cache>/panels/<symbol>.parquet and results/manifests/panels_manifest.json (row counts, date ranges, validity
rates, sha256; NO return values)."""
from __future__ import annotations

import argparse
import hashlib
import json
import sys
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from wp46.preprocessing import build_daily_panel  # noqa: E402
from wp46.preprocessing.panel import panel_path  # noqa: E402


def one(cls: str, sym: str) -> dict:
    df = build_daily_panel(cls, sym)
    p = panel_path(sym)
    p.parent.mkdir(parents=True, exist_ok=True)
    df.to_parquet(p, index=False)
    h = hashlib.sha256(p.read_bytes()).hexdigest()
    return {"asset_class": cls, "symbol": sym, "rows": int(len(df)), "first": str(df["session"].min().date()),
            "last": str(df["session"].max().date()), "valid_rm_rate": float(df["valid_rm"].mean()),
            "n_nan_r": int(df["r"].isna().sum()), "sha256": h, "path": str(p)}


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default=str(ROOT / "configs" / "instruments.yaml"))
    ap.add_argument("--symbols", default="")
    ap.add_argument("--workers", type=int, default=6)
    a = ap.parse_args()
    cfg = yaml.safe_load(open(a.config))
    uni = cfg.get("universe") or cfg["candidates"]
    want = set(a.symbols.split(",")) if a.symbols else None
    jobs = [(c, s) for c, syms in uni.items() for s in syms if not want or s in want]
    mp = ROOT / "results" / "manifests" / "panels_manifest.json"
    man = json.loads(mp.read_text()) if mp.exists() else {}
    with ProcessPoolExecutor(a.workers) as ex:
        futs = {ex.submit(one, c, s): (c, s) for c, s in jobs}
        for f in as_completed(futs):
            c, s = futs[f]
            try:
                rec = f.result()
                man[s] = rec
                print(f"{c:8s} {s:8s} rows={rec['rows']} {rec['first']}..{rec['last']} valid_rm={rec['valid_rm_rate']:.3f}", flush=True)
            except Exception as e:  # noqa: BLE001
                print(f"{c:8s} {s:8s} FAILED {e!r}", flush=True)
                man[s] = {"asset_class": c, "symbol": s, "error": repr(e)}
            mp.write_text(json.dumps(man, indent=1))


if __name__ == "__main__":
    main()