"""P1 coverage audit: regular-hours 1-minute coverage per instrument and year over the sample window, for the
selected universe and (optionally) the candidate pool, via symbol-scoped SQL. Output: results/metrics/coverage_audit.csv"""
from __future__ import annotations

import argparse
import json
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from chronos_x.data.client import FinDataClient  # noqa: E402
from chronos_x.data.calendar import session_frame  # noqa: E402

ROOT = Path(__file__).resolve().parents[1]
SQL = ("SELECT session_date, count(*) AS n_reg FROM {view} WHERE symbol = '{sym}' AND year BETWEEN {y0} AND {y1} "
       "AND CAST(timezone('America/New_York', timestamp_utc) AS TIME) BETWEEN TIME '09:30:00' AND TIME '16:00:00' "
       "GROUP BY 1")


def one(c: FinDataClient, sym: str, ac: str, y0: int, y1: int, expected: pd.DataFrame) -> dict:
    view = f"market_bars_{ac}_1m_raw"
    try:
        cols, rows, meta = c.sql(SQL.format(view=view, sym=sym, y0=y0, y1=y1), max_rows=100_000)
    except Exception as exc:  # noqa: BLE001
        return {"symbol": sym, "error": str(exc)[:200]}
    df = pd.DataFrame(rows, columns=cols)
    if df.empty:
        return {"symbol": sym, "coverage": 0.0, "sessions_present": 0}
    df["session_date"] = pd.to_datetime(df["session_date"]).dt.date
    m = expected.merge(df, on="session_date", how="left").fillna({"n_reg": 0})
    m["cov"] = (m["n_reg"] / m["n_slots"]).clip(upper=1.0)
    m["year"] = pd.to_datetime(m["session_date"]).dt.year
    out = {"symbol": sym, "coverage": float(m["cov"].mean()), "sessions_present": int((m["n_reg"] > 0).sum()),
           "sessions_expected": int(len(m)), "first_session": str(m.loc[m["n_reg"] > 0, "session_date"].min()),
           "last_session": str(m.loc[m["n_reg"] > 0, "session_date"].max())}
    for y, cv in m.groupby("year")["cov"].mean().items():
        out[f"cov_{y}"] = float(cv)
    return out


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--manifest", default="data/manifests/universe_main.json")
    ap.add_argument("--start", default="2018-01-02"); ap.add_argument("--end", default="2025-12-31")
    ap.add_argument("--candidates", action="store_true", help="audit the whole candidate pool too")
    ap.add_argument("--workers", type=int, default=4)
    a = ap.parse_args()
    man = json.loads((ROOT / a.manifest).read_text())
    rows = [(r["symbol"], r["asset_class"]) for r in man["instruments"]]
    if a.candidates:
        cand = pd.read_csv(ROOT / "data" / "manifests" / "candidates_summary.csv")
        sel = {s for s, _ in rows}
        rows += [(r.symbol, r.asset_class) for r in cand.itertuples() if r.symbol not in sel]
    exp = session_frame(a.start, a.end)[["session_date", "n_slots"]]
    y0, y1 = int(a.start[:4]), int(a.end[:4])
    c = FinDataClient(log_path=ROOT / "results" / "logs" / "provenance_coverage.jsonl")
    t0 = time.time(); res = []
    with ThreadPoolExecutor(max_workers=a.workers) as ex:
        futs = [ex.submit(one, c, s, ac, y0, y1, exp) for s, ac in rows]
        for i, f in enumerate(as_completed(futs)):
            res.append(f.result())
            if (i + 1) % 50 == 0:
                print(f"{i + 1}/{len(rows)} {time.time() - t0:.0f}s", flush=True)
    df = pd.DataFrame(res).sort_values("symbol")
    out = ROOT / "results" / "metrics" / ("coverage_audit_candidates.csv" if a.candidates else "coverage_audit.csv")
    df.to_csv(out, index=False)
    print(df[["coverage"]].describe()); print(f"wall {time.time() - t0:.0f}s -> {out}")


if __name__ == "__main__":
    main()