Model Monitoring

Models degrade in production. Data distributions shift, user behavior changes, and upstream systems evolve. Without monitoring, you won't know until business metrics tank. Good monitoring catches problems early - before customers notice.

Where It Is Used

Model monitoring is essential for any ML in production:

• All ML systems: Performance tracking, drift detection
• High-stakes: Healthcare, finance, autonomous systems
• Real-time: Recommendations, fraud, pricing
• Batch: Forecasting, scoring pipelines
• Regulatory: Compliance and audit requirements

Products: Evidently AI, WhyLabs, Fiddler, Arize, MLflow, custom solutions.

Typical Monitoring Signals

What You Do in Practice

1. Define baselines - What does "normal" look like?
2. Identify key metrics - Model performance and data quality
3. Set up data collection - Log inputs, outputs, and labels
4. Implement drift detection - Statistical tests on features
5. Configure alerts - Actionable, not noisy
6. Build dashboards - Visibility for team
7. Establish playbooks - What to do when alerts fire
8. Automate responses - Retraining triggers, fallbacks
9. Review regularly - Are monitors still relevant?

Baseline First

Start with simple monitoring before complex solutions:

import pandas as pd
import numpy as np
from scipy import stats

class SimpleMonitor:
    def __init__(self, reference_data):
        """Initialize with reference/training data."""
        self.reference = reference_data
        self.reference_stats = self._compute_stats(reference_data)

    def _compute_stats(self, data):
        return {
            'mean': data.mean(),
            'std': data.std(),
            'min': data.min(),
            'max': data.max(),
            'missing_rate': data.isnull().mean()
        }

    def check_data_quality(self, current_data):
        """Check for data quality issues."""
        issues = []

        current_missing = current_data.isnull().mean()
        ref_missing = self.reference_stats['missing_rate']

        for col in current_data.columns:
            if current_missing[col] > ref_missing[col] * 2:
                issues.append(f"High missing rate: {col}")

        return issues

    def check_drift(self, current_data, threshold=0.05):
        """Simple distribution drift check using KS test."""
        drift_detected = {}

        for col in current_data.columns:
            if current_data[col].dtype in ['float64', 'int64']:
                stat, p_value = stats.ks_2samp(
                    self.reference[col].dropna(),
                    current_data[col].dropna()
                )
                drift_detected[col] = {
                    'drifted': p_value < threshold,
                    'p_value': p_value,
                    'statistic': stat
                }

        return drift_detected

Why simple baseline: - Catches major issues - Low maintenance overhead - Easy to understand - Foundation for more complex monitoring

Data Drift Detection

Detect when input distributions change:

from scipy import stats
import numpy as np

def detect_numerical_drift(reference, current, method='ks'):
    """Detect drift in numerical features."""
    if method == 'ks':
        # Kolmogorov-Smirnov test
        stat, p_value = stats.ks_2samp(reference, current)
    elif method == 'psi':
        # Population Stability Index
        stat = calculate_psi(reference, current)
        p_value = None  # PSI doesn't have p-value
    elif method == 'wasserstein':
        # Wasserstein distance
        stat = stats.wasserstein_distance(reference, current)
        p_value = None

    return {'statistic': stat, 'p_value': p_value}

def calculate_psi(reference, current, bins=10):
    """Calculate Population Stability Index."""
    # Create bins from reference
    breakpoints = np.percentile(reference, np.linspace(0, 100, bins + 1))
    breakpoints[0] = -np.inf
    breakpoints[-1] = np.inf

    # Calculate proportions
    ref_counts = np.histogram(reference, bins=breakpoints)[0]
    cur_counts = np.histogram(current, bins=breakpoints)[0]

    ref_props = ref_counts / len(reference)
    cur_props = cur_counts / len(current)

    # Avoid division by zero
    ref_props = np.clip(ref_props, 0.0001, None)
    cur_props = np.clip(cur_props, 0.0001, None)

    psi = np.sum((cur_props - ref_props) * np.log(cur_props / ref_props))
    return psi

