ml-monitoring

name	ml-monitoring
description	Monitor ML models in production for data drift, concept drift, and performance degradation. Outputs drift detection pipelines, alerting thresholds, retraining triggers, and monitoring dashboards.
argument-hint	["model type","prediction latency requirements","retraining frequency","monitoring tools"]
allowed-tools	Read, Write, Bash

ML Model Monitoring

Models degrade silently. Monitor for data drift, prediction drift, and business metric drift — and trigger automated retraining before users notice the degradation.

Process

1. Define baselines — capture training data statistics and initial model performance.
2. Identify drift types — data drift (input distribution), concept drift (relationship change), prediction drift (output shift).
3. Choose detection methods — statistical tests, distance metrics, windowed comparisons.
4. Set alert thresholds — based on acceptable degradation, not arbitrary numbers.
5. Build monitoring pipeline — scheduled jobs that compute and log drift metrics.
6. Configure dashboards — visualize drift over time alongside business KPIs.
7. Define retraining triggers — automatic vs. manual based on severity.
8. Test monitoring — inject synthetic drift to validate alerts fire correctly.

Output Format

Drift Detection Pipeline

# monitoring/drift_detector.py
import numpy as np
import pandas as pd
from scipy import stats
from dataclasses import dataclass, field
from typing import Optional
import json
import logging
from datetime import datetime, timezone

logger = logging.getLogger(__name__)

@dataclass
class DriftResult:
    feature: str
    drift_type: str           # "data", "concept", "prediction"
    method: str               # "ks_test", "psi", "chi2", "wasserstein"
    statistic: float
    p_value: Optional[float]
    threshold: float
    is_drift: bool
    severity: str             # "none", "warning", "critical"
    timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

