Distribution Shift — When AI Models Fail in the Real World

import numpy as np
from scipy import stats
from sklearn.preprocessing import StandardScaler
import pandas as pd

# ── Covariate shift detection using statistical tests ──
def detect_covariate_shift(X_train: np.ndarray, X_production: np.ndarray,
                             feature_names: list, alpha: float = 0.05) -> dict:
    """
    Test each feature for distribution shift between training and production.
    Uses Kolmogorov-Smirnov test for continuous features.
    """
    results = {}
    for i, feature in enumerate(feature_names):
        train_vals = X_train[:, i]
        prod_vals  = X_production[:, i]

        # KS test: H0 = same distribution, reject if p < alpha
        ks_stat, p_value = stats.ks_2samp(train_vals, prod_vals)

        results[feature] = {
            'ks_statistic': round(ks_stat, 4),
            'p_value':      round(p_value, 4),
            'shift_detected': p_value < alpha,
            'train_mean':   round(train_vals.mean(), 3),
            'prod_mean':    round(prod_vals.mean(), 3),
            'mean_change':  round(prod_vals.mean() - train_vals.mean(), 3),
        }
    return results

# Simulate distribution shift (e.g., age distribution changed in production)
np.random.seed(42)
n_train, n_prod = 10000, 1000
feature_names = ['age', 'income', 'credit_score', 'debt_ratio']

X_train = np.column_stack([
    np.random.normal(35, 10, n_train),      # age: mean 35 in training
    np.random.normal(60000, 20000, n_train),
    np.random.normal(680, 80, n_train),
    np.random.normal(0.3, 0.1, n_train),
])

X_prod = np.column_stack([
    np.random.normal(50, 12, n_prod),       # age: mean 50 in production (SHIFT!)
    np.random.normal(62000, 22000, n_prod), # income: similar
    np.random.normal(660, 90, n_prod),      # credit: slight shift
    np.random.normal(0.28, 0.09, n_prod),   # debt: similar
])

shift_results = detect_covariate_shift(X_train, X_prod, feature_names)
print("Distribution Shift Detection Results:")
print(f"{'Feature':<15} {'KS Stat':<10} {'P-value':<10} {'Shift?':<8} {'Mean Change'}")
for feat, r in shift_results.items():
    status = '⚠️ YES' if r['shift_detected'] else '✓ No'
    print(f"{feat:<15} {r['ks_statistic']:<10} {r['p_value']:<10} {status:<8} {r['mean_change']:+.3f}")

# ── Concept drift: monitor model performance over time ──
class ProductionMonitor:
    """Monitor model performance in production for drift."""

    def __init__(self, model, window_size=100, alert_threshold=0.1):
        self.model          = model
        self.window_size    = window_size
        self.alert_threshold = alert_threshold
        self.baseline_accuracy = None
        self.accuracy_history  = []

    def set_baseline(self, X_test, y_test):
        preds = self.model.predict(X_test)
        self.baseline_accuracy = (preds == y_test).mean()
        print(f"Baseline accuracy: {self.baseline_accuracy:.3f}")

    def update_and_check(self, X_new, y_new_true) -> bool:
        preds = self.model.predict(X_new)
        window_accuracy = (preds == y_new_true).mean()
        self.accuracy_history.append(window_accuracy)

        # Alert if accuracy drops by more than threshold from baseline
        if window_accuracy < self.baseline_accuracy - self.alert_threshold:
            print(f"⚠️ DRIFT ALERT: accuracy dropped from {self.baseline_accuracy:.3f} "
                  f"to {window_accuracy:.3f}")
            return True
        return False