Case 3 — Set Up Drift Detection on Production Predictions

# Case 3 — Set Up Drift Detection on Production Predictions > *"Your churn model is live, but you have no monitoring beyond service health. Stand up drift detection: feature distribution checks, prediction monitoring, alerting."* --- ## Step 1 — Define what's normal Before monitoring "drift", define the baseline. Pull a stable training-time sample: ```python import pandas as pd baseline = pd.read_parquet('data/training_v3.parquet') baseline_stats = {} for col in feature_columns: series = baseline[col] baseline_stats[col] = { 'mean': float(series.mean()), 'std': float(series.std()), 'p25': float(series.quantile(0.25)), 'p50': float(series.quantile(0.50)), 'p75': float(series.quantile(0.75)), 'p95': float(series.quantile(0.95)), 'p99': float(series.quantile(0.99)), } # Save for monitoring with open('baselines/training_v3_stats.json', 'w') as f: json.dump(baseline_stats, f, indent=2) ``` Now you have "what features looked like at training time" as a reference. ## Step 2 — Capture production predictions The service logs predictions (Case 2 already does this). Aggregate them into a warehouse table: ```sql -- production.predictions CREATE TABLE production.predictions ( customer_id TEXT, served_at TIMESTAMP, feature_1 NUMERIC, feature_2 NUMERIC, -- ... all features ... churn_probability NUMERIC, model_version TEXT ); ``` Stream from the service logs. Daily updates fine; hourly better. ## Step 3 — Daily drift check job ```python # scripts/check_drift.py import json import numpy as np import pandas as pd from datetime import datetime, timedelta from sqlalchemy import create_engine from scipy.stats import ks_2samp def psi(expected, actual, buckets=10): expected_pct = pd.cut(expected, buckets, duplicates='drop').value_counts(normalize=True) actual_pct = pd.cut(actual, buckets, duplicates='drop').value_counts(normalize=True) expected_pct = expected_pct.where(expected_pct > 0, 0.0001) actual_pct = actual_pct.where(actual_pct > 0, 0.0001) return float(((actual_pct - expected_pct) * np.log(actual_pct / expected_pct)).sum()) def load_baseline(path): return pd.read_parquet(path) def load_recent_production(days=1): engine = create_engine(os.environ['DB_URL']) end = datetime.utcnow() start = end - timedelta(days=days) return pd.read_sql( f"SELECT * FROM production.predictions WHERE served_at >= '{start}' AND served_at < '{end}'", engine ) def check_drift(): baseline = load_baseline('data/training_v3.parquet') recent = load_recent_production(days=1) if len(recent) < 100: print('Not enough recent predictions, skipping.') return feature_cols = [c for c in baseline.columns if c not in ('target', 'customer_id')] drift_report = { 'checked_at': datetime.utcnow().isoformat(), 'recent_count': len(recent), 'features': {}, 'prediction_dist': {}, } # Per-feature drift for col in feature_cols: if col not in recent.columns: continue psi_score = psi(baseline[col], recent[col]) ks_stat, ks_p = ks_2samp(baseline[col], recent[col]) drift_report['features'][col] = { 'psi': psi_score, 'ks_stat': float(ks_stat), 'ks_p': float(ks_p), 'baseline_mean': float(baseline[col].mean()), 'recent_mean': float(recent[col].mean()), } # Prediction distribution drift if 'churn_probability' in recent.columns: drift_report['prediction_dist'] = { 'mean': float(recent['churn_probability'].mean()), 'p50': float(recent['churn_probability'].quantile(0.5)), 'p95': float(recent['churn_probability'].quantile(0.95)), } # Save report out_path = f'drift_reports/{datetime.utcnow():%Y-%m-%d}.json' with open(out_path, 'w') as f: json.dump(drift_report, f, indent=2) # Check thresholds and alert alerts = [] for col, stats in drift_report['features'].items(): if stats['psi'] > 0.25: alerts.append(f"PSI on {col}: {stats['psi']:.3f} (>0.25)") if stats['ks_p'] 0.02: alerts.append(f"Prediction mean drifted: {drift_report['prediction_dist']['mean']:.3f} (baseline 0.05)") if alerts: send_slack_alert('\n'.join(alerts)) return drift_report if __name__ == '__main__': check_drift() ``` Schedule: ```yaml # .github/workflows/drift-check.yml on: schedule: - cron: '0 8 * * *' # daily at 08:00 UTC jobs: drift: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - run: pip install -r requirements.txt - run: python scripts/check_drift.py