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.

```python
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)

        # Calculate PSI formula
        psi_value = np.sum((cur_props - ref_props) * np.log(cur_props / ref_props))
        return float(psi_value)

    def evaluate_numerical_feature(self, feature_name: str, current_data: pd.DataFrame) -> dict:
        ref_col = self.reference_data[feature_name].dropna().values
        cur_col = current_data[feature_name].dropna().values

        # Kolmogorov-Smirnov Test
        ks_stat, p_value = ks_2samp(ref_col, cur_col)
        drift_detected_ks = p_value < self.alpha

        # Population Stability Index
        psi_val = self.calculate_psi(ref_col, cur_col)
        drift_detected_psi = psi_val > self.psi_threshold

        return {
            "feature": feature_name,
            "type": "numerical",
            "ks_statistic": float(ks_stat),
            "p_value": float(p_value),
            "psi_value": psi_val,
            "drift_detected": drift_detected_ks or drift_detected_psi
        }

    def evaluate_categorical_feature(self, feature_name: str, current_data: pd.DataFrame) -> dict:
        ref_counts = self.reference_data[feature_name].value_counts()
        cur_counts = current_data[feature_name].value_counts()

        # Combine counts into contingency table
        df_contingency = pd.DataFrame({'ref': ref_counts, 'cur': cur_counts}).fillna(0)
        chi2_stat, p_value, _, _ = chi2_contingency(df_contingency.T)

        return {
            "feature": feature_name,
            "type": "categorical",
            "chi2_statistic": float(chi2_stat),
            "p_value": float(p_value),
            "drift_detected": p_value < self.alpha
        }

    def run_drift_analysis(self, current_data: pd.DataFrame) -> dict:
        results = []
        drifted_count = 0

        for col in self.reference_data.columns:
            if np.issubdtype(self.reference_data[col].dtype, np.number):
                res = self.evaluate_numerical_feature(col, current_data)
            else:
                res = self.evaluate_categorical_feature(col, current_data)
            
            if res["drift_detected"]:
                drifted_count += 1
            results.append(res)

        dataset_drift_ratio = drifted_count / len(self.reference_data.columns)

        return {
            "dataset_drift_ratio": dataset_drift_ratio,
            "dataset_drift_detected": dataset_drift_ratio >= 0.3,
            "feature_metrics": results
        }
```

---

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

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

```python
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 = abs((val - mean) / std)
            
            # Export Z-Score as online proxy for drift
            FEATURE_DRIFT_GAUGE.labels(feature_name=feature_name).set(z_score)

    # Perform prediction logic (mocked)
    score = np.random.random()

    return {"credit_decision": "approved" if score > 0.5 else "rejected", "score": score}

# Expose /metrics endpoint for Prometheus scrapers
metrics_app = make_asgi_app()
app.mount("/metrics", metrics_app)
```

---

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

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

```python
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

```python
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).