# filename: src/monitoring/drift_detector.py # purpose: KS drift detection on live classifier confidence scores. # Compares rolling live buffer against val-split baseline (Section 7). # # Rules applied: # Rule 4: src/ never imports from api/ — RedisCache injected as constructor arg # Rule 21: KS test on confidence scores (not raw embeddings) # Rule 23: Redis optional — deque fallback when unavailable # Rule 33: label name "class_name" consistent across all Prometheus metrics from __future__ import annotations import json import logging import time from collections import deque from pathlib import Path from typing import TYPE_CHECKING, Optional import numpy as np from prometheus_client import Gauge from scipy.stats import ks_2samp import config as cfg if TYPE_CHECKING: from src.inference.redis_cache import RedisCache logger = logging.getLogger(__name__) # ── Prometheus Gauges — module-level singletons ─────────────────────────────── # Defined here (not api/prometheus_metrics.py) because Rule 4 prohibits src/ # from importing api/. The global prometheus_client registry collects all metrics # regardless of definition location — GET /metrics still exposes all of them. ks_drift_statistic = Gauge( "smartvision_ks_drift_statistic", "KS D-statistic from last test (0=no drift, 1=complete separation)", ["class_name"], ) ks_drift_p_value = Gauge( "smartvision_ks_drift_p_value", "KS p-value from last test (1.0=no evidence of drift, <0.05=drift likely)", ["class_name"], ) ks_drift_alert = Gauge( "smartvision_ks_drift_alert", "1 if KS stat > threshold AND p < 0.05, else 0 (double-gate prevents false positives)", ["class_name"], ) live_buffer_size = Gauge( "smartvision_live_buffer_size", "Number of live inference scores in rolling buffer", ["class_name"], ) # ── Rate-limit constant ─────────────────────────────────────────────────────── KS_RUN_EVERY_N: int = cfg.KS_RUN_EVERY_N # run KS every N new samples per class # ── Helper ──────────────────────────────────────────────────────────────────── def _safe_float(v: object) -> Optional[float]: """Return Python float or None for nan/inf (prevents json.dumps failure). scipy.stats.ks_2samp returns nan pvalue on zero-variance live distributions on some scipy versions. Storing None propagates cleanly to Pydantic Optional[float]. """ try: f = float(v) # type: ignore[arg-type] return f if (f == f and abs(f) != float("inf")) else None except (TypeError, ValueError): return None # ── DriftDetector ───────────────────────────────────────────────────────────── class DriftDetector: """Per-class KS drift detector for classifier confidence scores. Architecture ------------ - Baseline: val-split MobileNet confidence scores loaded from .npy files (22 classes × 30 samples, generated in Section 7). - Live buffer: rolling deque(maxlen=200) per class, optionally persisted to Redis list ``sv:drift:live:{class_name}`` so state survives API restarts. - KS test: runs every KS_RUN_EVERY_N new samples once buffer >= min_samples. Rate-limit prevents blocking the asyncio event loop on high-traffic paths. - Alert: double-gate (stat > threshold AND p < 0.05) to avoid ~95% false- positive rate with n=30 baseline samples. Thread safety ------------- ``record()`` and ``get_status()`` are both synchronous and run in the asyncio event loop thread. Dict assignments in ``_run_ks()`` are GIL-protected single operations. DO NOT add ``await`` points to ``record()`` or ``_run_ks()``, and DO NOT call them via ``run_in_executor()``, without adding a ``threading.Lock`` around ``_last_results`` writes. """ MAX_BUFFER = 200 # maximum live scores kept per class def __init__( self, baseline_path: Path, redis_client: "RedisCache", min_samples: int = 100, alert_threshold: float = 0.10, ) -> None: # Baseline JSON is required — raise immediately if missing (not graceful-degrade) baseline_path = Path(baseline_path) if not baseline_path.exists(): raise FileNotFoundError( f"Drift baseline not found: {baseline_path}. " "Run notebooks/06_model_comparison.py to generate it." ) self._min_samples = min_samples self._alert_threshold = alert_threshold self._redis = redis_client # Load baseline metadata with baseline_path.open() as f: self._baseline_meta: dict = json.load(f) # Resolve .npy paths relative to the JSON file's directory (not CWD) baseline_dir = baseline_path.parent self._baselines: dict[str, np.ndarray] = {} for cls_name, entry in self._baseline_meta.get("classes", {}).items(): npy_path = baseline_dir / entry["scores_file"] if not npy_path.exists(): logger.warning( f"Baseline scores missing for '{cls_name}': {npy_path}. Skipping." ) continue self._baselines[cls_name] = np.load(str(npy_path)) logger.info( f"DriftDetector: loaded baseline for {len(self._baselines)} classes" ) # Rolling buffers — deque provides O(1) append and automatic maxlen eviction self._buffers: dict[str, deque[float]] = { cls: deque(maxlen=self.MAX_BUFFER) for cls in self._baselines } # Per-class call counter for KS rate-limiting self._counters: dict[str, int] = {cls: 0 for cls in self._baselines} # Last KS result per class — written in _run_ks, read in get_status self._last_results: dict[str, dict] = {} # Step 9: Initialize all Gauge label series before any classify request # so Prometheus has time series from startup (alert rules evaluate empty sets otherwise) for cls in cfg.CLASSES: ks_drift_statistic.labels(class_name=cls).set(0.0) ks_drift_p_value.labels(class_name=cls).set( 1.0 ) # 1.0 = "no evidence of drift" ks_drift_alert.labels(class_name=cls).set(0.0) live_buffer_size.labels(class_name=cls).set(0.0) # Step 10-11: Restore buffers from Redis; update Gauges to reflect restored state for cls in self._baselines: redis_key = f"sv:drift:live:{cls}" restored = redis_client.get_list(redis_key) # newest-first (LPUSH order) if restored: # Reverse to chronological order before extending deque so that # maxlen eviction removes the oldest item (not the most recently restored) self._buffers[cls].extend(reversed(restored)) live_buffer_size.labels(class_name=cls).set(len(self._buffers[cls])) logger.info( f" Restored {len(self._buffers[cls])} scores for '{cls}' from Redis" ) # ── Public API ──────────────────────────────────────────────────────────── def record(self, class_name: str, confidence: float) -> None: """Record one live inference score; run KS if buffer threshold met. Called from classify route after every non-cached inference. Cache hits MUST NOT call record() — cached responses double-count scores. """ if class_name not in self._baselines: return # silently skip "unknown" and any class without baseline confidence = max(0.0, min(1.0, float(confidence))) # clamp to [0, 1] self._buffers[class_name].append(confidence) # Push to Redis for cross-restart persistence (best-effort) if self._redis.available: self._redis.push_to_list( f"sv:drift:live:{class_name}", confidence, self.MAX_BUFFER ) buf_len = len(self._buffers[class_name]) live_buffer_size.labels(class_name=class_name).set(buf_len) self._counters[class_name] += 1 if ( buf_len >= self._min_samples and self._counters[class_name] % KS_RUN_EVERY_N == 0 ): self._run_ks(class_name, list(self._buffers[class_name])) def get_status(self) -> dict: """Return per-class KS results + summary counts for /drift/status. All numeric values are Python native types (float/int/bool). ``_safe_float`` in ``_run_ks`` ensures no np.float32 or nan/inf leaks here. """ classes_out: dict[str, dict] = {} for cls in self._baselines: buf_len = len(self._buffers[cls]) result = self._last_results.get(cls) classes_out[cls] = { "buffer_size": buf_len, "min_samples_required": self._min_samples, "tested": result is not None, **(result if result is not None else {}), } alerting = [cls for cls, r in self._last_results.items() if r.get("is_alert")] return { "summary": { "total_classes": len(self._baselines), "classes_with_data": sum( 1 for b in self._buffers.values() if len(b) > 0 ), "classes_tested": len(self._last_results), "classes_alerting": len(alerting), "baseline_model": self._baseline_meta.get("model_used", "mobilenet"), "baseline_split": self._baseline_meta.get("split_used", "val"), }, "classes": classes_out, } # ── Internal ────────────────────────────────────────────────────────────── def _run_ks(self, class_name: str, live_scores: list[float]) -> None: """Run scipy KS test and update Prometheus Gauges + _last_results. THREADING NOTE: This method is intentionally synchronous and runs in the asyncio event loop thread (called only from record()). DO NOT add await points here or call from run_in_executor() without adding a threading.Lock around _last_results to prevent partial-update races. """ baseline = self._baselines[class_name] live = np.array(live_scores, dtype=np.float32) result = ks_2samp(baseline, live) stat = float(result.statistic) pval = float(result.pvalue) # Double-gate alert: stat alone has ~95% false-positive rate with n=30 baseline # (KS variance is high at small reference sample sizes). # Requiring p < 0.05 as second gate eliminates false positives. is_alert = bool(stat > self._alert_threshold and pval < 0.05) # _safe_float converts nan/inf (degenerate inputs) to None so _last_results # is always JSON-safe without a round-trip in get_status() safe_stat = _safe_float(stat) safe_pval = _safe_float(pval) self._last_results[class_name] = { "ks_stat": safe_stat, "p_value": safe_pval, "is_alert": is_alert, "n_live": int(len(live_scores)), "n_baseline": int(len(baseline)), "tested_at": float(time.time()), } # Fall back to neutral Gauge values if stat/pval are nan/inf ks_drift_statistic.labels(class_name=class_name).set( safe_stat if safe_stat is not None else 0.0 ) ks_drift_p_value.labels(class_name=class_name).set( safe_pval if safe_pval is not None else 1.0 ) ks_drift_alert.labels(class_name=class_name).set(1.0 if is_alert else 0.0) if is_alert: logger.warning( f"Drift alert: class={class_name!r} stat={stat:.4f} p={pval:.6f} " f"n_live={len(live_scores)} n_baseline={len(baseline)}" )