ML Monitoring and Drift Detection: Keeping Production Models Honest

The most expensive bug I ever shipped was not a software bug—it was a model that silently degraded over six weeks. A fraud detection model that had been catching 94% of fraudulent transactions dropped to 71% without anyone noticing. By the time we caught it, the company had lost over $400,000 to undetected fraud. The model had not crashed. It had not thrown an error. It had simply become less accurate as the world changed around it.

That failure shaped my entire approach to ML monitoring. After nine years of deploying models in production, I have learned that monitoring is not optional—it is the difference between a model that delivers sustained value and one that becomes a liability. This is a practical guide to ML monitoring and drift detection, grounded in real production systems and the hard lessons of what happens when you do not monitor.

Why ML Monitoring Is Different

Monitoring a traditional software service is straightforward: you track latency, error rates, throughput, and resource utilization. If the service returns 200 OK in under 100ms, it is working.

Monitoring an ML model is fundamentally harder because:

1. Silent degradation: A model can return perfectly valid predictions that are increasingly wrong. The API works fine; the model does not.
2. Delayed feedback: You often do not know the ground truth for days or weeks (e.g., whether a flagged transaction was actually fraud).
3. Data dependency: Model performance depends on the data distribution, which changes over time in ways you cannot control.
4. Multiple failure modes: Data quality issues, concept drift, feature drift, label drift, and infrastructure problems all manifest differently.

The Three Types of Drift

Data Drift (Covariate Shift)

The distribution of input features changes, but the relationship between features and target remains the same.

Example: A model trained on summer transaction data sees winter transaction patterns—different amounts, different merchants, different timing.

import numpy as np
from scipy import stats

def detect_data_drift(reference_data: np.ndarray,
                      current_data: np.ndarray,
                      feature_names: list[str],
                      threshold: float = 0.05) -> dict:
    """Detect data drift using Kolmogorov-Smirnov test."""
    drift_results = {}

    for i, feature in enumerate(feature_names):
        ks_stat, p_value = stats.ks_2samp(reference_data[:, i], current_data[:, i])
        drift_results[feature] = {
            "ks_statistic": ks_stat,
            "p_value": p_value,
            "drift_detected": p_value < threshold,
        }

    return drift_results

Concept Drift

The relationship between features and target changes. The same input features now map to different outputs.

Example: A fraud detection model trained on pre-pandemic data sees post-pandemic spending patterns. What was unusual behavior in 2019 (large online purchases) became normal in 2020.

def detect_concept_drift(reference_predictions: np.ndarray,
                         current_predictions: np.ndarray,
                         reference_labels: np.ndarray = None,
                         current_labels: np.ndarray = None) -> dict:
    """Detect concept drift by comparing prediction distributions."""
    # Compare prediction distributions
    ks_stat, p_value = stats.ks_2samp(reference_predictions, current_predictions)

    results = {
        "prediction_distribution_drift": {
            "ks_statistic": ks_stat,
            "p_value": p_value,
            "drift_detected": p_value < 0.05,
        }
    }

    # If labels are available, compare accuracy
    if reference_labels is not None and current_labels is not None:
        from sklearn.metrics import accuracy_score
        ref_acc = accuracy_score(reference_labels,
                                 (reference_predictions > 0.5).astype(int))
        curr_acc = accuracy_score(current_labels,
                                  (current_predictions > 0.5).astype(int))
        results["accuracy_change"] = {
            "reference_accuracy": ref_acc,
            "current_accuracy": curr_acc,
            "absolute_change": curr_acc - ref_acc,
            "significant_drop": (ref_acc - curr_acc) > 0.05,
        }

    return results

Label Drift (Prior Probability Shift)

The distribution of the target variable changes, independent of the features.

Example: A churn prediction model trained during a period of high churn (15%) sees deployment during a period of low churn (5%). The model over-predicts churn because its baseline is wrong.

Population Stability Index (PSI)

PSI is my go-to metric for drift detection. It is intuitive, well-understood, and works for both categorical and continuous features.

def calculate_psi(reference: np.ndarray, current: np.ndarray,
                  bins: int = 10) -> float:
    """Calculate Population Stability Index."""
    # Create bins from reference data
    breakpoints = np.percentile(reference, np.linspace(0, 100, bins + 1))
    breakpoints = np.unique(breakpoints)  # Handle duplicate values

    # Count proportions in each bin
    ref_counts = np.histogram(reference, bins=breakpoints)[0]
    curr_counts = np.histogram(current, bins=breakpoints)[0]

    # Convert to proportions (add small epsilon to avoid division by zero)
    epsilon = 1e-4
    ref_proportions = ref_counts / ref_counts.sum() + epsilon
    curr_proportions = curr_counts / curr_counts.sum() + epsilon

    # Calculate PSI
    psi = np.sum(
        (curr_proportions - ref_proportions) *
        np.log(curr_proportions / ref_proportions)
    )

    return psi

