Continuous Eval Pipeline Drift Detection

Continuous Eval Pipeline Drift Detection: Catching Model Decay Before It Catches You

Last March, a recommendation model at a mid-size e-commerce company silently lost 14% of its conversion lift over six weeks. No alerts fired. Accuracy on the test set looked fine. The problem? The test set was frozen in January, but customer behavior shifted in February when a competitor launched a major promotion. The model wasn't broken — the world moved. By the time someone noticed the revenue dip, the damage was done.

This is the drift problem, and the only defense is continuous evaluation: a pipeline that doesn't just check whether your model is correct, but whether the data flowing through it still resembles the data it was trained on.

The Problem: Static Tests, Dynamic Worlds

Most ML teams ship a model with a held-out test set, measure F1 or RMSE, and call it done. That test set is a photograph. Production data is a river. Three types of drift can corrupt your river:

• **Data drift (covariate shift):** The input distribution changes. Users from a new demographic start using your app. Sensor calibration drifts. A new data source gets merged.
• **Concept drift:** The relationship between inputs and outputs changes. Spam filters face novel attack patterns. Stock market regimes shift. Seasonality evolves.
• **Prediction drift:** The model's output distribution changes, often a symptom of one of the above.

The industry insight from the 2024 "State of ML Ops" survey is blunt: 62% of production model failures are caused by drift, not bugs. Yet most monitoring stacks only watch latency and error rates — infrastructure signals, not statistical ones.

Why PSI Is the Workhorse

Think of drift detection like a smoke detector. You don't need to know what's burning — you need to know the air composition changed. The Population Stability Index (PSI) is that smoke detector for feature distributions.

PSI compares how two distributions allocate observations across the same set of bins. It's robust, interpretable, and doesn't assume normality. The interpretation is standardized:

| PSI | Interpretation |

|-----|---------------|

| < 0.10 | No significant drift |

| 0.10 – 0.25 | Moderate drift, investigate |

| > 0.25 | Significant drift, act now |

PSI works on any numeric feature, handles missing bins gracefully, and is cheap to compute — making it ideal for streaming pipelines where you evaluate thousands of batches per day.

A Continuous Eval Pipeline in Pure Python

Here's a working drift detection pipeline using only the standard library. It maintains a baseline distribution, evaluates incoming batches, and flags drift via PSI with an EWMA smoothing layer to reduce false positives from noisy batches.

import math
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Tuple

@dataclass
class DriftReport:
    feature: str
    psi: float
    smoothed_psi: float
    drifted: bool
    threshold: float

@dataclass
class FeatureMonitor:
    """Tracks drift for a single feature using PSI + EWMA smoothing."""
    baseline_bins: List[Tuple[float, float]]  # (bin_edge_low, bin_edge_high)
    baseline_probs: List[float]               # expected proportion per bin
    threshold: float = 0.20
    ewma_alpha: float = 0.30
    _ewma: float = field(default=0.0, repr=False)

    def _bin_counts(self, values: List[float]) -> List[int]:
        counts = [0] * len(self.baseline_bins)
        for v in values:
            for i, (lo, hi) in enumerate(self.baseline_bins):
                if lo <= v < hi or (i == len(self.baseline_bins) - 1 and v == hi):
                    counts[i] += 1
                    break
        return counts

    def compute_psi(self, current_values: List[float]) -> float:
        if not current_values:
            return 0.0
        counts = self._bin_counts(current_values)
        total = sum(counts)
        psi = 0.0
        for i, expected in enumerate(self.baseline_probs):
            actual = (counts[i] / total) if total > 0 else 0.0
            # Avoid log(0) — add small epsilon
            expected = max(expected, 1e-6)
            actual = max(actual, 1e-6)
            psi += (actual - expected) * math.log(actual / expected)
        return psi

    def evaluate(self, current_values: List[float]) -> DriftReport:
        psi = self.compute_psi(current_values)
        # EWMA smoothing: dampen single-batch noise
        self._ewma = self.ewma_alpha * psi + (1 - self.ewma_alpha) * self._ewma
        return DriftReport(
            feature="",  # set by pipeline
            psi=psi,
            smoothed_psi=self._ewma,
            drifted=self._ewma > self.threshold,
            threshold=self.threshold,
        )


