spb/wp15_uqo

1"""Schema and integrity validation for raw monthly files (CLAUDE.md §5.1)."""23from __future__ import annotations45from pathlib import Path67import polars as pl89EXPECTED_SCHEMA = {10    "timestamp": pl.Datetime("ns", "UTC"),11    "open": pl.Float64,12    "high": pl.Float64,13    "low": pl.Float64,14    "close": pl.Float64,15    "volume": pl.Float64,16    "ticker": pl.String,17}181920def validate_schema(path: Path) -> list[str]:21    """Return list of schema problems (empty = pass)."""22    schema = pl.scan_parquet(path).collect_schema()23    problems = []24    for name, dtype in EXPECTED_SCHEMA.items():25        if name not in schema:26            problems.append(f"missing column {name}")27        elif schema[name] != dtype:28            problems.append(f"{name}: expected {dtype}, got {schema[name]}")29    for name in schema:30        if name not in EXPECTED_SCHEMA:31            problems.append(f"unexpected column {name}")32    return problems333435def integrity_flags(df: pl.DataFrame) -> pl.DataFrame:36    """Row-level integrity flags. Flag, never delete (CLAUDE.md §5.1)."""37    return df.with_columns(38        (39            (pl.col("low") > pl.min_horizontal("open", "close"))40            | (pl.col("high") < pl.max_horizontal("open", "close"))41            | (pl.col("low") > pl.col("high"))42        ).alias("flag_ohlc_violation"),43        (44            (pl.col("open") <= 0) | (pl.col("high") <= 0)45            | (pl.col("low") <= 0) | (pl.col("close") <= 0)46        ).alias("flag_nonpositive_price"),47        (pl.col("volume") < 0).alias("flag_negative_volume"),48        pl.struct(["ticker", "timestamp"]).is_duplicated().alias("flag_duplicate_key"),49    )505152def file_audit_summary(path: Path) -> dict:53    """Cheap whole-file audit used to enrich the manifest after download."""54    df = pl.read_parquet(path)55    flagged = integrity_flags(df)56    return {57        "row_count": df.height,58        "column_names": df.columns,59        "column_types": [str(t) for t in df.dtypes],60        "min_timestamp": str(df["timestamp"].min()),61        "max_timestamp": str(df["timestamp"].max()),62        "unique_symbol_count": df["ticker"].n_unique(),63        "duplicate_key_count": int(flagged["flag_duplicate_key"].sum()) // 2,64        "ohlc_violation_count": int(flagged["flag_ohlc_violation"].sum()),65        "null_count_by_column": {c: int(df[c].null_count()) for c in df.columns},66    }67