/model-monitoring

name: Model Monitoring
description: Monitor model performance, detect data drift, concept drift, and anomalies in production using Prometheus, Grafana, and MLflow

Model Monitoring

Overview

Monitoring deployed machine learning models ensures they continue to perform well in production, detecting data drift, concept drift, and performance degradation.

When to Use

• When models are deployed in production environments serving real users
• When detecting data drift or concept drift in input features
• When tracking model performance metrics over time
• When ensuring model reliability, accuracy, and operational health
• When implementing ML observability and alerting systems
• When establishing thresholds for model retraining or intervention

Monitoring Components

• **Performance Metrics**: Accuracy, latency, throughput
• **Data Drift**: Distribution changes in input features
• **Concept Drift**: Changes in target variable relationships
• **Output Drift**: Changes in prediction distribution
• **Feature Drift**: Individual feature distribution changes
• **Anomaly Detection**: Unusual samples in production

Monitoring Tools

• **Prometheus**: Metrics collection and storage
• **Grafana**: Visualization and dashboarding
• **MLflow**: Model tracking and registry
• **TensorFlow Data Validation**: Data statistics
• **Evidently**: Drift detection and monitoring
• **Great Expectations**: Data quality assertions

Python Implementation

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
from scipy import stats
import json
from datetime import datetime, timedelta
import logging

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

print("=== 1. Production Monitoring System ===")

class ModelMonitoringSystem:
    def __init__(self, model, scaler, baseline_data, baseline_targets):
        self.model = model
        self.scaler = scaler
        self.baseline_data = baseline_data
        self.baseline_targets = baseline_targets
        self.baseline_mean = baseline_data.mean(axis=0)
        self.baseline_std = baseline_data.std(axis=0)
        self.baseline_predictions = model.predict(baseline_data)

        self.metrics_history = []
        self.drift_alerts = []
        self.performance_history = []

    def log_predictions(self, X, y_true, y_pred):
        """Log predictions and compute metrics"""
        timestamp = datetime.now()

        # Compute metrics
        accuracy = accuracy_score(y_true, y_pred)
        precision = precision_score(y_true, y_pred, average='weighted', zero_division=0)
        recall = recall_score(y_true, y_pred, average='weighted', zero_division=0)
        f1 = f1_score(y_true, y_pred, average='weighted', zero_division=0)

        metric_record = {
            'timestamp': timestamp,
            'accuracy': accuracy,
            'precision': precision,
            'recall': recall,
            'f1': f1,
            'n_samples': len(X)
        }

        self.metrics_history.append(metric_record)
        return metric_record

    def detect_data_drift(self, X_new):
        """Detect data drift using Kolmogorov-Smirnov test"""
        drift_detected = False
        drift_features = []

        for feature_idx in range(X_new.shape[1]):
            baseline_feature = self.baseline_data[:, feature_idx]
            new_feature = X_new[:, feature_idx]

            # KS Test
            ks_statistic, p_value = stats.ks_2samp(baseline_feature, new_feature)

            if p_value < 0.05:  # Significant drift detected
                drift_detected = True
                drift_features.append({
                    'feature_index': feature_idx,
                    'ks_statistic': float(ks_statistic),
                    'p_value': float(p_value)
                })

        if drift_detected:
            alert = {
                'timestamp': datetime.now(),
                'type': 'data_drift',
                'severity': 'high',
                'drifted_features': drift_features,
                'n_drifted': len(drift_features)
            }
            self.drift_alerts.append(alert)
            logger.warning(f"Data drift detected in {len(drift_features)} features")

        return drift_detected, drift_features

    def detect_output_drift(self, y_pred_new):
        """Detect drift in model predictions"""
        baseline_pred_dist = pd.Series(self.baseline_predictions).value_counts(normalize=True)
        new_pred_dist = pd.Series(y_pred_new).value_counts(normalize=True)

        # Compare distributions
        classes = set(baseline_pred_dist.index) | set(new_pred_dist.index)
        chi2_stat = 0

        for cls in classes:
            exp = baseline_pred_dist.get(cls, 0.01)
            obs = new_pred_dist.get(cls, 0.01)
            chi2_stat += (obs - exp) ** 2 / max(exp, 0.01)

        p_value = 1 - stats.chi2.cdf(chi2_stat, len(classes) - 1)

        if p_value < 0.05:
            alert = {
                'timestamp': datetime.now(),
                'type': 'output_drift',
                'severity': 'medium',
                'chi2_statistic': float(chi2_stat),
                'p_value': float(p_value)
            }
            self.drift_alerts.append(alert)
            logger.warning("Output drift detected in predictions")
            return True

        return False

    def detect_performance_degradation(self, y_true, y_pred):
        """Detect if model performance has degraded"""
        current_accuracy = accuracy_score(y_true, y_pred)
        baseline_accuracy = accuracy_score(self.baseline_targets, self.baseline_predictions)

        degradation_threshold = 0.05  # 5% drop
        degradation = baseline_accuracy - current_accuracy

        if degradation > degradation_threshold:
            alert = {