def build_baseline(values: List[float], n_bins: int = 10) -> FeatureMonitor:
    """Construct a FeatureMonitor from a baseline sample using quantile bins."""
    sorted_vals = sorted(values)
    n = len(sorted_vals)
    quantiles = [sorted_vals[int(n * q / n_bins)] for q in range(n_bins)]
    quantiles.append(sorted_vals[-1])

    bins = [(quantiles[i], quantiles[i + 1]) for i in range(n_bins)]
    # Baseline probabilities are uniform by construction (quantile bins)
    probs = [1.0 / n_bins] * n_bins
    return FeatureMonitor(baseline_bins=bins, baseline_probs=probs)


@dataclass
class ContinuousEvalPipeline:
    monitors: Dict[str, FeatureMonitor] = field(default_factory=dict)
    alert_handler: Callable[[DriftReport], None] = field(default=lambda r: None)
    history: deque = field(default_factory=lambda: deque(maxlen=500))

    def register(self, feature: str, baseline_values: List[float]):
        self.monitors[feature] = build_baseline(baseline_values)

    def evaluate_batch(self, batch: Dict[str, List[float]]):
        """Run drift checks on a batch of production data."""
        for feature, monitor in self.monitors.items():
            if feature not in batch:
                continue
            report = monitor.evaluate(batch[feature])
            report.feature = feature
            self.history.append(report)
            if report.drifted:
                self.alert_handler(report)

    def summary(self) -> Dict[str, float]:
        return {
            f: round(m._ewma, 4) for f, m in self.monitors.items()
        }

Wiring It Into Production

The pipeline above is transport-agnostic. In practice, you call `evaluate_batch` from wherever your data lands — a Kafka consumer, a Lambda trigger, a scheduled Airflow task. Here's a minimal alert handler and a simulated run:

def pagerduty_alert(report: DriftReport):
    # In production: push to PagerDuty, Slack, or your incident system
    print(f"[ALERT] Drift on '{report.feature}': "
          f"PSI={report.psi:.4f} (smoothed={report.smoothed_psi:.4f}, "
          f"threshold={report.threshold})")

# --- Setup ---
import random
random.seed(42)

baseline = [random.gauss(50, 10) for _ in range(5000)]
pipeline = ContinuousEvalPipeline(alert_handler=pagerduty_alert)
pipeline.register("session_duration", baseline)

# --- Simulate production batches ---
for batch_num in range(20):
    # After batch 10, inject drift: mean shifts from 50 to 58
    mean = 58 if batch_num >= 10 else 50
    batch_data = {
        "session_duration": [random.gauss(mean, 10) for _ in range(500)]
    }
    pipeline.evaluate_batch(batch_data)

print("Final PSI summary:", pipeline.summary())

You'll see the smoothed PSI climb past the threshold around batch 12-13 — two batches after the drift begins, which is the EWMA lag working as designed. Without smoothing, batch 11 alone might trigger a false positive from sampling noise.

Key Takeaways

• **Drift is the dominant failure mode in production ML.** Infrastructure monitoring (latency, memory, 5xx errors) won't catch it. You need statistical monitoring.
• **PSI is the best default detector.** It's distribution-agnostic, cheap, and has industry-standard thresholds. Use it as your first line of defense on every numeric feature.
• **Smooth before you alert.** Single-batch PSI is noisy. An EWMA layer (alpha ≈ 0.2–0.3) dramatically reduces false positives while keeping detection latency acceptable.
• **Baseline on quantile bins, not equal-width bins.** Quantile bins give uniform baseline probabilities, which makes PSI maximally sensitive to any distributional change.
• **Concept drift needs label feedback.** PSI detects input drift. To catch concept drift, you need delayed ground-truth labels flowing back into the same pipeline — log predictions, join with outcomes, and run the same statistical tests on error distributions.
• **Make drift detection a CI gate, not just an alert.** When retraining pipelines run, the drift detector should be a precondition: if PSI on the new training data exceeds 0.25 versus the last production model's baseline, block the deploy and require human review.

What's Next

At AmtocSoft, we're building automated eval pipelines that integrate drift detection directly into content generation workflows — so when your input distribution shifts, you know before your users do. Check out our companion code repository for the full pipeline with Kafka integration and concept-drift detection extensions. For a deeper dive into building self-healing retraining triggers, read our earlier post on automated ML retraining pipelines.

Companion code

Written with AI assistance — reviewed by Toc Am