Model Drift Explained: Why Good Models Become Bad Models

You deployed a model that scored 95% accuracy. Six months later, it scores 70%. No code changed. No one retrained the model. The world simply moved on.

This is model drift — the silent killer of production ML systems.

Model drift occurs when the statistical properties of the target variable or input data change over time, causing the model's predictions to become less accurate.

This article explains the three types of drift, shows Python detection examples, and provides a complete monitoring workflow.

The Three Types of Drift

Data Drift

Data drift occurs when the distribution of input features changes over time. The model receives data it was not trained on.

Types of Data Drift

Detecting Data Drift

import numpy as np
from scipy.stats import ks_2samp, chi2_contingency
import pandas as pd

def detect_data_drift(
    reference_data: pd.DataFrame,
    current_data: pd.DataFrame,
    significance_level: float = 0.05
) -> dict:
    """Detect data drift using statistical tests."""
    
    drift_results = {}
    
    for column in reference_data.columns:
        ref = reference_data[column].dropna()
        cur = current_data[column].dropna()
        
        if pd.api.types.is_numeric_dtype(ref):
            # KS test for numerical features
            stat, p_value = ks_2samp(ref, cur)
            drift_results[column] = {
                'test': 'KS',
                'statistic': stat,
                'p_value': p_value,
                'drift_detected': p_value < significance_level
            }
        else:
            # Chi-square test for categorical features
            ref_counts = ref.value_counts()
            cur_counts = cur.value_counts()
            
            # Align categories
            all_cats = set(ref_counts.index) | set(cur_counts.index)
            ref_aligned = [ref_counts.get(c, 0) for c in all_cats]
            cur_aligned = [cur_counts.get(c, 0) for c in all_cats]
            
            if sum(ref_aligned) > 0 and sum(cur_aligned) > 0:
                chi2, p_value, _, _ = chi2_contingency(
                    [ref_aligned, cur_aligned]
                )
                drift_results[column] = {
                    'test': 'Chi-square',
                    'statistic': chi2,
                    'p_value': p_value,
                    'drift_detected': p_value < significance_level
                }
    
    return drift_results

Concept Drift

Concept drift occurs when the relationship between inputs and targets changes. The model learned patterns that no longer apply.

Types of Concept Drift

Detecting Concept Drift

import numpy as np
from collections import deque

class ConceptDriftDetector:
    def __init__(self, window_size=100, threshold=0.1):
        self.window_size = window_size
        self.threshold = threshold
        self.errors = deque(maxlen=window_size)
        self.reference_error = None
    
    def set_reference(self, errors: list):
        """Set reference error rate from training."""
        self.reference_error = np.mean(errors)
    
    def update(self, true_label, prediction):
        """Add new observation and check for drift."""
        error = 1 if true_label != prediction else 0
        self.errors.append(error)
        
        if len(self.errors) < self.window_size:
            return False
        
        current_error = np.mean(self.errors)
        drift_detected = abs(current_error - self.reference_error) > self.threshold
        
        return drift_detected

# Usage
detector = ConceptDriftDetector(window_size=100, threshold=0.1)
detector.set_reference([0, 0, 0, 0, 0, 1, 0, 0])  # 12.5% error rate

# Simulate production
for true_label, pred in zip(y_true, y_pred):
    if detector.update(true_label, pred):
        print("⚠️ Concept drift detected!")

Prediction Drift

Prediction drift occurs when the distribution of model outputs changes. This can indicate data drift, concept drift, or both.

def detect_prediction_drift(
    reference_predictions: np.ndarray,
    current_predictions: np.ndarray,
    threshold: float = 0.05
) -> dict:
    """Detect drift in model predictions."""
    
    # KS test on prediction distributions
    stat, p_value = ks_2samp(reference_predictions, current_predictions)
    
    # Check mean prediction shift
    ref_mean = np.mean(reference_predictions)
    cur_mean = np.mean(current_predictions)
    mean_shift = abs(cur_mean - ref_mean)
    
    return {
        'ks_statistic': stat,
        'p_value': p_value,
        'drift_detected': p_value < threshold,
        'reference_mean': ref_mean,
        'current_mean': cur_mean,
        'mean_shift': mean_shift
    }

