Model Monitoring

Model monitoring ensures ML models perform well in production. This sheet covers drift detection, performance monitoring, data quality, alerting, and retraining triggers.

Table of Contents

• Installation / Setup
• Core Syntax
• Common Operations
• Practical Examples
• Common Patterns
• Best Practices
• Common Mistakes
• Troubleshooting
• Related Cheat Sheets
• Official Documentation

Installation / Setup

pip install evidently whylabs
pip install great-expectations

Core Syntax

Types of Monitoring

Type	Description	What to Track
Data Drift	Input data distribution changes	Feature distributions, missing values
Concept Drift	Relationship between inputs and outputs changes	Model accuracy, prediction distribution
Performance Drift	Model performance degrades	Accuracy, latency, throughput
Data Quality	Data integrity issues	Nulls, duplicates, outliers

Drift Detection Methods

Method	Description	When to Use
PSI (Population Stability Index)	Measures distribution shift	Tabular data
KS Test	Kolmogorov-Smirnov test	Continuous features
Chi-Square	Categorical distribution test	Categorical features
Page-Hinkley	Sequential change detection	Streaming data
ADWIN	Adaptive windowing	Streaming data

Common Operations

Evidently Monitoring

from evidently.report import Report
from evidently.metric_preset import (
    DataDriftPreset,
    DataQualityPreset,
    TargetDriftPreset,
)

# Data drift report
report = Report(metrics=[DataDriftPreset()])
report.run(reference_data=train_data, current_data=production_data)
report.show()
report.save_html("drift_report.html")

# Data quality report
report = Report(metrics=[DataQualityPreset()])
report.run(reference_data=train_data, current_data=production_data)
report.show()

# Target drift report
report = Report(metrics=[TargetDriftPreset()])
report.run(reference_data=train_data, current_data=production_data)
report.show()

Monitoring Dashboard

from evidently.dashboard import Dashboard
from evidently.dashboard_tabs import DataDriftTab

# Create dashboard
dashboard = Dashboard(tabs=[DataDriftTab()])
dashboard.run(reference_data=train_data, current_data=production_data)
dashboard.save("drift_dashboard.html")

Great Expectations

import great_expectations as gx

context = gx.get_context()

# Create expectation suite
suite = context.add_expectation_suite("data_quality")
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="age", min_value=0, max_value=150
    )
)

# Validate data
batch_request = context.sources.pandas_default.read_dataframe(df)
result = context.run_validator(batch_request, suite)
print(result.success)

Custom Monitoring

import numpy as np
from scipy import stats

def detect_drift(reference, current, threshold=0.05):
    """Detect drift using KS test."""
    drift_detected = {}
    for col in reference.columns:
        if reference[col].dtype in ['float64', 'int64']:
            stat, p_value = stats.ks_2samp(reference[col], current[col])
            drift_detected[col] = {
                'statistic': stat,
                'p_value': p_value,
                'drifted': p_value < threshold
            }
    return drift_detected

Practical Examples

Example 1 — Drift detection pipeline

import mlflow
import pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

def monitor_drift(train_data, production_data, threshold=0.05):
    # Run drift report
    report = Report(metrics=[DataDriftPreset()])
    report.run(reference_data=train_data, current_data=production_data)

    # Extract results
    result = report.as_dict()
    drift_detected = result['metrics'][0]['result']['dataset_drift']

    # Log to MLflow
    with mlflow.start_run(run_name="drift_monitoring"):
        mlflow.log_metric("drift_detected", int(drift_detected))
        mlflow.log_artifact("drift_report.html")

    # Alert if drift detected
    if drift_detected:
        send_alert("Data drift detected!")
        trigger_retraining()

    return drift_detected

# Schedule daily monitoring
train_data = pd.read_parquet("data/training.parquet")
production_data = pd.read_parquet("data/production_today.parquet")
monitor_drift(train_data, production_data)

Example 2 — Performance monitoring

import mlflow
from datetime import datetime, timedelta