def interpret_psi(psi: float) -> str:
    """Interpret PSI value."""
    if psi < 0.1:
        return "no_significant_drift"
    elif psi < 0.2:
        return "moderate_drift"
    else:
        return "significant_drift"

PSI Monitoring Dashboard

class PSIMonitor:
    def __init__(self, reference_data: dict, feature_names: list[str]):
        self.reference_data = reference_data
        self.feature_names = feature_names
        self.history = []

    def check(self, current_data: dict) -> dict:
        """Check PSI for all features."""
        results = {}
        for feature in self.feature_names:
            psi = calculate_psi(
                self.reference_data[feature],
                current_data[feature],
            )
            results[feature] = {
                "psi": psi,
                "status": interpret_psi(psi),
            }

        # Record history
        self.history.append({
            "timestamp": datetime.utcnow().isoformat(),
            "results": results,
        })

        return results

    def get_drift_summary(self) -> dict:
        """Summarize drift status across all features."""
        if not self.history:
            return {}

        latest = self.history[-1]["results"]
        drifted = [
            f for f, r in latest.items()
            if r["status"] != "no_significant_drift"
        ]

        return {
            "total_features": len(self.feature_names),
            "drifted_features": len(drifted),
            "drifted_feature_names": drifted,
            "overall_status": "degraded" if len(drifted) > len(self.feature_names) * 0.2 else "healthy",
        }

KL-Divergence for Distribution Comparison

KL-divergence measures how one probability distribution differs from another. It is more sensitive than PSI for detecting subtle distribution shifts.

from scipy.stats import entropy

def calculate_kl_divergence(reference: np.ndarray, current: np.ndarray,
                            bins: int = 50) -> float:
    """Calculate KL-divergence between two distributions."""
    # Create common bins
    all_data = np.concatenate([reference, current])
    breakpoints = np.histogram_bin_edges(all_data, bins=bins)

    # Calculate histograms
    ref_hist, _ = np.histogram(reference, bins=breakpoints, density=True)
    curr_hist, _ = np.histogram(current, bins=breakpoints, density=True)

    # Add epsilon and normalize
    epsilon = 1e-10
    ref_hist = ref_hist + epsilon
    curr_hist = curr_hist + epsilon
    ref_hist = ref_hist / ref_hist.sum()
    curr_hist = curr_hist / curr_hist.sum()

    # KL-divergence: D_KL(P || Q) where P is reference, Q is current
    kl_div = entropy(ref_hist, curr_hist)

    return kl_div

Jensen-Shannon Divergence

For a symmetric measure, use JS-divergence (the average of KL-divergences in both directions):

def calculate_js_divergence(reference: np.ndarray, current: np.ndarray,
                            bins: int = 50) -> float:
    """Calculate Jensen-Shannon divergence."""
    all_data = np.concatenate([reference, current])
    breakpoints = np.histogram_bin_edges(all_data, bins=bins)

    ref_hist, _ = np.histogram(reference, bins=breakpoints, density=True)
    curr_hist, _ = np.histogram(current, bins=breakpoints, density=True)

    epsilon = 1e-10
    ref_hist = (ref_hist + epsilon) / (ref_hist + epsilon).sum()
    curr_hist = (curr_hist + epsilon) / (curr_hist + epsilon).sum()

    # M is the average distribution
    m = 0.5 * (ref_hist + curr_hist)

    js_div = 0.5 * entropy(ref_hist, m) + 0.5 * entropy(curr_hist, m)

    return js_div

Production Monitoring System

Here is a comprehensive monitoring system I have used in production:

import json
import time
from datetime import datetime, timedelta
from collections import deque
from dataclasses import dataclass, field

@dataclass
class MonitoringConfig:
    # Drift detection
    psi_warning_threshold: float = 0.1
    psi_critical_threshold: float = 0.2
    kl_divergence_threshold: float = 0.1

    # Performance monitoring
    accuracy_window_hours: int = 24
    accuracy_drop_threshold: float = 0.05

    # Data quality
    null_rate_threshold: float = 0.1
    value_range_checks: dict = field(default_factory=dict)

    # Alerting
    alert_cooldown_minutes: int = 30
    max_alerts_per_hour: int = 5