# Example
ref_preds = model.predict(X_reference)
cur_preds = model.predict(X_current)

result = detect_prediction_drift(ref_preds, cur_preds)
if result['drift_detected']:
    print(f"⚠️ Prediction drift detected!")
    print(f"   Mean shifted from {result['reference_mean']:.3f} "
          f"to {result['current_mean']:.3f}")

Complete Monitoring Workflow

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ModelMonitor:
    def __init__(self, model, reference_data, reference_labels):
        self.model = model
        self.reference_data = reference_data
        self.reference_labels = reference_labels
        self.reference_predictions = model.predict(reference_data)
        
        # Store metrics history
        self.metrics_history = []
    
    def log_prediction(self, features, prediction, true_label=None):
        """Log a single prediction for monitoring."""
        
        entry = {
            'timestamp': datetime.now(),
            'prediction': prediction,
            'features': features
        }
        
        if true_label is not None:
            entry['true_label'] = true_label
            entry['error'] = 1 if prediction != true_label else 0
        
        self.metrics_history.append(entry)
    
    def check_data_drift(self, current_batch: pd.DataFrame) -> dict:
        """Check for data drift in current batch."""
        
        results = {}
        for col in self.reference_data.columns:
            stat, p_value = ks_2samp(
                self.reference_data[col].dropna(),
                current_batch[col].dropna()
            )
            results[col] = {
                'p_value': p_value,
                'drift': p_value < 0.05
            }
        
        drifted_features = [k for k, v in results.items() if v['drift']]
        return {
            'drift_detected': len(drifted_features) > 0,
            'drifted_features': drifted_features,
            'details': results
        }
    
    def check_prediction_drift(self) -> dict:
        """Check for prediction drift."""
        
        recent = [m['prediction'] for m in self.metrics_history[-100:]]
        
        stat, p_value = ks_2samp(self.reference_predictions, recent)
        
        return {
            'drift_detected': p_value < 0.05,
            'p_value': p_value,
            'reference_mean': np.mean(self.reference_predictions),
            'current_mean': np.mean(recent)
        }
    
    def get_health_report(self) -> dict:
        """Generate complete health report."""
        
        recent_errors = [
            m['error'] for m in self.metrics_history[-100:] 
            if 'error' in m
        ]
        
        return {
            'timestamp': datetime.now(),
            'total_predictions': len(self.metrics_history),
            'error_rate': np.mean(recent_errors) if recent_errors else None,
            'data_drift': self.check_data_drift(self._get_current_batch()),
            'prediction_drift': self.check_prediction_drift()
        }

When to Retrain

Monitoring Checklist

Try It Yourself

• ML Models Fail in Production — 6 critical failure modes
• Data Leakage in ML — Avoid training data contamination
• Train/Val/Test Sets — Split correctly before training
• RAG Production Guide — Production deployment patterns

Further Reading

• Evidently AI: Concept Drift
• ML Mastery: Detecting Data Drift
• Aerospike: Mitigating Model Drift
• BestWordz: Why ML Models Fail in Production

Conclusion

Model drift is inevitable. The world changes, data evolves, and patterns shift. The question is not whether your model will drift, but when.

The three types of drift — data drift, concept drift, and prediction drift — each require different detection methods:

• Data drift — Monitor input distributions with KS tests or PSI
• Concept drift — Track error rates and prediction accuracy
• Prediction drift — Monitor output distributions for shifts

The solution is continuous monitoring. Build a monitoring system that tracks input distributions, prediction distributions, and performance metrics. Configure alerts for significant drift. Maintain a retraining pipeline that can update models quickly.

Deploying a model is not the finish line. It is the starting line of monitoring.