def detect_categorical_drift(reference, current, method='chi2'):
    """Detect drift in categorical features."""
    if method == 'chi2':
        # Chi-squared test
        ref_counts = reference.value_counts()
        cur_counts = current.value_counts()

        # Align categories
        all_categories = set(ref_counts.index) | set(cur_counts.index)
        ref_aligned = [ref_counts.get(c, 0) for c in all_categories]
        cur_aligned = [cur_counts.get(c, 0) for c in all_categories]

        stat, p_value = stats.chisquare(cur_aligned, ref_aligned)

    return {'statistic': stat, 'p_value': p_value}

PSI interpretation: - PSI < 0.1: No significant drift - 0.1 ≤ PSI < 0.2: Moderate drift, investigate - PSI ≥ 0.2: Significant drift, action required

Prediction Monitoring

Track model outputs over time:

class PredictionMonitor:
    def __init__(self, baseline_predictions):
        self.baseline_mean = baseline_predictions.mean()
        self.baseline_std = baseline_predictions.std()
        self.baseline_distribution = baseline_predictions

    def check_prediction_drift(self, current_predictions):
        """Check if predictions have drifted."""
        current_mean = current_predictions.mean()
        current_std = current_predictions.std()

        # Mean shift detection
        mean_shift = abs(current_mean - self.baseline_mean) / self.baseline_std

        # Distribution test
        stat, p_value = stats.ks_2samp(
            self.baseline_distribution, 
            current_predictions
        )

        return {
            'mean_shift_zscore': mean_shift,
            'ks_statistic': stat,
            'ks_pvalue': p_value,
            'alert': mean_shift > 2 or p_value < 0.01
        }

    def check_confidence_distribution(self, confidences, threshold=0.5):
        """Monitor confidence score distribution."""
        low_confidence_rate = (confidences < threshold).mean()
        high_confidence_rate = (confidences > 0.9).mean()

        return {
            'low_confidence_rate': low_confidence_rate,
            'high_confidence_rate': high_confidence_rate,
            'mean_confidence': confidences.mean()
        }

Performance Monitoring

Track actual model performance when labels are available:

from sklearn.metrics import accuracy_score, precision_score, recall_score
import pandas as pd

class PerformanceMonitor:
    def __init__(self, baseline_metrics, window_size=1000):
        self.baseline = baseline_metrics
        self.window_size = window_size
        self.history = []

    def add_prediction(self, prediction, label, timestamp):
        """Add a prediction-label pair."""
        self.history.append({
            'prediction': prediction,
            'label': label,
            'timestamp': timestamp
        })

        # Keep only recent history
        if len(self.history) > self.window_size * 2:
            self.history = self.history[-self.window_size:]

    def get_current_metrics(self):
        """Calculate metrics on recent window."""
        recent = self.history[-self.window_size:]

        if len(recent) < 100:  # Not enough data
            return None

        preds = [h['prediction'] for h in recent]
        labels = [h['label'] for h in recent]

        return {
            'accuracy': accuracy_score(labels, preds),
            'precision': precision_score(labels, preds, zero_division=0),
            'recall': recall_score(labels, preds, zero_division=0),
            'sample_size': len(recent)
        }

    def check_degradation(self, threshold=0.1):
        """Check if performance has degraded from baseline."""
        current = self.get_current_metrics()

        if current is None:
            return {'status': 'insufficient_data'}

        degradation = {}
        for metric in ['accuracy', 'precision', 'recall']:
            baseline_val = self.baseline.get(metric, 0)
            current_val = current.get(metric, 0)

            if baseline_val > 0:
                drop = (baseline_val - current_val) / baseline_val
                degradation[metric] = {
                    'baseline': baseline_val,
                    'current': current_val,
                    'relative_drop': drop,
                    'alert': drop > threshold
                }

        return degradation

Alerting Strategy

Good alerts are actionable, not noisy:

class AlertManager:
    def __init__(self, config):
        self.config = config
        self.alert_history = []

    def evaluate_alert(self, metric_name, value, threshold, comparison='gt'):
        """Evaluate if alert should fire."""
        if comparison == 'gt':
            triggered = value > threshold
        elif comparison == 'lt':
            triggered = value < threshold
        else:
            triggered = abs(value - threshold) > 0

        # Check for alert suppression (don't spam)
        if triggered and self._recently_alerted(metric_name):
            return {'triggered': False, 'reason': 'suppressed'}

        if triggered:
            self._record_alert(metric_name)

        return {
            'triggered': triggered,
            'metric': metric_name,
            'value': value,
            'threshold': threshold
        }

    def _recently_alerted(self, metric_name, cooldown_minutes=60):
        """Check if we recently alerted for this metric."""
        recent = [a for a in self.alert_history 
                  if a['metric'] == metric_name 
                  and (datetime.now() - a['time']).minutes < cooldown_minutes]
        return len(recent) > 0

    def _record_alert(self, metric_name):
        self.alert_history.append({
            'metric': metric_name,
            'time': datetime.now()
        })

