1"""Quality assurance on raw bars (never alters prices; produces audit tables and row masks)."""2from __future__ import annotations34import numpy as np5import polars as pl678def qa_report(bars: pl.DataFrame) -> dict:9 """Row-level integrity checks on a bars frame (one or more symbols). Returns counts, never mutates."""10 n = bars.height11 if n == 0:12 return {"rows": 0}13 ohlc_bad = bars.filter((pl.col("high") < pl.col("low")) | (pl.col("open") > pl.col("high")) |14 (pl.col("open") < pl.col("low")) | (pl.col("close") > pl.col("high")) |15 (pl.col("close") < pl.col("low"))).height16 nonpos = bars.filter((pl.col("open") <= 0) | (pl.col("high") <= 0) | (pl.col("low") <= 0) |17 (pl.col("close") <= 0)).height18 negvol = bars.filter(pl.col("volume") < 0).height19 zerovol = bars.filter(pl.col("volume") == 0).height20 dup = n - bars.select(["instrument_id", "timestamp_utc"]).unique().height21 offint = bars.filter(pl.col("timestamp_utc").dt.second() != 0).height22 flagged = bars.filter(pl.col("quality_flags").list.len() > 0).height23 revs = bars.select(pl.col("revision_id").unique()).to_series().to_list()24 return {"rows": n, "ohlc_inconsistent": ohlc_bad, "nonpositive_price": nonpos, "negative_volume": negvol,25 "zero_volume": zerovol, "duplicates": dup, "off_interval": offint, "flagged_rows": flagged,26 "revision_ids": sorted(int(r) for r in revs), "symbols": bars["symbol_as_reported"].n_unique(),27 "first_ts": str(bars["timestamp_utc"].min()), "last_ts": str(bars["timestamp_utc"].max())}282930def dedupe(bars: pl.DataFrame) -> pl.DataFrame:31 """Keep the highest revision per (instrument, timestamp). The lake may publish the last session twice (base+delta)."""32 return (bars.sort(["instrument_id", "timestamp_utc", "revision_id"])33 .unique(subset=["instrument_id", "timestamp_utc"], keep="last", maintain_order=True))343536def price_jump_mask(close: np.ndarray, valid: np.ndarray, thresh: float = 0.25) -> np.ndarray:37 """Flag consecutive valid closes whose absolute log change exceeds `thresh` (vendor code: price_jump)."""38 out = np.zeros_like(valid, dtype=bool)39 idx = np.flatnonzero(valid)40 if len(idx) < 2:41 return out42 r = np.abs(np.diff(np.log(close[idx])))43 out[idx[1:][r > thresh]] = True44 return out454647def session_coverage(bars: pl.DataFrame, expected: pl.DataFrame) -> pl.DataFrame:48 """Per (symbol, session): regular-hours bars present / expected slots. `expected` has session_date, n_slots."""49 reg = bars.filter(pl.col("slot") >= 0).group_by(["symbol_as_reported", "session_date"]).agg(50 pl.len().alias("present"), pl.col("volume").sum().alias("volume"))51 cov = reg.join(expected, on="session_date", how="left")52 return cov.with_columns((pl.col("present") / pl.col("n_slots")).alias("coverage")).sort(53 ["symbol_as_reported", "session_date"])54