spb/wp15_uqo

1"""Stage A pilot audit of mito0o852/OHLCV-1m monthly files.23Audits schema, timestamp convention, session coverage, duplicates,4OHLC validity, zeros/negatives, and symbol counts for the three pilot5months. Writes one JSON diagnostic per file to outputs/diagnostics/.67Usage:  python scripts/01_pilot_audit.py8"""910from __future__ import annotations1112import json13from pathlib import Path1415import polars as pl1617ROOT = Path(__file__).resolve().parents[1]18RAW = ROOT / "data" / "raw" / "data"19OUT = ROOT / "outputs" / "diagnostics"2021PILOT_MONTHS = ["1992-06", "2009-06", "2025-12"]222324def audit_month(month: str) -> dict:25    path = RAW / f"ohlcv_{month}.parquet"26    lf = pl.scan_parquet(path)27    schema = lf.collect_schema()28    df = lf.collect()29    n = df.height3031    ts_utc = df["timestamp"]32    ts_et = ts_utc.dt.convert_time_zone("America/New_York")33    et = df.with_columns(34        ts_et.alias("ts_et"),35        ts_et.dt.hour().cast(pl.Int32).alias("hour_et"),36        ts_et.dt.minute().cast(pl.Int32).alias("minute_et"),37        ts_et.dt.weekday().alias("weekday"),  # 1=Mon .. 7=Sun38        ts_et.dt.date().alias("date_et"),39        ts_utc.dt.second().alias("second"),40    )41    minute_of_day = et.select(42        (pl.col("hour_et") * 60 + pl.col("minute_et")).alias("mod")43    )["mod"]4445    rth = ((minute_of_day >= 570) & (minute_of_day < 960)).sum()  # 09:30-16:00 ET46    pre = (minute_of_day < 570).sum()47    post = (minute_of_day >= 960).sum()4849    dup = n - df.select(["ticker", "timestamp"]).unique().height5051    viol = df.filter(52        (pl.col("low") > pl.min_horizontal("open", "close"))53        | (pl.col("high") < pl.max_horizontal("open", "close"))54        | (pl.col("low") > pl.col("high"))55    ).height5657    nonpos_price = df.filter(58        (pl.col("open") <= 0) | (pl.col("high") <= 0)59        | (pl.col("low") <= 0) | (pl.col("close") <= 0)60    ).height61    neg_vol = df.filter(pl.col("volume") < 0).height62    zero_vol = df.filter(pl.col("volume") == 0).height63    nonint_vol = df.filter(pl.col("volume") != pl.col("volume").round(0)).height6465    nulls = {c: int(df[c].null_count()) for c in df.columns}66    weekend = et.filter(pl.col("weekday") >= 6).height67    nonzero_sec = (et["second"] != 0).sum()6869    by_day = (70        et.group_by("ticker", "date_et")71        .agg(pl.len().alias("bars"))72        .group_by("date_et")73        .agg(74            pl.len().alias("symbols"),75            pl.col("bars").median().alias("median_bars"),76        )77        .sort("date_et")78    )7980    mod_hist = (81        et.select((pl.col("hour_et") * 60 + pl.col("minute_et")).alias("mod"))82        .group_by("mod").agg(pl.len().alias("n")).sort("mod")83    )8485    return {86        "month": month,87        "rows": n,88        "columns": {name: str(dtype) for name, dtype in schema.items()},89        "unique_tickers": df["ticker"].n_unique(),90        "min_ts_utc": str(ts_utc.min()),91        "max_ts_utc": str(ts_utc.max()),92        "nonzero_seconds": int(nonzero_sec),93        "rows_rth_0930_1600_et": int(rth),94        "rows_premarket_et": int(pre),95        "rows_afterhours_et": int(post),96        "rth_share": round(float(rth) / n, 4),97        "duplicate_ticker_timestamp": int(dup),98        "ohlc_violations": int(viol),99        "nonpositive_prices": int(nonpos_price),100        "negative_volume": int(neg_vol),101        "zero_volume": int(zero_vol),102        "noninteger_volume": int(nonint_vol),103        "null_counts": nulls,104        "weekend_rows": int(weekend),105        "trading_days_observed": by_day.height,106        "symbols_per_day_median": float(by_day["symbols"].median()),107        "median_bars_per_symbol_day_overall": float(108            et.group_by("ticker", "date_et").agg(pl.len().alias("b"))["b"].median()109        ),110        "price_summary_close": {111            "min": float(df["close"].min()),112            "p50": float(df["close"].median()),113            "max": float(df["close"].max()),114        },115        "minute_of_day_hist": {116            str(r["mod"]): r["n"] for r in mod_hist.iter_rows(named=True)117        },118        "per_day": [119            {"date": str(r["date_et"]), "symbols": r["symbols"],120             "median_bars": r["median_bars"]}121            for r in by_day.iter_rows(named=True)122        ],123    }124125126def main() -> None:127    OUT.mkdir(parents=True, exist_ok=True)128    for month in PILOT_MONTHS:129        rep = audit_month(month)130        out = OUT / f"pilot_audit_{month}.json"131        out.write_text(json.dumps(rep, indent=1))132        keep = {k: v for k, v in rep.items()133                if k not in ("minute_of_day_hist", "per_day")}134        print(json.dumps(keep, indent=1))135136137if __name__ == "__main__":138    main()139