class DriftDetector:
    """Statistical drift detection for ML models."""
    
    def __init__(self, reference_data: pd.DataFrame, config: dict = None):
        self.reference = reference_data
        self.config = config or {}
        self._compute_reference_stats()
    
     ():
        
        .reference_stats = {}
         col  .reference.columns:
             pd.api.types.is_numeric_dtype(.reference[col]):
                .reference_stats[col] = {
                    : .reference[col].mean(),
                    : .reference[col].std(),
                    : .reference[col].(),
                    : .reference[col].(),
                    : .reference[col].quantile(),
                    : .reference[col].quantile(),
                    : .reference[col].quantile(),
                    : .reference[col].dropna().values,
                }
            :
                .reference_stats[col] = {
                    : .reference[col].value_counts(normalize=).to_dict(),
                }
    
     () -> DriftResult:
        
        ref_values = .reference_stats[feature][]
        cur_values = current.dropna().values
        
        statistic, p_value = stats.ks_2samp(ref_values, cur_values)
        
        
        threshold = .config.get(, )
        is_drift = statistic > threshold
        severity = 
         statistic > threshold * :
            severity = 
         is_drift:
            severity = 
        
         DriftResult(
            feature=feature,
            drift_type=,
            method=,
            statistic=statistic,
            p_value=p_value,
            threshold=threshold,
            is_drift=is_drift,
            severity=severity,
        )
    
     () -> DriftResult:
        
        ref_values = .reference_stats[feature][]
        cur_values = current.dropna().values
        
        
        _, bin_edges = np.histogram(ref_values, bins=n_bins)
        bin_edges[] = -np.inf
        bin_edges[-] = np.inf
        
        ref_counts, _ = np.histogram(ref_values, bins=bin_edges)
        cur_counts, _ = np.histogram(cur_values, bins=bin_edges)
        
        
        epsilon = 
        ref_pct = (ref_counts + epsilon) / (ref_values)
        cur_pct = (cur_counts + epsilon) / (cur_values)
        
        psi = np.((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))
        
        
        is_drift = psi > 
        severity = 
         psi > :
            severity = 
         is_drift:
            severity = 
        
         DriftResult(
            feature=feature,
            drift_type=,
            method=,
            statistic=psi,
            p_value=,
            threshold=,
            is_drift=is_drift,
            severity=severity,
        )
    
     () -> DriftResult:
        
        ref_dist = .reference_stats[feature][]
        cur_dist = current.value_counts(normalize=).to_dict()
        
        
        all_categories = (ref_dist.keys()) | (cur_dist.keys())
        ref_vals = np.array([ref_dist.get(c, )  c  all_categories])
        cur_vals = np.array([cur_dist.get(c, )  c  all_categories])
        
        
        n = (current)
        expected = ref_vals * n
        observed = cur_vals * n
        
        
        mask = expected > 
        statistic, p_value = stats.chisquare(
            f_obs=observed[mask],
            f_exp=expected[mask]
        )
        
        is_drift = p_value < 
        severity =   p_value <   (  is_drift  )
        
         DriftResult(
            feature=feature,
            drift_type=,
            method=,
            statistic=statistic,
            p_value=p_value,
            threshold=,
            is_drift=is_drift,
            severity=severity,
        )
    
     () -> DriftResult:
        
        
        distance = stats.wasserstein_distance(reference_preds, current_preds)
        
        
        ref_mean = reference_preds.mean()
        cur_mean = current_preds.mean()
        mean_shift = (cur_mean - ref_mean)
        
        threshold = 
        is_drift = distance > threshold  mean_shift > 
        severity =   distance >   (  is_drift  )
        
         DriftResult(
            feature=,
            drift_type=,
            method=,
            statistic=distance,
            p_value=,
            threshold=threshold,
            is_drift=is_drift,
            severity=severity,
        )
    
     () -> :
        
        results = []
        
         col  current_data.columns:
             col   .reference_stats:
                
            
             pd.api.types.is_numeric_dtype(current_data[col]):
                results.append(.detect_psi(col, current_data[col]))
                results.append(.detect_ks(col, current_data[col]))
            :
                results.append(.detect_chi2(col, current_data[col]))
        
         predictions   :
            ref_preds = .reference_stats.get(, {}).get()
             ref_preds   :
                results.append(.detect_prediction_drift(ref_preds, predictions))
        
        
        n_drift = (  r  results  r.is_drift)
        critical = [r  r  results  r.severity == ]
        warnings = [r  r  results  r.severity == ]
        
        overall_status = 
         critical:
            overall_status = 
         warnings:
            overall_status = 
        
         {
            : datetime.now(timezone.utc).isoformat(),
            : overall_status,
            : (results),
            : n_drift,
            : [r.feature  r  critical],
            : [r.feature  r  warnings],
            : [(r)  r  results],
        }

Scheduled Monitoring Job

# monitoring/monitor_job.py
import schedule
import time
import mlflow
import boto3
from monitoring.drift_detector import DriftDetector

class ModelMonitor:
    def __init__(self, model_name: str, model_stage: str = "Production"):
        self.model_name = model_name
        self.model_stage = model_stage
        self.s3 = boto3.client("s3")
        
        # Load reference data and model
        self.reference_data = self._load_reference_data()
        self.model = self._load_production_model()
        self.detector = DriftDetector(self.reference_data)
    
    def _load_production_model(self):
        client = mlflow.tracking.MlflowClient()
        versions = client.get_latest_versions(self.model_name, stages=[self.model_stage])
        if not versions:
            raise ValueError(f"No {self.model_stage} model found for {self.model_name}")
        
        model_uri = f"models:/{self.model_name}/{self.model_stage}"
        return mlflow.pyfunc.load_model(model_uri)
    
    def _load_reference_data() -> pd.DataFrame:
        
         pd.read_parquet()
    
     ():
        logger.info()
        
        
        current_data = ._load_recent_predictions(hours=)
         (current_data) < :
            logger.warning()
            
        
        features = current_data.drop(columns=[, , ])
        predictions = current_data[].values
        
        
        report = .detector.run_full_report(features, predictions)
        
        
         mlflow.start_run(run_name=,
                              tags={: }):
            mlflow.log_metrics({
                : r[]
                 r  report[]
            })
            mlflow.log_dict(report, )
        
        
         report[] == :
            ._send_alert(report, severity=)
            ._trigger_retraining(report)
         report[] == :
            ._send_alert(report, severity=)
        
        logger.info()
    
     ():
         boto3
        sns = boto3.client()
        
        message = (
            
            
            
            
        )
        
        sns.publish(
            TopicArn=os.environ[],
            Subject=,
            Message=message,
            MessageAttributes={
                : {: , : severity}
            }
        )
    
     ():
        
         requests
        requests.post(
            os.environ[],
            json={
                : .model_name,
                : ,
                : report,
            },
            headers={: }
        )
    
     () -> pd.DataFrame:
        
        end = datetime.now(timezone.utc)
        start = end - timedelta(hours=hours)
         pd.read_parquet(
            ,
            filters=[
                (, , start),
                (, , end)
            ]
        )



