model-monitoring-and-drift

name	model-monitoring-and-drift
metadata	{"category":"Machine Learning Operations (MLOps)"}
description	Implement automated machine learning model performance monitoring, statistical data drift (KS-test, PSI, Wasserstein distance), concept drift detection, and prediction quality evaluation in production ML systems using Evidently AI, NannyML, SciPy, and Prometheus. Triggers when establishing model observability, real-time feature drift middleware, automated HTML drift reporting, or retraining trigger alerts.
compatibility	Python (>= 3.9), Evidently AI (>= 0.4.0), SciPy, Prometheus Client, FastAPI / Flask

Model Monitoring & Drift Detection

Production patterns for measuring Data Drift, Concept Drift, and Performance Decay in live machine learning applications.

---

1. Drift Monitoring Conceptual Framework

Drift Category	Definition	Statistical Metric	Remediation Strategy
Data Drift (Covariate Shift)	Distribution of input features $P(X)$ changes over time.	Kolmogorov-Smirnov Test, Population Stability Index (PSI), Wasserstein Distance	Feature re-normalization, Retraining on recent data window
Concept Drift	Relationship between input features and target $P(Y \mid X)$ changes.	Cramér's V, Wasserstein Distance, Accuracy/F1 degradation	Architecture update, Label re-annotation, Retraining
Prior Probability Shift	Target label distribution $P(Y)$ changes.	Chi-Square Goodness-of-Fit, Jensen-Shannon Divergence	Decision threshold re-calibration

---

2. Statistical Data Drift Engine (drift_engine.py)

A pure Python + SciPy statistical drift evaluator for batch production datasets.

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

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

class StatisticalDriftDetector:
    def __init__(self, reference_data: pd.DataFrame, alpha: float = 0.05, psi_threshold: float = 0.2):
        self.reference_data = reference_data
        self.alpha = alpha
        self.psi_threshold = psi_threshold

    def calculate_psi(self, reference: np.ndarray, current: np.ndarray, num_buckets: int = 10) -> float:
        """Calculates Population Stability Index (PSI) for continuous variables."""
        # Define bucket boundaries using reference quantiles
        percentiles = np.linspace(0, 100, num_buckets + 1)
        buckets = np.percentile(reference, percentiles)
        buckets[0] = -np.inf
        buckets[-1] = np.inf

        # Calculate counts per bucket
        ref_counts, _ = np.histogram(reference, bins=buckets)
        cur_counts, _ = np.histogram(current, bins=buckets)

        # Convert to proportions with zero-handling smoothing
        ref_props = np.where(ref_counts == 0, 1e-4, ref_counts) / len(reference)
        cur_props = np.where(cur_counts == 0, 1e-4, cur_counts) / len(current)

        
        psi_value = np.((cur_props - ref_props) * np.log(cur_props / ref_props))
         (psi_value)

     () -> :
        ref_col = .reference_data[feature_name].dropna().values
        cur_col = current_data[feature_name].dropna().values

        
        ks_stat, p_value = ks_2samp(ref_col, cur_col)
        drift_detected_ks = p_value < .alpha

        
        psi_val = .calculate_psi(ref_col, cur_col)
        drift_detected_psi = psi_val > .psi_threshold

         {
            : feature_name,
            : ,
            : (ks_stat),
            : (p_value),
            : psi_val,
            : drift_detected_ks  drift_detected_psi
        }

     () -> :
        ref_counts = .reference_data[feature_name].value_counts()
        cur_counts = current_data[feature_name].value_counts()

        
        df_contingency = pd.DataFrame({: ref_counts, : cur_counts}).fillna()
        chi2_stat, p_value, _, _ = chi2_contingency(df_contingency.T)

         {
            : feature_name,
            : ,
            : (chi2_stat),
            : (p_value),
            : p_value < .alpha
        }

     () -> :
        results = []
        drifted_count = 

         col  .reference_data.columns:
             np.issubdtype(.reference_data[col].dtype, np.number):
                res = .evaluate_numerical_feature(col, current_data)
            :
                res = .evaluate_categorical_feature(col, current_data)
            
             res[]:
                drifted_count += 
            results.append(res)

        dataset_drift_ratio = drifted_count / (.reference_data.columns)

         {
            : dataset_drift_ratio,
            : dataset_drift_ratio >= ,
            : results
        }

---

