Data Drift Detection — Is My Model Still Valid?
Models degrade silently in production because the real world changes: customer demographics shift, economic conditions change, seasonal patterns evolve. Data drift (input features changing) and concept drift (relationship between features and target changing) cause model performance to drop without any code change. Drift detection triggers retraining alerts before performance degrades significantly.
Statistical Drift Detection with PSI and KS Tests
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.stats import ks_2samp, chi2_contingency
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import roc_auc_score
np.random.seed(42)
# TRAINING DISTRIBUTION (reference)
N_train = 5000
df_train = pd.DataFrame({
"age": np.random.normal(38, 12, N_train).clip(18, 75),
"income": np.random.exponential(55000, N_train).clip(15000, 200000),
"credit": np.random.normal(680, 80, N_train).clip(300, 850),
})
# PRODUCTION DATA (month 1: minor drift, month 6: major drift)
N_prod = 1000
df_prod_m1 = pd.DataFrame({
"age": np.random.normal(40, 13, N_prod).clip(18, 75), # slight drift
"income": np.random.exponential(55000, N_prod).clip(15000, 200000),
"credit": np.random.normal(680, 80, N_prod).clip(300, 850),
})
df_prod_m6 = pd.DataFrame({
"age": np.random.normal(30, 10, N_prod).clip(18, 75), # major drift (younger)
"income": np.random.exponential(40000, N_prod).clip(15000, 200000), # lower income
"credit": np.random.normal(620, 90, N_prod).clip(300, 850), # lower credit
})
# ━━━━━━━━━━━━━━━━━━━━━━━━━━
# METHOD 1: KS TEST (two-sample)
# ━━━━━━━━━━━━━━━━━━━━━━━━━━
def ks_drift_report(ref: pd.DataFrame, prod: pd.DataFrame, threshold: float = 0.05) -> pd.DataFrame:
results = []
for col in ref.columns:
stat, pval = ks_2samp(ref[col], prod[col])
drifted = pval < threshold
results.append({"Feature": col, "KS_Stat": stat, "P_Value": pval, "Drifted": drifted})
return pd.DataFrame(results)
print("KS Drift Test -- Month 1 (minor drift):")
print(ks_drift_report(df_train, df_prod_m1).round(4).to_string(index=False))
print("\nKS Drift Test -- Month 6 (major drift):")
print(ks_drift_report(df_train, df_prod_m6).round(4).to_string(index=False))
# ━━━━━━━━━━━━━━━━━━━━━━━━━━
# METHOD 2: PSI (Population Stability Index)
# PSI < 0.1: stable | 0.1-0.2: slight drift | > 0.2: significant drift
# ━━━━━━━━━━━━━━━━━━━━━━━━━━
def compute_psi(reference: np.ndarray, production: np.ndarray, n_bins: int = 10) -> float:
bins = np.percentile(reference, np.linspace(0, 100, n_bins + 1))
bins[0] -= 1e-8; bins[-1] += 1e-8 # ensure all values captured
ref_pct = np.histogram(reference, bins=bins)[0] / len(reference)
prod_pct = np.histogram(production, bins=bins)[0] / len(production)
# Avoid log(0)
ref_pct = np.clip(ref_pct, 1e-6, None)
prod_pct = np.clip(prod_pct, 1e-6, None)
psi = np.sum((prod_pct - ref_pct) * np.log(prod_pct / ref_pct))
return psi
print("\nPSI Scores:")
for col in df_train.columns:
psi_m1 = compute_psi(df_train[col].values, df_prod_m1[col].values)
psi_m6 = compute_psi(df_train[col].values, df_prod_m6[col].values)
status_m1 = "STABLE" if psi_m1 < 0.1 else ("SLIGHT" if psi_m1 < 0.25 else "DRIFT!")
status_m6 = "STABLE" if psi_m6 < 0.1 else ("SLIGHT" if psi_m6 < 0.25 else "DRIFT!")
print(f" {col:10s}: Month 1 PSI={psi_m1:.3f} ({status_m1:7s}) | Month 6 PSI={psi_m6:.3f} ({status_m6})")
# ━━━━━━━━━━━━━━━━━━━━━━━━━━
# DRIFT RESPONSE FRAMEWORK
# ━━━━━━━━━━━━━━━━━━━━━━━━━━
response_plan = {
"Monitor": "PSI < 0.10 or KS p > 0.10: no action. Log and watch.",
"Alert": "PSI 0.10-0.25 or KS p < 0.10: investigate, trigger shadow evaluation",
"Retrain": "PSI > 0.25 or model AUC drops > 3%: retrain on recent data + validation",
"Emergency": "AUC drops > 10% or data pipeline failure: roll back to previous version",
}
print("\nDrift Response Framework:")
for severity, action in response_plan.items():
print(f" {severity:12s}: {action}")Tip
Tip
Practice Data Drift Detection Is My Model Still Valid in small, isolated examples before integrating into larger projects. Breaking concepts into small experiments builds genuine understanding faster than reading alone.
80% of ML work is data preparation — garbage in = garbage out
Practice Task
Note
Practice Task — (1) Write a working example of Data Drift Detection Is My Model Still Valid from scratch without looking at notes. (2) Modify it to handle an edge case (empty input, null value, or error state). (3) Share your solution in the Priygop community for feedback.
Quick Quiz
Common Mistake
Warning
A common mistake with Data Drift Detection Is My Model Still Valid is skipping edge case testing — empty inputs, null values, and unexpected data types. Always validate boundary conditions to write robust, production-ready ml code.