def monitor_performance(model, production_data):
    # Make predictions
    predictions = model.predict(production_data.drop(columns=["target"]))

    # Calculate metrics
    accuracy = (predictions == production_data["target"]).mean()

    # Log metrics
    with mlflow.start_run(run_name="performance_monitoring"):
        mlflow.log_metric("accuracy", accuracy)
        mlflow.log_metric("timestamp", datetime.now().timestamp())

    # Check for degradation
    baseline_accuracy = 0.95
    if accuracy < baseline_accuracy * 0.95:  # 5% degradation
        send_alert(f"Model accuracy dropped to {accuracy:.4f}")
        trigger_retraining()

    return accuracy

# Monitor daily
production_data = pd.read_parquet("data/production.parquet")
monitor_performance(model, production_data)

Example 3 — Data quality checks

import great_expectations as gx

def validate_data_quality(df):
    context = gx.get_context()

    # Create expectations
    suite = context.add_expectation_suite("production_data")
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id")
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeUnique(column="user_id")
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="age", min_value=0, max_value=150
        )
    )

    # Validate
    batch_request = context.sources.pandas_default.read_dataframe(df)
    result = context.run_validator(batch_request, suite)

    if not result.success:
        send_alert("Data quality issues detected!")
        return False

    return True

Common Patterns

Pattern: alerting

import smtplib
from email.mime.text import MIMEText

def send_alert(message, level="warning"):
    msg = MIMEText(f"Level: {level}\n\n{message}")
    msg["Subject"] = f"ML Alert: {level}"
    msg["From"] = "alerts@example.com"
    msg["To"] = "team@example.com"

    with smtplib.SMTP("smtp.example.com") as server:
        server.send_message(msg)

Pattern: retraining trigger

def check_retraining_trigger(drift_detected, accuracy_degraded):
    """Decide if retraining is needed."""
    if drift_detected and accuracy_degraded:
        return "urgent"
    elif drift_detected or accuracy_degraded:
        return "recommended"
    return "not_needed"

def trigger_retraining():
    """Trigger model retraining pipeline."""
    # Call Airflow DAG or GitHub Action
    pass

Pattern: baseline comparison

def compare_with_baseline(current_metrics, baseline_metrics):
    """Compare current performance with baseline."""
    results = {}
    for metric, current_value in current_metrics.items():
        baseline_value = baseline_metrics.get(metric)
        if baseline_value:
            change = (current_value - baseline_value) / baseline_value
            results[metric] = {
                "current": current_value,
                "baseline": baseline_value,
                "change": change,
                "degraded": change < -0.05  # 5% degradation
            }
    return results

Pattern: dashboard

from fastapi import FastAPI
import pandas as pd

app = FastAPI()

@app.get("/dashboard")
def dashboard():
    # Load recent metrics
    metrics = pd.read_parquet("metrics/metrics.parquet")

    # Calculate summary
    summary = {
        "accuracy": metrics["accuracy"].mean(),
        "latency_p99": metrics["latency"].quantile(0.99),
        "drift_score": metrics["drift_score"].iloc[-1],
    }

    return summary

Best Practices

• Monitor continuously — not just after deployment.
• Track data drift — compare production with training data.
• Track performance drift — monitor accuracy, latency, throughput.
• Set up alerts — notify when issues are detected.
• Automate retraining — trigger on drift or degradation.
• Log everything — metrics, predictions, data samples.
• Use dashboards — visualize trends over time.
• Validate data quality — check for nulls, outliers, anomalies.

Common Mistakes

Mistake	Fix
No monitoring	Implement drift and performance monitoring
No alerts	Set up alerting for issues
No baseline	Establish performance baselines
Manual monitoring	Automate monitoring pipelines
Ignoring data quality	Validate data at ingestion

Troubleshooting

Symptom	Cause / Fix
Drift not detected	Check thresholds; check data quality
High false alarms	Tune thresholds; use ensemble methods
Slow monitoring	Sample data; use faster methods
No alerts	Check alert configuration; check email

Related Cheat Sheets

• mlops.md — MLOps overview
• mlflow.md — experiment tracking
• model-deployment.md — deployment strategies
• model-versioning.md — model versioning
• scikit-learn — model training

Official Documentation

• Evidently AI — https://docs.evidentlyai.com/
• Great Expectations — https://docs.greatexpectations.io/
• Whylabs — https://docs.whylabs.ai/

Last Updated

2026-09-06 — Initial version.