1"""P1 coverage audit: regular-hours 1-minute coverage per instrument and year over the sample window, for the2selected universe and (optionally) the candidate pool, via symbol-scoped SQL. Output: results/metrics/coverage_audit.csv"""3from __future__ import annotations45import argparse6import json7import sys8import time9from concurrent.futures import ThreadPoolExecutor, as_completed10from pathlib import Path1112import pandas as pd1314sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))15from chronos_x.data.client import FinDataClient # noqa: E40216from chronos_x.data.calendar import session_frame # noqa: E4021718ROOT = Path(__file__).resolve().parents[1]19SQL = ("SELECT session_date, count(*) AS n_reg FROM {view} WHERE symbol = '{sym}' AND year BETWEEN {y0} AND {y1} "20 "AND CAST(timezone('America/New_York', timestamp_utc) AS TIME) BETWEEN TIME '09:30:00' AND TIME '16:00:00' "21 "GROUP BY 1")222324def one(c: FinDataClient, sym: str, ac: str, y0: int, y1: int, expected: pd.DataFrame) -> dict:25 view = f"market_bars_{ac}_1m_raw"26 try:27 cols, rows, meta = c.sql(SQL.format(view=view, sym=sym, y0=y0, y1=y1), max_rows=100_000)28 except Exception as exc: # noqa: BLE00129 return {"symbol": sym, "error": str(exc)[:200]}30 df = pd.DataFrame(rows, columns=cols)31 if df.empty:32 return {"symbol": sym, "coverage": 0.0, "sessions_present": 0}33 df["session_date"] = pd.to_datetime(df["session_date"]).dt.date34 m = expected.merge(df, on="session_date", how="left").fillna({"n_reg": 0})35 m["cov"] = (m["n_reg"] / m["n_slots"]).clip(upper=1.0)36 m["year"] = pd.to_datetime(m["session_date"]).dt.year37 out = {"symbol": sym, "coverage": float(m["cov"].mean()), "sessions_present": int((m["n_reg"] > 0).sum()),38 "sessions_expected": int(len(m)), "first_session": str(m.loc[m["n_reg"] > 0, "session_date"].min()),39 "last_session": str(m.loc[m["n_reg"] > 0, "session_date"].max())}40 for y, cv in m.groupby("year")["cov"].mean().items():41 out[f"cov_{y}"] = float(cv)42 return out434445def main() -> None:46 ap = argparse.ArgumentParser()47 ap.add_argument("--manifest", default="data/manifests/universe_main.json")48 ap.add_argument("--start", default="2018-01-02"); ap.add_argument("--end", default="2025-12-31")49 ap.add_argument("--candidates", action="store_true", help="audit the whole candidate pool too")50 ap.add_argument("--workers", type=int, default=4)51 a = ap.parse_args()52 man = json.loads((ROOT / a.manifest).read_text())53 rows = [(r["symbol"], r["asset_class"]) for r in man["instruments"]]54 if a.candidates:55 cand = pd.read_csv(ROOT / "data" / "manifests" / "candidates_summary.csv")56 sel = {s for s, _ in rows}57 rows += [(r.symbol, r.asset_class) for r in cand.itertuples() if r.symbol not in sel]58 exp = session_frame(a.start, a.end)[["session_date", "n_slots"]]59 y0, y1 = int(a.start[:4]), int(a.end[:4])60 c = FinDataClient(log_path=ROOT / "results" / "logs" / "provenance_coverage.jsonl")61 t0 = time.time(); res = []62 with ThreadPoolExecutor(max_workers=a.workers) as ex:63 futs = [ex.submit(one, c, s, ac, y0, y1, exp) for s, ac in rows]64 for i, f in enumerate(as_completed(futs)):65 res.append(f.result())66 if (i + 1) % 50 == 0:67 print(f"{i + 1}/{len(rows)} {time.time() - t0:.0f}s", flush=True)68 df = pd.DataFrame(res).sort_values("symbol")69 out = ROOT / "results" / "metrics" / ("coverage_audit_candidates.csv" if a.candidates else "coverage_audit.csv")70 df.to_csv(out, index=False)71 print(df[["coverage"]].describe()); print(f"wall {time.time() - t0:.0f}s -> {out}")727374if __name__ == "__main__":75 main()76