"""Schema and integrity validation for raw monthly files (CLAUDE.md §5.1)."""

from __future__ import annotations

from pathlib import Path

import polars as pl

EXPECTED_SCHEMA = {
    "timestamp": pl.Datetime("ns", "UTC"),
    "open": pl.Float64,
    "high": pl.Float64,
    "low": pl.Float64,
    "close": pl.Float64,
    "volume": pl.Float64,
    "ticker": pl.String,
}


def validate_schema(path: Path) -> list[str]:
    """Return list of schema problems (empty = pass)."""
    schema = pl.scan_parquet(path).collect_schema()
    problems = []
    for name, dtype in EXPECTED_SCHEMA.items():
        if name not in schema:
            problems.append(f"missing column {name}")
        elif schema[name] != dtype:
            problems.append(f"{name}: expected {dtype}, got {schema[name]}")
    for name in schema:
        if name not in EXPECTED_SCHEMA:
            problems.append(f"unexpected column {name}")
    return problems


def integrity_flags(df: pl.DataFrame) -> pl.DataFrame:
    """Row-level integrity flags. Flag, never delete (CLAUDE.md §5.1)."""
    return df.with_columns(
        (
            (pl.col("low") > pl.min_horizontal("open", "close"))
            | (pl.col("high") < pl.max_horizontal("open", "close"))
            | (pl.col("low") > pl.col("high"))
        ).alias("flag_ohlc_violation"),
        (
            (pl.col("open") <= 0) | (pl.col("high") <= 0)
            | (pl.col("low") <= 0) | (pl.col("close") <= 0)
        ).alias("flag_nonpositive_price"),
        (pl.col("volume") < 0).alias("flag_negative_volume"),
        pl.struct(["ticker", "timestamp"]).is_duplicated().alias("flag_duplicate_key"),
    )


def file_audit_summary(path: Path) -> dict:
    """Cheap whole-file audit used to enrich the manifest after download."""
    df = pl.read_parquet(path)
    flagged = integrity_flags(df)
    return {
        "row_count": df.height,
        "column_names": df.columns,
        "column_types": [str(t) for t in df.dtypes],
        "min_timestamp": str(df["timestamp"].min()),
        "max_timestamp": str(df["timestamp"].max()),
        "unique_symbol_count": df["ticker"].n_unique(),
        "duplicate_key_count": int(flagged["flag_duplicate_key"].sum()) // 2,
        "ohlc_violation_count": int(flagged["flag_ohlc_violation"].sum()),
        "null_count_by_column": {c: int(df[c].null_count()) for c in df.columns},
    }