Drift Detection — Covariate Drift Cost a Fraud Model $2M

Every ML model has an expiry date — you just don't know when it is. The moment you deploy a model to production, the clock starts ticking. Real-world data is a living thing: customer behaviour shifts, sensor calibrations drift, economic conditions flip, and language evolves. A model trained on yesterday's data makes yesterday's decisions, and in fast-moving domains that gap kills business value silently and expensively. Unlike a crashed server, a drifting model doesn't throw an error. It just quietly becomes wrong.

The core problem is that ML models are frozen snapshots of a world that keeps moving. Traditional software has deterministic logic you can test; a model's 'logic' is baked into millions of learned parameters that have no automatic self-correction mechanism. When the statistical relationship between your input features and your target label changes, the model has no way of knowing. It will keep producing confident predictions that are increasingly divorced from reality — and your monitoring stack needs to catch that before your users or your business does.

By the end of this article you'll be able to implement a production-grade monitoring pipeline that detects covariate drift, concept drift, and prediction drift using PSI, KL divergence, and the Kolmogorov-Smirnov test. You'll understand which detector to reach for in which situation, the statistical subtleties that trip up even experienced engineers, and how to wire all of it into an alerting workflow that won't wake you up for false positives at 3 a.m.

What Is Model Monitoring and Drift Detection?

Model monitoring is the practice of continuously observing a deployed ML model's performance and input data. Drift detection identifies when the statistical properties of the data or the relationship between inputs and outputs change from the training baseline. Without monitoring, you're flying blind: your model could be making decisions based on patterns that no longer exist.

In production, you need to detect all three. Each requires a different statistical test and a different response.

from typing import List, Optional
import numpy as np
from scipy.stats import ks_2samp

def detect_covariate_drift(
    train: np.ndarray,
    production: np.ndarray,
    threshold: float = 0.1
) -> Optional[float]:
    """Detect covariate drift using two-sample KS test.
    Namespace: io.thecodeforge.monitoring.drift
    """
    if len(train) == 0 or len(production) == 0:
        return None
    stat, p_value = ks_2samp(train, production)
    if p_value < 0.05 and stat > threshold:
        return stat
    return None

Statistical Tests: PSI, KL Divergence, and Kolmogorov-Smirnov

Three tests dominate production drift detection:

1. Population Stability Index (PSI): Measures how much a variable's distribution has shifted between two samples. Formula: sum((actual_prop_i - expected_prop_i) * ln(actual_prop_i / expected_prop_i)). PSI < 0.1 = no shift, 0.1–0.25 = minor, > 0.25 = significant.
2. KL Divergence: Measures the information lost when using expected distribution to approximate actual. Asymmetric — order matters. Use PSI for symmetric stability, KL for asymmetrical change detection.
3. Kolmogorov-Smirnov (KS) Test: Non-parametric test comparing two empirical distributions. Returns a statistic (max difference) and a p-value. Works for continuous features. More sensitive than PSI for location shifts.

In practice, use PSI for categorical/binned features, KS for continuous. KL divergence is useful when you care about directionality of change.

import numpy as np
from scipy.stats import ks_2samp
from io.thecodeforge.monitoring.stats import psi

def compute_drift_report(train: np.ndarray, prod: np.ndarray, bins: int = 10):
    train_bins = np.histogram(train, bins=bins)[0] / len(train)
    prod_bins = np.histogram(prod, bins=bins)[0] / len(prod)
    psi_value = psi(train_bins, prod_bins)
    ks_stat, p_value = ks_2samp(train, prod)
    return {
        'psi': round(psi_value, 4),
        'ks_stat': round(ks_stat, 4),
        'ks_p_value': round(p_value, 6),
        'drift_detected': psi_value > 0.25 or (p_value < 0.05 and ks_stat > 0.1)
    }

Building a Production Monitoring Pipeline

A robust monitoring pipeline has four layers:

1. Data collection: Log model inputs and outputs to a time-series store (e.g., Kafka + InfluxDB). Store at least 30 days of raw feature vectors and predictions.
2. Drift computation: Run scheduled jobs (e.g., Airflow DAG every 6 hours) that compute PSI, KS, and prediction drift for each feature vs. the training baseline. Store results in a separate metrics table.
3. Alerting: Tiered alerts: INFO (PSI 0.1–0.2), WARNING (0.2–0.3), CRITICAL (>0.3). Confirm drift over at least two consecutive windows before paging. Avoid single-day spikes that are just noise.
4. Retraining trigger: When drift exceeds threshold and is confirmed, automatically trigger a retraining job with the latest 30 days of production data. Validate on a recent holdout set before deploying.

This architecture separates detection from action — you can tune alerts without affecting retraining logic.

from datetime import datetime, timedelta
import pandas as pd
from io.thecodeforge.monitoring.drift import compute_all_drift
from io.thecodeforge.alerts import evaluate_alert

def run_monitoring_check():
    now = datetime.utcnow()
    window_start = now - timedelta(days=30)
    # Load production features from the last 30 days
    prod_data = load_features(start_time=window_start, end_time=now)
    # Load training baseline (stored as parquet)
    train = pd.read_parquet('s3://model-baselines/latest/train_features.parquet')
    # Compute drift for each feature
    drift_results = compute_all_drift(train, prod_data)
    # Evaluate alert severity
    alert = evaluate_alert(drift_results)
    if alert.severity in ['WARNING', 'CRITICAL']:
        trigger_retraining_job(reason=alert.summary)
    log_metrics(drift_results, alert)

Common Pitfalls in Drift Detection

Even seasoned MLOps teams make these mistakes:

1. Testing drift on the wrong baseline: Always compare against the training data distribution, not a previous production snapshot. Production distributions shift gradually — if you compare against last month, you'll miss long-term drift.
2. Ignoring feature interactions: Drift in one feature may be harmless when another feature compensates. For example, if 'age' drifts up but 'income' drifts up proportionally, the model may still work. Single-feature drift tests alone can cause false alarms.
3. Using only p-values: A tiny p-value with a tiny KS statistic (e.g., 0.02) may be statistically significant but practically irrelevant. Always check effect size alongside p-value.
4. Not handling missing data: If production data is missing for a feature, the distribution collapses to a spike at 0, which looks like extreme drift. Handle missing values explicitly before computing tests.

Advanced: Multivariate Drift Detection and A/B Testing Integration

Single-feature tests scale linearly but miss interactions. For high-dimensional models (e.g., embeddings, tabular with 100+ features), use:

• Maximum Mean Discrepancy (MMD): A kernel-based test that compares two high-dimensional distributions. More powerful than per-feature tests but computationally expensive.
• Drift Detection on Model Embeddings: If your model has a latent layer (e.g., 64-dim), compute PSI on the embedding distribution. This catches joint shifts that single features miss.
• A/B Test Validation: When you deploy a new model version, run both models in shadow mode for a week. Compute drift between the candidate's predictions and the champion's. Treat prediction distribution divergence as a prerequisite for go-live.

In production, combine single-feature tests for explainability with multivariate tests for sensitivity. This gives you both the 'what changed' and the 'where to look'.

from sklearn.metrics import pairwise_kernels
import numpy as np
from io.thecodeforge.monitoring.mmd import mmd_test

def detect_embedding_drift(
    train_embeddings: np.ndarray,
    prod_embeddings: np.ndarray,
    kernel: str = 'rbf',
    threshold: float = 0.05
) -> bool:
    """Detect drift in high-dimensional embeddings using MMD."""
    stat, p_value = mmd_test(train_embeddings, prod_embeddings, kernel=kernel)
    return p_value < threshold  # significant drift

Frequently Asked Questions