3. Real-Time Inference Monitoring Middleware (FastAPI + Prometheus Exporter)

Export real-time feature drift statistics directly to Prometheus from your inference endpoints.

from fastapi import FastAPI, Request
from prometheus_client import Gauge, Counter, make_asgi_app
import numpy as np
import time

app = FastAPI(title="ML Model Serving with Drift Exporter")

# Prometheus Metrics Definition
PREDICTION_COUNT = Counter("model_predictions_total", "Total inference requests executed")
FEATURE_DRIFT_GAUGE = Gauge("model_feature_psi", "Population Stability Index per Feature", ["feature_name"])
DATASET_DRIFT_STATUS = Gauge("model_dataset_drift_detected", "1 if dataset level drift detected else 0")

# Reference baseline mean and std for online z-score check
BASELINE_STATS = {
    "income": {"mean": 65000.0, "std": 15000.0},
    "credit_score": {"mean": 710.0, "std": 45.0}
}

@app.post("/predict")
async def predict_endpoint(payload: dict):
    start_time = time.time()
    PREDICTION_COUNT.inc()

    features = payload.get("features", {})
    
    # Online Z-score Outlier & Drift Metric Export
    for feature_name, val in features.items():
        if feature_name in BASELINE_STATS:
            mean = BASELINE_STATS[feature_name]["mean"]
            std = BASELINE_STATS[feature_name]["std"]
            z_score = ((val - mean) / std)
            
            
            FEATURE_DRIFT_GAUGE.labels(feature_name=feature_name).(z_score)

    
    score = np.random.random()

     {:   score >   , : score}


metrics_app = make_asgi_app()
app.mount(, metrics_app)

---

4. Evidently AI Automated HTML Drift Reporting (generate_report.py)

Generate offline interactive HTML drift and data quality reports using Evidently AI.

import pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset, DataQualityPreset

def generate_evidently_drift_report(reference_df: pd.DataFrame, current_df: pd.DataFrame, output_filepath: str):
    """Generates an HTML Data & Target Drift Report."""
    report = Report(metrics=[
        DataDriftPreset(),
        DataQualityPreset(),
        TargetDriftPreset()
    ])

    report.run(reference_data=reference_df, current_data=current_df)
    report.save_html(output_filepath)
    print(f"Drift report generated successfully at: {output_filepath}")

if __name__ == "__main__":
    ref = pd.DataFrame({"income": [50000, 60000, 70000, 80000], "target": [0, 1, 0, 1]})
    cur = pd.DataFrame({"income": [90000, 120000, 110000, 95000], "target": [1, 1, 1, 1]})
    generate_evidently_drift_report(ref, cur, "drift_report.html")

---

5. Retraining Trigger & Alerting Integration

import requests
import json

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/YOUR_WORKSPACE/YOUR_CHANNEL/YOUR_WEBHOOK_TOKEN"

def send_drift_alert(drift_results: dict):
    if drift_results["dataset_drift_detected"]:
        payload = {
            "text": ":warning: *ML Model Drift Alert*",
            "attachments": [
                {
                    "color": "#FF0000",
                    "fields": [
                        {"title": "Dataset Drift Ratio", "value": f"{drift_results['dataset_drift_ratio']:.2%}", "short": True},
                        {"title": "Status", "value": "CRITICAL - Retraining Triggered", "short": True}
                    ]
                }
            ]
        }
        requests.post(SLACK_WEBHOOK_URL, data=json.dumps(payload), headers={'Content-Type': 'application/json'})

        # Trigger Retraining Airflow DAG or GitHub Actions Workflow
        requests.post(
            "https://airflow.prod.internal/api/v1/dags/retrain_churn_model/dagRuns",
            auth=('admin', 'admin_password'),
            json={"conf": {"reason": "data_drift_exceeded_threshold"}}
        )

---

6. Best Practices

1. Reference Dataset Hygiene: Freeze and version reference datasets matching exact training validation splits; update references only when a newly retrained model is deployed to production.
2. Sampling Windows: Select appropriate drift calculation windows (e.g., 7-day rolling window for high-volume endpoints) to prevent false alerts caused by transient weekend noise.
3. Alerting Thresholds: Set PSI warning thresholds at 0.1 <= PSI < 0.2 (slight shift) and critical alert thresholds at PSI >= 0.2 (significant shift requiring action).