scripts/generate_drift_report.py

"""Generate the Evidently data-drift HTML report.

Reads:
- Reference: ``data/reference_dataset.parquet`` (training sample, frozen)
- Current: last N production predictions from Supabase, ordered by most
  recent (ignores history beyond that window)

The "last N" approach makes the pipeline replayable: re-run seed_traffic
and re-run this script — the report reflects the most recent batch
without needing to truncate the DB.

Writes an HTML report consumable by the Streamlit dashboard.

Usage:
    uv run python scripts/generate_drift_report.py                  # last 100
    uv run python scripts/generate_drift_report.py --limit 500
    uv run python scripts/generate_drift_report.py --output dashboard/static/drift_report.html
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import warnings
from pathlib import Path

import pandas as pd
from sqlalchemy import create_engine, text

logger = logging.getLogger("scripts.generate_drift_report")
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

DEFAULT_REFERENCE = Path("data/reference_dataset.parquet")
DEFAULT_OUTPUT = Path("dashboard/static/drift_report.html")
DEFAULT_FEATURE_NAMES = Path("models/feature_names.json")


def _load_reference(path: Path, feature_names: list[str]) -> pd.DataFrame:
    df = pd.read_parquet(path)
    keep = [c for c in feature_names if c in df.columns]
    missing = set(feature_names) - set(df.columns)
    if missing:
        logger.warning(
            "%d feature(s) absent from reference dataset (drift skipped on those)",
            len(missing),
        )
    return df[keep].copy()


def _load_current(database_url: str, limit: int, feature_names: list[str]) -> pd.DataFrame:
    """Pull the last ``limit`` successful predictions, ordered by most recent.

    Returns the JSONB feature payloads expanded into a wide DataFrame for
    Evidently. Newer rows come first in SQL, but order does not matter for
    drift (the test is distribution-based, not temporal).
    """
    engine = create_engine(database_url, future=True)
    with engine.connect() as conn:
        rows = conn.execute(
            text(
                "SELECT features FROM predictions_log "
                "WHERE status_code = 200 "
                "ORDER BY timestamp DESC "
                "LIMIT :lim"
            ),
            {"lim": limit},
        ).all()
    if not rows:
        raise SystemExit(
            "No successful prediction rows in predictions_log — generate some "
            "traffic first (scripts/seed_traffic.py)."
        )
    flat = pd.json_normalize([r.features for r in rows])
    keep = [c for c in feature_names if c in flat.columns]
    logger.info(
        "Loaded %d production rows (last %d requested), %d features available",
        len(flat), limit, len(keep),
    )
    return flat[keep].copy()


def _build_report(reference: pd.DataFrame, current: pd.DataFrame):
    """Import Evidently lazily so the script still works when only inspecting.

    Returns the evaluation object produced by ``Report.run``. The 0.7+ API
    no longer mutates ``report`` in place — ``.run()`` returns a separate
    snapshot that carries ``save_html``.
    """
    try:
        from evidently import Report
        from evidently.presets import DataDriftPreset
    except ImportError as exc:
        raise SystemExit(
            "evidently not installed. Add it to dashboard/requirements.txt "
            "or `uv add --group dev 'evidently>=0.7'`."
        ) from exc

    report = Report([DataDriftPreset()])
    # Evidently emits dozens of "divide by zero" RuntimeWarnings from scipy
    # when statistical tests run on near-constant columns. These are benign
    # (the resulting NaN/inf is interpreted as "skip this feature" downstream)
    # and they drown out real log output. Silence them only inside this call.
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", category=RuntimeWarning)
        warnings.filterwarnings("ignore", category=FutureWarning)
        return report.run(reference, current)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--reference", type=Path, default=DEFAULT_REFERENCE)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument(
        "--limit",
        type=int,
        default=100,
        help="Compare against the last N successful predictions (default %(default)s).",
    )
    parser.add_argument("--feature-names", type=Path, default=DEFAULT_FEATURE_NAMES)
    parser.add_argument("--database-url", default=os.getenv("DATABASE_URL"))
    args = parser.parse_args()

    if not args.database_url:
        # Allow loading from database/.env for local runs.
        try:
            from dotenv import load_dotenv

            load_dotenv(Path("database/.env"))
            args.database_url = os.getenv("DATABASE_URL")
        except ImportError:
            pass
    if not args.database_url:
        raise SystemExit("DATABASE_URL is required (env var or database/.env).")

    feature_names = json.loads(args.feature_names.read_text())
    reference = _load_reference(args.reference, feature_names)
    current = _load_current(args.database_url, args.limit, feature_names)

    # Align columns: drop any column not in BOTH frames (Evidently expects parity)
    common = [c for c in reference.columns if c in current.columns]
    reference = reference[common]
    current = current[common]
    logger.info("Aligned on %d common features", len(common))

    # Evidently rejects all-NaN columns ("empty column ... was provided for drift
    # calculation"). Drop any column that is fully null on either side.
    empty_ref = {c for c in reference.columns if reference[c].isna().all()}
    empty_cur = {c for c in current.columns if current[c].isna().all()}
    empty = empty_ref | empty_cur
    if empty:
        sample = ", ".join(sorted(empty)[:5])
        logger.warning(
            "Dropping %d all-NaN column(s) (empty in ref=%d, in current=%d). Sample: %s%s",
            len(empty), len(empty_ref), len(empty_cur), sample,
            " ..." if len(empty) > 5 else "",
        )
        reference = reference.drop(columns=list(empty))
        current = current.drop(columns=list(empty))
    logger.info("Comparing on %d features after NaN filter", len(reference.columns))

    evaluation = _build_report(reference, current)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    evaluation.save_html(str(args.output))
    logger.info("Saved drift report to %s", args.output)

    # Also dump a JSON snapshot of the run so the dashboard can parse
    # per-feature drift scores without scraping the HTML.
    json_path = args.output.with_suffix(".json")
    try:
        json_payload = evaluation.json()
        if not isinstance(json_payload, str):
            import json as _json
            json_payload = _json.dumps(json_payload)
        json_path.write_text(json_payload, encoding="utf-8")
        logger.info("Saved drift report JSON to %s", json_path)
    except Exception as exc:  # noqa: BLE001
        logger.warning("Could not save JSON snapshot of the drift report: %s", exc)


if __name__ == "__main__":
    main()