class MLModelMonitor:
    def __init__(self, model_name: str, config: MonitoringConfig,
                 reference_data: dict):
        self.model_name = model_name
        self.config = config
        self.reference_data = reference_data
        self.psi_monitor = PSIMonitor(reference_data, list(reference_data.keys()))

        # Prediction tracking
        self.predictions = deque(maxlen=100000)
        self.ground_truths = deque(maxlen=100000)

        # Alert management
        self.alert_history = []
        self.last_alert_time = {}

    def log_prediction(self, features: dict, prediction: float,
                       entity_id: str, timestamp: datetime = None):
        """Log a prediction for monitoring."""
        self.predictions.append({
            "features": features,
            "prediction": prediction,
            "entity_id": entity_id,
            "timestamp": timestamp or datetime.utcnow(),
        })

    def log_ground_truth(self, entity_id: str, actual: float,
                         timestamp: datetime = None):
        """Log ground truth for performance evaluation."""
        self.ground_truths.append({
            "entity_id": entity_id,
            "actual": actual,
            "timestamp": timestamp or datetime.utcnow(),
        })

    def run_checks(self) -> dict:
        """Run all monitoring checks."""
        results = {
            "timestamp": datetime.utcnow().isoformat(),
            "model_name": self.model_name,
            "checks": {},
        }

        # 1. Data drift check
        if self.predictions:
            current_features = self._extract_recent_features()
            drift_results = self.psi_monitor.check(current_features)
            results["checks"]["data_drift"] = {
                "status": self._assess_drift_status(drift_results),
                "details": drift_results,
            }

        # 2. Prediction distribution check
        if len(self.predictions) > 100:
            pred_drift = self._check_prediction_drift()
            results["checks"]["prediction_drift"] = pred_drift

        # 3. Performance check (if ground truth available)
        if self.ground_truths:
            perf = self._check_performance()
            results["checks"]["performance"] = perf

        # 4. Data quality check
        if self.predictions:
            dq = self._check_data_quality()
            results["checks"]["data_quality"] = dq

        # 5. Generate alerts
        alerts = self._generate_alerts(results)
        results["alerts"] = alerts

        return results

    def _extract_recent_features(self, hours: int = 1) -> dict:
        """Extract features from recent predictions."""
        cutoff = datetime.utcnow() - timedelta(hours=hours)
        recent = [
            p for p in self.predictions
            if p["timestamp"] > cutoff
        ]

        if not recent:
            return {}

        feature_names = list(recent[0]["features"].keys())
        return {
            name: np.array([p["features"][name] for p in recent])
            for name in feature_names
        }

    def _check_prediction_drift(self) -> dict:
        """Check if prediction distribution has shifted."""
        predictions = np.array([p["prediction"] for p in self.predictions])
        ref_predictions = np.array(self.reference_data.get("predictions", []))

        if len(ref_predictions) == 0:
            return {"status": "no_reference", "message": "No reference predictions available"}

        psi = calculate_psi(ref_predictions, predictions)
        kl = calculate_kl_divergence(ref_predictions, predictions)

        return {
            "psi": psi,
            "kl_divergence": kl,
            "status": interpret_psi(psi),
            "mean_prediction": float(np.mean(predictions)),
            "reference_mean": float(np.mean(ref_predictions)),
        }

    def _check_performance(self) -> dict:
        """Check model performance against ground truth."""
        cutoff = datetime.utcnow() - timedelta(hours=self.config.accuracy_window_hours)

        # Match predictions with ground truths
        gt_by_id = {
            gt["entity_id"]: gt["actual"]
            for gt in self.ground_truths
            if gt["timestamp"] > cutoff
        }

        matched = [
            (p["prediction"], gt_by_id[p["entity_id"]])
            for p in self.predictions
            if p["entity_id"] in gt_by_id and p["timestamp"] > cutoff
        ]

        if len(matched) < 100:
            return {"status": "insufficient_data", "matched_count": len(matched)}

        predictions, actuals = zip(*matched)
        predictions = np.array(predictions)
        actuals = np.array(actuals)

        from sklearn.metrics import (
            accuracy_score, precision_score, recall_score, f1_score, roc_auc_score
        )

        binary_predictions = (predictions > 0.5).astype(int)

        return {
            "status": "ok",
            "accuracy": accuracy_score(actuals, binary_predictions),
            "precision": precision_score(actuals, binary_predictions, zero_division=0),
            "recall": recall_score(actuals, binary_predictions, zero_division=0),
            "f1": f1_score(actuals, binary_predictions, zero_division=0),
            "auc_roc": roc_auc_score(actuals, predictions),
            "sample_count": len(matched),
        }

    def _check_data_quality(self) -> dict:
        """Check data quality of recent predictions."""
        recent = list(self.predictions)[-1000:]
        issues = []

        for feature_name in recent[0]["features"]:
            values = [p["features"].get(feature_name) for p in recent]
            null_count = sum(1 for v in values if v is None)
            null_rate = null_count / len(values)

            if null_rate > self.config.null_rate_threshold:
                issues.append({
                    "feature": feature_name,
                    "issue": "high_null_rate",
                    "null_rate": null_rate,
                    "threshold": self.config.null_rate_threshold,
                })

            # Check value ranges
            if feature_name in self.config.value_range_checks:
                range_config = self.config.value_range_checks[feature_name]
                non_null_values = [v for v in values if v is not None]
                if non_null_values:
                    min_val = min(non_null_values)
                    max_val = max(non_null_values)
                    if min_val < range_config.get("min", float("-inf")):
                        issues.append({
                            "feature": feature_name,
                            "issue": "below_min",
                            "value": min_val,
                            "threshold": range_config["min"],
                        })
                    if max_val > range_config.get("max", float("inf")):
                        issues.append({
                            "feature": feature_name,
                            "issue": "above_max",
                            "value": max_val,
                            "threshold": range_config["max"],
                        })

        return {
            "status": "degraded" if issues else "ok",
            "issues": issues,
        }

    def _assess_drift_status(self, drift_results: dict) -> str:
        """Assess overall drift status."""
        drifted_count = sum(
            1 for r in drift_results.values()
            if r["status"] != "no_significant_drift"
        )
        if drifted_count == 0:
            return "healthy"
        elif drifted_count < len(drift_results) * 0.2:
            return "warning"
        else:
            return "critical"

    def _generate_alerts(self, results: dict) -> list[dict]:
        """Generate alerts based on monitoring results."""
        alerts = []
        now = datetime.utcnow()

        # Check data drift
        drift_status = results["checks"].get("data_drift", {}).get("status")
        if drift_status == "critical":
            alerts.append({
                "level": "critical",
                "type": "data_drift",
                "message": f"Significant data drift detected in model {self.model_name}",
                "timestamp": now.isoformat(),
            })

        # Check performance
        perf = results["checks"].get("performance", {})
        if perf.get("status") == "ok":
            ref_accuracy = self.reference_data.get("accuracy", 0)
            current_accuracy = perf.get("accuracy", 0)
            if ref_accuracy - current_accuracy > self.config.accuracy_drop_threshold:
                alerts.append({
                    "level": "critical",
                    "type": "performance_degradation",
                    "message": f"Accuracy dropped from {ref_accuracy:.3f} to {current_accuracy:.3f}",
                    "timestamp": now.isoformat(),
                })

        # Check data quality
        dq_status = results["checks"].get("data_quality", {}).get("status")
        if dq_status == "degraded":
            issues = results["checks"]["data_quality"]["issues"]
            alerts.append({
                "level": "warning",
                "type": "data_quality",
                "message": f"Data quality issues: {len(issues)} features affected",
                "timestamp": now.isoformat(),
                "details": issues,
            })

        # Deduplicate and throttle alerts
        filtered_alerts = self._throttle_alerts(alerts)
        self.alert_history.extend(filtered_alerts)

        return filtered_alerts

    def _throttle_alerts(self, alerts: list[dict]) -> list[dict]:
        """Throttle alerts to prevent alert fatigue."""
        now = datetime.utcnow()
        cooldown = timedelta(minutes=self.config.alert_cooldown_minutes)
        max_per_hour = self.config.max_alerts_per_hour

        # Count recent alerts
        recent_count = sum(
            1 for a in self.alert_history
            if datetime.fromisoformat(a["timestamp"]) > now - timedelta(hours=1)
        )

        if recent_count >= max_per_hour:
            return []

        filtered = []
        for alert in alerts:
            alert_type = alert["type"]
            last_time = self.last_alert_time.get(alert_type)
            if last_time is None or (now - last_time) > cooldown:
                filtered.append(alert)
                self.last_alert_time[alert_type] = now

        return filtered