monitor = ModelMonitor()

schedule.every().hours.do(monitor.run_monitoring_check)
schedule.every().day.at().do(monitor.run_monitoring_check)  

 :
    schedule.run_pending()
    time.sleep()

Performance Monitoring (when labels available)

# monitoring/performance_tracker.py
from sklearn.metrics import roc_auc_score, f1_score, precision_score, recall_score

class PerformanceMonitor:
    """Track model accuracy metrics when ground truth is available."""
    
    def __init__(self, model_name: str, baseline_metrics: dict):
        self.model_name = model_name
        self.baseline = baseline_metrics
        self.degradation_threshold = 0.05  # 5% relative degradation
    
    def evaluate_window(
        self,
        predictions: np.ndarray,
        labels: np.ndarray,
        window_start: datetime,
        window_end: datetime
    ) -> dict:
        
        current_metrics = {
            "auc": roc_auc_score(labels, predictions),
            "f1": f1_score(labels, (predictions > 0.5).astype(int)),
            "precision": precision_score(labels, (predictions > 0.5).astype(int)),
            "recall": recall_score(labels, (predictions > 0.5).astype(int)),
        }
        
        # Compute degradation vs. baseline
        degradation = {
            metric: (self.baseline[metric] - current_metrics[metric]) / self.baseline[metric]
            for metric in current_metrics
        }
        
        alerts = []
        for metric, pct_drop in degradation.items():
            if pct_drop > .degradation_threshold:
                alerts.append({
                    : metric,
                    : .baseline[metric],
                    : current_metrics[metric],
                    : pct_drop * ,
                })
        
         {
            : window_start.isoformat(),
            : window_end.isoformat(),
            : (labels),
            : current_metrics,
            : degradation,
            : alerts,
            :   alerts  ,
        }

Grafana Dashboard Config

{
  "title": "ML Model Monitoring — Order Propensity",
  "panels": [
    {
      "title": "PSI by Feature (1h rolling)",
      "type": "timeseries",
      "targets": [{
        "expr": "ml_feature_psi{model=\"order-propensity\"}",
        "legendFormat": "{{feature}}"
      }],
      "thresholds": [
        {"value": 0.1, "color": "yellow"},
        {"value": 0.2, "color": "red"}
      ]
    },
    {
       
       
       
         
      
    
    
       
       
       
         
         
      
       
           
           
           
      
    
  

Rules

• Monitor from day one — don't wait for a production incident to discover drift.
• Log predictions with features — you need both to diagnose drift later.
• Separate data drift from concept drift — same input distribution but wrong predictions = concept drift.
• Set thresholds from business impact — a 5% AUC drop may not matter; 20% may be catastrophic.
• Monitor prediction distribution, not just inputs — output shift is the first sign of trouble.
• Label delay is real — for models predicting future events, ground truth arrives days or weeks later.
• Track feature importance drift — a feature becoming more/less important signals concept drift.
• Alert on missing features — if a feature that was always present goes missing, something is broken upstream.
• Retrain on a schedule even without drift — scheduled retraining keeps models fresh even when drift is gradual.
• Test your monitoring — inject synthetic drift in staging to confirm alerts work.