MLOps Model Monitoring
Detect data drift, monitor model performance, set up alerts, and build retraining triggers for production ML systems.
Real-World Scenario
A bank deployed a fraud detection model in January. By August, the model’s precision had fallen from 94% to 71% — fraud patterns changed, payment methods shifted, and a new category of synthetic identity fraud appeared. The team only noticed when customer complaints spiked. With proper monitoring, they would have had 6 weeks of warning to retrain before production impact.
Statistical Drift Detection
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import Optional
@dataclass
class DriftResult:
feature_name: str
drift_detected: bool
p_value: float
statistic: float
method: str
severity: str # LOW | MEDIUM | HIGH
def ks_drift_test(
reference: np.ndarray,
production: np.ndarray,
feature_name: str,
threshold: float = 0.05,
) -> DriftResult:
"""Kolmogorov-Smirnov test: compare distributions without assuming normality."""
statistic, p_value = stats.ks_2samp(reference, production)
drift = p_value < threshold
if statistic > 0.3:
severity = "HIGH"
elif statistic > 0.1:
severity = "MEDIUM"
else:
severity = "LOW"
return DriftResult(
feature_name=feature_name,
drift_detected=drift,
p_value=p_value,
statistic=statistic,
method="KS",
severity=severity if drift else "NONE",
)
def psi_score(
reference: np.ndarray,
production: np.ndarray,
bins: int = 10,
) -> float:
"""Population Stability Index — standard metric in banking/credit scoring.
PSI < 0.1: stable, 0.1–0.2: minor shift, > 0.2: major shift (retrain)."""
# Build bins on reference distribution
breakpoints = np.quantile(reference, np.linspace(0, 1, bins + 1))
breakpoints[0] = -np.inf
breakpoints[-1] = np.inf
ref_counts = np.histogram(reference, bins=breakpoints)[0]
prod_counts = np.histogram(production, bins=breakpoints)[0]
# Avoid division by zero
ref_pct = (ref_counts + 0.5) / (len(reference) + 0.5 * bins)
prod_pct = (prod_counts + 0.5) / (len(production) + 0.5 * bins)
psi = np.sum((prod_pct - ref_pct) * np.log(prod_pct / ref_pct))
return float(psi)
# Simulate drift scenario
rng = np.random.default_rng(42)
# Training (reference) distribution
reference_age = rng.normal(35, 10, 10000)
reference_income = rng.lognormal(10.8, 0.5, 10000)
reference_balance = rng.exponential(5000, 10000)
# Production distribution — age and income shifted
production_age = rng.normal(42, 12, 1000) # older users
production_income = rng.lognormal(10.5, 0.6, 1000) # lower income
production_balance = rng.exponential(5000, 1000) # unchanged
features = {
"age": (reference_age, production_age),
"income": (reference_income, production_income),
"balance": (reference_balance, production_balance),
}
print("=== Drift Detection Report ===\n")
for name, (ref, prod) in features.items():
result = ks_drift_test(ref, prod, name)
psi = psi_score(ref, prod)
print(f"{name:10s}: KS={result.statistic:.3f} (p={result.p_value:.4f}) | "
f"PSI={psi:.3f} | Drift={'YES ⚠' if result.drift_detected else 'NO'} | {result.severity}")Performance Monitoring with Label Delay
import numpy as np
import pandas as pd
from sklearn.metrics import accuracy_score, f1_score
from datetime import datetime, timedelta
rng = np.random.default_rng(42)
# Simulate 6 months of predictions + delayed labels
n_months = 6
months = []
for month in range(n_months):
n = 1000
# True accuracy degrades over time (concept drift)
true_accuracy = 0.94 - month * 0.04 # 94% → 74% over 6 months
y_true = rng.integers(0, 2, n)
# Model gets worse over time
flip_rate = 1 - true_accuracy
y_pred = y_true.copy()
flip_idx = rng.choice(n, int(n * flip_rate), replace=False)
y_pred[flip_idx] = 1 - y_pred[flip_idx]
months.append({
"month": f"2024-{month+1:02d}",
"n_predictions": n,
"accuracy": accuracy_score(y_true, y_pred),
"f1": f1_score(y_true, y_pred),
})
df = pd.DataFrame(months)
print("Monthly Model Performance:")
print(df.to_string(index=False))
# Alert when accuracy drops below threshold
ALERT_THRESHOLD = 0.85
degraded = df[df["accuracy"] < ALERT_THRESHOLD]
if not degraded.empty:
first_alert = degraded.iloc[0]
print(f"\n🚨 ALERT: Accuracy dropped below {ALERT_THRESHOLD:.0%} "
f"in {first_alert['month']} ({first_alert['accuracy']:.2%})")
print(" Action: Initiate retraining with recent data")Building a Monitoring Pipeline
import json
import hashlib
import numpy as np
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional
@dataclass
class MonitoringAlert:
timestamp: str
alert_type: str # DATA_DRIFT | PERFORMANCE_DEGRADATION | SCHEMA_CHANGE
severity: str # LOW | MEDIUM | HIGH | CRITICAL
feature: Optional[str]
metric_name: str
metric_value: float
threshold: float
message: str
class ModelMonitor:
"""Production model monitoring with drift detection and alerting."""
def __init__(
self,
model_name: str,
reference_data: np.ndarray,
feature_names: list[str],
performance_threshold: float = 0.85,
psi_threshold: float = 0.2,
):
self.model_name = model_name
self.reference = reference_data
self.feature_names = feature_names
self.perf_threshold = performance_threshold
self.psi_threshold = psi_threshold
self.alerts: list[MonitoringAlert] = []
def check_drift(self, production_data: np.ndarray) -> list[MonitoringAlert]:
"""Run drift checks on all features."""
new_alerts = []
for i, feature in enumerate(self.feature_names):
ref_col = self.reference[:, i]
prod_col = production_data[:, i]
psi = psi_score(ref_col, prod_col)
ks = ks_drift_test(ref_col, prod_col, feature)
if psi > self.psi_threshold:
severity = "HIGH" if psi > 0.4 else "MEDIUM"
alert = MonitoringAlert(
timestamp = datetime.utcnow().isoformat(),
alert_type = "DATA_DRIFT",
severity = severity,
feature = feature,
metric_name = "PSI",
metric_value= psi,
threshold = self.psi_threshold,
message = f"Feature '{feature}' PSI={psi:.3f} exceeds threshold {self.psi_threshold}",
)
new_alerts.append(alert)
self.alerts.append(alert)
return new_alerts
def check_performance(self, y_true: np.ndarray, y_pred: np.ndarray,
metric_fn=None) -> Optional[MonitoringAlert]:
"""Alert if performance drops below threshold."""
from sklearn.metrics import accuracy_score
metric_fn = metric_fn or accuracy_score
score = metric_fn(y_true, y_pred)
if score < self.perf_threshold:
alert = MonitoringAlert(
timestamp = datetime.utcnow().isoformat(),
alert_type = "PERFORMANCE_DEGRADATION",
severity = "CRITICAL" if score < self.perf_threshold * 0.9 else "HIGH",
feature = None,
metric_name = "accuracy",
metric_value = score,
threshold = self.perf_threshold,
message = f"Model accuracy {score:.2%} below threshold {self.perf_threshold:.2%}",
)
self.alerts.append(alert)
return alert
return None
def report(self) -> dict:
return {
"model": self.model_name,
"total_alerts": len(self.alerts),
"critical": sum(1 for a in self.alerts if a.severity == "CRITICAL"),
"high": sum(1 for a in self.alerts if a.severity == "HIGH"),
"alerts": [asdict(a) for a in self.alerts[-5:]], # last 5
}
# Usage
rng = np.random.default_rng(42)
reference = rng.standard_normal((5000, 4))
production_drifted = rng.standard_normal((1000, 4))
production_drifted[:, 0] += 1.5 # feature 0 drifted significantly
monitor = ModelMonitor(
model_name="fraud-detector-v2",
reference_data=reference,
feature_names=["transaction_amount", "user_age", "session_duration", "device_risk"],
)
drift_alerts = monitor.check_drift(production_drifted)
for alert in drift_alerts:
print(f"🚨 {alert.severity}: {alert.message}")
print(json.dumps(monitor.report(), indent=2))Frequently Asked Questions
What is data drift and why does it cause model degradation?
Data drift is when the statistical distribution of input features changes between training and production. A model trained on summer traffic data deployed in winter sees different patterns. The model's learned decision boundaries are now wrong for the new distribution. Without monitoring, accuracy degrades silently for weeks before anyone notices.
What is the difference between data drift and concept drift?
Data drift (covariate shift) is when the input distribution P(X) changes. Concept drift is when the relationship between inputs and outputs P(y|X) changes — the same inputs now have different correct answers. Concept drift is harder to detect because you need ground truth labels, which often arrive with a delay.