Retraining Decision Framework

When drift is detected, the question becomes: should you retrain?

class RetrainingDecisionEngine:
    def __init__(self, monitor: MLModelMonitor):
        self.monitor = monitor

    def should_retrain(self) -> dict:
        """Determine if model should be retrained."""
        results = self.monitor.run_checks()
        reasons = []
        urgency = "low"

        # Check 1: Significant data drift
        drift = results["checks"].get("data_drift", {})
        if drift.get("status") == "critical":
            reasons.append("significant_data_drift")
            urgency = "high"

        # Check 2: Performance degradation
        perf = results["checks"].get("performance", {})
        if perf.get("status") == "ok":
            ref_acc = self.monitor.reference_data.get("accuracy", 0)
            curr_acc = perf.get("accuracy", 0)
            if ref_acc - curr_acc > 0.1:
                reasons.append("severe_performance_drop")
                urgency = "high"
            elif ref_acc - curr_acc > 0.05:
                reasons.append("moderate_performance_drop")
                urgency = "medium"

        # Check 3: Prediction distribution shift
        pred_drift = results["checks"].get("prediction_drift", {})
        if pred_drift.get("status") == "significant_drift":
            reasons.append("prediction_distribution_shift")
            if urgency == "low":
                urgency = "medium"

        # Check 4: Data quality issues
        dq = results["checks"].get("data_quality", {})
        if dq.get("status") == "degraded":
            reasons.append("data_quality_issues")
            if urgency == "low":
                urgency = "low"  # Data quality alone is low urgency

        return {
            "should_retrain": len(reasons) >= 2 or urgency == "high",
            "urgency": urgency,
            "reasons": reasons,
            "monitoring_results": results,
        }

