Expert patterns for data drift detection, concept drift, performance estimation without labels, and alert engineering.
Pattern 1: PSI Feature Drift Monitor
Population Stability Index for continuous and categorical features.
import numpy as np
import pandas as pd
from scipy.stats import ks_2samp, wasserstein_distance
def compute_psi(reference: np.ndarray, current: np.ndarray, n_bins: int = 10, epsilon: float = 1e-6) -> float: """ PSI < 0.1: stable | 0.1-0.25: investigate | > 0.25: significant shift """ # Create bins from reference distribution breakpoints = np.linspace(0, 100, n_bins + 1) ref_percentiles = np.percentile(reference, breakpoints) ref_percentiles = np.unique(ref_percentiles) # remove duplicates
ref_counts, _ = np.histogram(reference, bins=ref_percentiles)
cur_counts, _ = np.histogram(current, bins=ref_percentiles)
ref_pct = ref_counts / len(reference) + epsilon
cur_pct = cur_counts / len(current) + epsilon
psi = np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))
return round(psi, 4)
def drift_report(reference_df: pd.DataFrame, current_df: pd.DataFrame, numeric_cols: list, categorical_cols: list) -> pd.DataFrame: """Full drift report across all features.""" records = [] for col in numeric_cols: ref = reference_df[col].dropna().values cur = current_df[col].dropna().values psi = compute_psi(ref, cur) ks_stat, ks_p = ks_2samp(ref, cur) w_dist = wasserstein_distance(ref, cur) severity = "stable" if psi < 0.1 else ("investigate" if psi < 0.25 else "significant") records.append({"feature": col, "type": "numeric", "psi": psi, "ks_stat": round(ks_stat, 4), "ks_p": round(ks_p, 4), "wasserstein": round(w_dist, 4), "severity": severity})
for col in categorical_cols:
ref_dist = reference_df[col].value_counts(normalize=True)
cur_dist = current_df[col].value_counts(normalize=True)
all_cats = set(ref_dist.index) | set(cur_dist.index)
ref_arr = np.array([ref_dist.get(c, 1e-6) for c in all_cats])
cur_arr = np.array([cur_dist.get(c, 1e-6) for c in all_cats])
psi = float(np.sum((cur_arr - ref_arr) * np.log(cur_arr / ref_arr)))
severity = "stable" if psi < 0.1 else ("investigate" if psi < 0.25 else "significant")
records.append({"feature": col, "type": "categorical",
"psi": round(psi, 4), "ks_stat": None,
"ks_p": None, "wasserstein": None, "severity": severity})
return pd.DataFrame(records).sort_values("psi", ascending=False)
report = drift_report(reference_df, current_df, numeric_cols=["age", "income", "score"], categorical_cols=["country", "device_type"]) print(report[report["severity"] != "stable"].to_string(index=False))
## Pattern 2: Evidently Drift Report and TestSuite
Automated drift gate with Evidently.
```python
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset
from evidently.test_suite import TestSuite
from evidently.tests import TestNumberOfDriftedColumns, TestShareOfDriftedColumns
from evidently import ColumnMapping
column_mapping = ColumnMapping(
target="label",
prediction="predicted_label",
numerical_features=["age", "income", "tenure_days"],
categorical_features=["country", "device_type"],
)
# Visual report (HTML)
report = Report(metrics=[DataDriftPreset(), TargetDriftPreset()])
report.run(reference_data=reference_df, current_data=current_df,
column_mapping=column_mapping)
report.save_html("drift_report.html")
# Automated test gate (CI-friendly)
suite = TestSuite(tests=[
TestNumberOfDriftedColumns(lt=3), # fail if > 2 features drift
TestShareOfDriftedColumns(lt=0.3), # fail if > 30% features drift
])
suite.run(reference_data=reference_df, current_data=current_df,
column_mapping=column_mapping)
result = suite.as_dict()
if not result["summary"]["all_passed"]:
failed = [t for t in result["tests"] if t["status"] == "FAIL"]
for t in failed:
print(f"FAIL: {t['name']} — {t['description']}")
raise SystemExit(1)
Pattern 3: NannyML Performance Estimation Without Labels
Estimate model performance when ground truth labels are delayed.
import nannyml as nml
import pandas as pd
# CBPE: Confidence-Based Performance Estimation
# Requires calibrated predicted probabilities
estimator = nml.CBPE(
y_pred_proba="y_pred_proba",
y_pred="y_pred",
y_true="label",
timestamp_column_name="timestamp",
metrics=["roc_auc", "f1", "precision", "recall"],
chunk_size=500, # events per monitoring window
problem_type="binary_classification",
)
# Fit on reference (labeled) data
estimator.fit(reference_df)
# Estimate on production data (no labels needed)
estimated_performance = estimator.estimate(production_df)
fig = estimated_performance.plot()
fig.savefig("estimated_performance.png")
# Check for estimated degradation
results_df = estimated_performance.to_df()
latest = results_df.iloc[-1]
if latest["estimated_roc_auc"] < 0.75:
print(f"ALERT: Estimated AUC dropped to {latest['estimated_roc_auc']:.4f}")
# Univariate feature drift
univariate_drift = nml.UnivariateDriftCalculator(
column_names=["age", "income", "country"],
timestamp_column_name="timestamp",
chunk_size=500,
continuous_methods=["kolmogorov_smirnov", "wasserstein"],
categorical_methods=["chi2", "jensen_shannon"],
)
univariate_drift.fit(reference_df)
drift_results = univariate_drift.calculate(production_df)
drift_results.filter(period="analysis", alerts=True).to_df()
Pattern 4: whylogs Distribution Profiles
Lightweight sketches for continuous monitoring with WhyLabs.
import whylogs as why
from whylogs.core.constraints import ConstraintsBuilder
from whylogs.core.constraints.factories import (
no_missing_values,
is_in_range,
)
# Profile reference data
reference_profile = why.log(reference_df).profile()
reference_profile.set_dataset_timestamp(reference_start)
# Profile production batch
production_profile = why.log(production_df).profile()
# Extract summary stats
ref_view = reference_profile.view()
cur_view = production_profile.view()
# Constraints: validate schema + ranges
builder = ConstraintsBuilder(dataset_profile_view=cur_view)
builder.add_constraint(no_missing_values(column_name="user_id"))
builder.add_constraint(is_in_range(column_name="age", lower=18, upper=120))
constraints = builder.build()
valid = constraints.validate()
for name, result in constraints.report():
if not result:
print(f"CONSTRAINT FAIL: {name}")
# Upload to WhyLabs for trend tracking
import whylogs.api.writer.whylabs as whylabs_writer
writer = whylabs_writer.WhyLabsWriter()
writer.write(file=production_profile)
Pattern 5: Prometheus Metrics for Model Serving
Instrument prediction service for Grafana dashboards.
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import numpy as np
# Metrics
prediction_counter = Counter("model_predictions_total",
"Total prediction requests",
["model_version", "outcome"])
prediction_latency = Histogram("model_prediction_latency_seconds",
"Prediction latency",
buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0])
score_histogram = Histogram("model_prediction_score",
"Distribution of predicted probabilities",
buckets=[i/10 for i in range(11)])
drift_gauge = Gauge("feature_psi", "PSI score per feature", ["feature"])
# In serving code
start_http_server(8001) # Prometheus scrapes :8001/metrics
def predict_with_instrumentation(features: dict, model_version: str):
import time
start = time.time()
score = model.predict_proba(features)[1]
outcome = "positive" if score > 0.5 else "negative"
prediction_counter.labels(model_version=model_version, outcome=outcome).inc()
prediction_latency.observe(time.time() - start)
score_histogram.observe(score)
return score
# Background job: compute and push PSI per feature
def update_drift_gauges(reference_df, current_window_df, numeric_cols):
for col in numeric_cols:
psi = compute_psi(reference_df[col].values, current_window_df[col].values)
drift_gauge.labels(feature=col).set(psi)
Anti-Patterns
Anti-Pattern 1: Alerting on Every Metric Individually
Monitoring 50 features × 3 metrics = 150 time series. Individual alerts per metric at 0.05 significance → expect 7+ false alarms per monitoring window by chance alone. Use composite drift score or require multiple features drifting before alerting.
Anti-Pattern 2: Using Training Data as Reference Forever
Training data from 18 months ago may not represent current production distribution even when the model is working correctly (product changes, user base expansion). Periodically re-anchor reference windows or use stratified production samples from stable periods.
Anti-Pattern 3: Monitoring Only Aggregate Distributions
Overall feature distribution can be stable while a specific slice (mobile users, a new country) drifts severely. Always monitor per-segment distributions for high-traffic segments.
Anti-Pattern 4: Confusing Data Drift with Concept Drift
A feature distribution shift (data drift) does not necessarily cause model degradation. A model can handle new distribution ranges correctly. Concept drift (P(Y|X) shift) does cause degradation. Distinguish before triggering retraining.