# Alert configuration example
alert_config = {
    'data_drift': {
        'threshold': 0.2,  # PSI threshold
        'severity': 'warning',
        'cooldown_minutes': 60
    },
    'accuracy_drop': {
        'threshold': 0.05,  # 5% relative drop
        'severity': 'critical',
        'cooldown_minutes': 30
    },
    'latency_p99': {
        'threshold': 500,  # 500ms
        'severity': 'warning',
        'cooldown_minutes': 15
    }
}

Failure Modes and Debugging Checklist

• [ ] Silent failures: Model serving errors not captured
• [ ] Delayed labels: Can't compute accuracy in real-time
• [ ] Alert fatigue: Too many false alerts
• [ ] Missing baseline: No reference to compare against
• [ ] Upstream changes: Data pipeline modifications
• [ ] Seasonality: Normal variation flagged as drift
• [ ] Sample size: Not enough data for statistical tests

Production Checklist

Before deployment: - [ ] Baseline metrics computed from training/validation - [ ] Reference dataset saved for drift comparison - [ ] Logging infrastructure for inputs/outputs - [ ] Alerting configured with appropriate thresholds - [ ] Dashboards created for visibility

After deployment: - [ ] Drift detection running on schedule - [ ] Performance metrics tracked (when labels available) - [ ] Latency and throughput monitored - [ ] Regular review of alert relevance - [ ] Playbooks updated based on incidents

Automation: - [ ] Automatic retraining triggers - [ ] Fallback to previous model version - [ ] Escalation for critical alerts - [ ] Regular model validation reports

Monitoring Architecture

Production Traffic
       │
       ▼
  ┌─────────┐
  │  Model  │ ──── Predictions ────┐
  │ Service │                      │
  └─────────┘                      ▼
       │                    ┌──────────────┐
       │                    │   Logging    │
   Inputs                   │   Service    │
       │                    └──────────────┘
       ▼                           │
  ┌─────────┐                      ▼
  │ Feature │              ┌──────────────┐
  │  Store  │              │  Monitoring  │
  └─────────┘              │   Pipeline   │
                           └──────────────┘
                                   │
                    ┌──────────────┼──────────────┐
                    ▼              ▼              ▼
              ┌─────────┐   ┌─────────┐   ┌─────────┐
              │  Drift  │   │ Perf    │   │ Alert   │
              │ Detect  │   │ Monitor │   │ Manager │
              └─────────┘   └─────────┘   └─────────┘
                    │              │              │
                    └──────────────┼──────────────┘
                                   ▼
                            ┌─────────────┐
                            │  Dashboard  │
                            └─────────────┘

Tools and Stack

Example Mini-Case

Scenario: Production recommendation model showing declining engagement metrics.

Investigation: 1. Dashboard shows CTR dropped 15% over 2 weeks 2. Drift detection: User feature days_since_signup shifted 3. Root cause: Marketing campaign brought new user cohort 4. Model trained on established users, fails for new users

Response: 1. Alert: CTR dropped below threshold 2. Investigation: Identified drift in user demographics 3. Short-term: Segment-specific fallback (popular items for new users) 4. Long-term: Retrain model with new user cohort data

Prevention: - Added drift monitoring for user cohort features - Configured alert for new user proportion > threshold - Scheduled monthly model refresh

Related Pages

• Index: ML Field Guide
• MLOps Foundations
• A/B Testing for ML
• Anomaly Detection

Further Reading

• Evidently AI Documentation
• ML Monitoring Best Practices
• WhyLabs Documentation
• Google ML Production Guide

Try it live

Everything above runs in your browser — open Model Monitoring Lab — Catch Silent Failures Before Users Do and change the parameters while it is running. Nothing is installed, nothing is uploaded, the whole model lives in one tab.