"""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()