Dashboards

Grafana Dashboard Configuration

{
  "dashboard": {
    "title": "ML Model Monitoring",
    "panels": [
      {
        "title": "Prediction Distribution",
        "type": "histogram",
        "targets": [
          {
            "expr": "ml_predictions{model='fraud_detection'}",
            "legendFormat": "Current"
          },
          {
            "expr": "ml_reference_predictions{model='fraud_detection'}",
            "legendFormat": "Reference"
          }
        ]
      },
      {
        "title": "Feature Drift (PSI)",
        "type": "heatmap",
        "targets": [
          {
            "expr": "ml_feature_psi{model='fraud_detection'}",
            "legendFormat": "{{feature}}"
          }
        ],
        "thresholds": [
          {"value": 0.1, "color": "yellow"},
          {"value": 0.2, "color": "red"}
        ]
      },
      {
        "title": "Model Accuracy Over Time",
        "type": "timeseries",
        "targets": [
          {
            "expr": "ml_accuracy{model='fraud_detection'}",
            "legendFormat": "Accuracy"
          }
        ]
      },
      {
        "title": "Data Quality Score",
        "type": "gauge",
        "targets": [
          {
            "expr": "ml_data_quality_score{model='fraud_detection'}",
            "legendFormat": "Quality Score"
          }
        ]
      }
    ]
  }
}

Lessons from Production

1. Monitor from day one: Do not wait until you have a monitoring system. Start with basic logging—predictions, features, timestamps. You can build monitoring on top of logs; you cannot retroactively create logs.

2. PSI is your friend: PSI is simple, intuitive, and effective. It catches most drift cases. Do not over-engineer your drift detection before PSI proves insufficient.

3. Alert on actionable things: An alert that no one acts on is noise. Every alert should have a clear owner and a clear action plan.

4. Separate monitoring from alerting: Log everything, alert selectively. You need the full picture for debugging, but you need focused alerts for action.

5. Ground truth is the bottleneck: The biggest challenge in ML monitoring is getting timely ground truth. Invest in faster feedback loops—automated labeling, human-in-the-loop sampling, proxy metrics.

6. Monitor the pipeline, not just the model: Data quality, feature freshness, inference latency, and throughput are all part of the model’s health. Monitor the entire pipeline.

7. Baseline is everything: You cannot detect drift without a baseline. Capture a reference distribution during the model’s “golden period” (when it is performing well) and compare against it.

8. Drift is not always bad: Seasonal patterns, business growth, and intentional changes can all cause drift. Context matters. A drift alert should trigger investigation, not automatic retraining.

Conclusion

ML monitoring is not a nice-to-have—it is a survival requirement for production ML systems. The models you deploy today will degrade over time as the world changes around them. The only question is whether you will catch the degradation before it costs you money, reputation, or trust.

Start with the basics: log predictions, compute PSI, track performance. Build from there. The monitoring system I have described here is battle-tested across fraud detection, risk scoring, and recommendation systems. It has caught multiple silent degradations before they became expensive failures.

The most important lesson: monitoring is not a technical problem—it is an organizational one. You need clear ownership, clear alerting, and clear action plans. A monitoring system that no one looks at is worse than no monitoring at all, because it creates a false sense of security.