spb/wp15_uqo

1"""Stage D.3 — clean minute panel, partitioned by year/month.23Per month:4  * duplicate (ticker, minute) keys resolved by the logged rule: open=first,5    high=max, low=min, close=last, volume=sum (DECISION_LOG 2026-07-13);6  * ET fields and session label using the XNYS calendar with TRUE close times7    (early closes truncate the regular session; bars after an early close are8    labelled 'post', cf. DATA_AUDIT §4);9  * integrity flag (OHLC violation) and split-window flag (±1 day around10    high-confidence split events) — flagged, never deleted;11  * written to data/processed/minute/year=YYYY/month=MM/data.parquet (zstd).1213The extended-hours "panel" is the same file filtered on session != 'rth' — one14store, both baselines. Idempotent per month; --force to rebuild.1516Usage: python scripts/08_build_clean_minute_panel.py [--workers N] [--month YYYY-MM]17"""1819from __future__ import annotations2021import argparse22import json23from concurrent.futures import ProcessPoolExecutor, as_completed24from pathlib import Path2526import polars as pl2728ROOT = Path(__file__).resolve().parents[1]29RAW = ROOT / "data" / "raw" / "data"30OUT = ROOT / "data" / "processed" / "minute"31SPLITS = ROOT / "outputs" / "diagnostics" / "stage_d" / "split_windows.parquet"32LOG = ROOT / "logs" / "preprocessing"3334RTH_START = 570353637def session_close_map() -> pl.DataFrame:38    """date -> regular close as minute-of-day ET (960, or 780 on early closes)."""39    import exchange_calendars as xcals40    cal = xcals.get_calendar("XNYS", start="1992-01-01", end="2026-04-01")41    closes = cal.closes.dt.tz_convert("America/New_York")42    return pl.DataFrame({43        "date": [d.date() for d in closes.index],44        "close_mod": [c.hour * 60 + c.minute for c in closes],45    }).with_columns(pl.col("date").cast(pl.Date))464748def build_month(month: str, force: bool = False) -> dict:49    year, mm = month.split("-")50    dest_dir = OUT / f"year={year}" / f"month={mm}"51    dest = dest_dir / "data.parquet"52    if dest.exists() and not force:53        return {"month": month, "skipped": True}5455    df = pl.read_parquet(RAW / f"ohlcv_{month}.parquet").sort("ticker", "timestamp")56    n_raw = df.height5758    df = (59        df.group_by("ticker", "timestamp", maintain_order=True)60        .agg(61            pl.col("open").first(),62            pl.col("high").max(),63            pl.col("low").min(),64            pl.col("close").last(),65            pl.col("volume").sum(),66        )67    )68    n_dedup = df.height6970    et = df.with_columns(71        pl.col("timestamp").dt.convert_time_zone("America/New_York").alias("ts_et"))72    et = et.with_columns(73        pl.col("ts_et").dt.date().alias("date"),74        (pl.col("ts_et").dt.hour().cast(pl.Int32) * 6075         + pl.col("ts_et").dt.minute()).alias("mod"),76    ).drop("ts_et")7778    et = et.join(session_close_map(), on="date", how="left")79    et = et.with_columns(80        pl.when(pl.col("mod") < RTH_START).then(pl.lit("pre"))81        .when(pl.col("mod") < pl.col("close_mod")).then(pl.lit("rth"))82        .otherwise(pl.lit("post")).alias("session"),83        (84            (pl.col("low") > pl.min_horizontal("open", "close"))85            | (pl.col("high") < pl.max_horizontal("open", "close"))86            | (pl.col("low") > pl.col("high"))87        ).alias("flag_ohlc"),88    ).drop("close_mod")8990    win = pl.read_parquet(SPLITS).with_columns(pl.lit(True).alias("flag_split_window"))91    et = et.join(win, on=["ticker", "date"], how="left").with_columns(92        pl.col("flag_split_window").fill_null(False))9394    dest_dir.mkdir(parents=True, exist_ok=True)95    tmp = dest_dir / ".data.parquet.tmp"96    et.sort("ticker", "timestamp").write_parquet(tmp, compression="zstd")97    tmp.replace(dest)9899    rep = {100        "month": month, "rows_raw": n_raw, "rows": n_dedup,101        "dup_rows_merged": n_raw - n_dedup,102        "rth": int((et["session"] == "rth").sum()),103        "flag_ohlc": int(et["flag_ohlc"].sum()),104        "flag_split_window": int(et["flag_split_window"].sum()),105    }106    LOG.mkdir(parents=True, exist_ok=True)107    with open(LOG / "08_panel_reports.jsonl", "a") as f:108        f.write(json.dumps(rep) + "\n")109    return rep110111112def main() -> None:113    ap = argparse.ArgumentParser()114    ap.add_argument("--workers", type=int, default=4)115    ap.add_argument("--month", default=None)116    ap.add_argument("--force", action="store_true")117    args = ap.parse_args()118119    months = ([args.month] if args.month else120              sorted(p.stem.replace("ohlcv_", "") for p in RAW.glob("ohlcv_*.parquet")))121    with ProcessPoolExecutor(max_workers=args.workers) as ex:122        futs = {ex.submit(build_month, m, args.force): m for m in months}123        done = 0124        for fut in as_completed(futs):125            rep = fut.result()126            done += 1127            print(f"[{done}/{len(months)}] {rep['month']}"128                  + (" skip" if rep.get("skipped") else129                     f" rows={rep['rows']:,} merged={rep['dup_rows_merged']}"),130                  flush=True)131    print("clean minute panel complete")132133134if __name__ == "__main__":135    main()136