Introduction

Model evaluation does not end when a Machine Learning model is deployed. Production data, user behavior, business processes, upstream systems, feature pipelines, software dependencies, and external conditions can change after release.

A prediction service may continue returning successful responses while its inputs become invalid or its predictions become less useful. Monitoring only server availability therefore provides an incomplete view of an ML system.

Model monitoring is the continuous or scheduled collection, analysis, and interpretation of signals describing a deployed model, its data, its serving environment, and its business use.

Learning Objectives

Prerequisites

python -m pip install pandas numpy scipy scikit-learn

What Is Model Monitoring?

Model monitoring examines whether a deployed ML system remains available, receives valid data, produces expected outputs, satisfies quality requirements, and supports the intended business process.

Application Monitoring vs Model Monitoring

Monitoring Architecture

Monitoring Event Schema

Every prediction should be traceable to the model and serving configuration that produced it. The monitoring record should contain only fields justified by the monitoring purpose.

{
    "request_id": "generated-request-identifier",
    "event_time": "recorded-event-time",
    "model_name": "customer-churn-classifier",
    "model_version": "1.0.0",
    "application_version": "1.0.0",
    "schema_version": "1.0.0",
    "prediction": "churned",
    "probability": 0.71,
    "decision_threshold": 0.6,
    "latency_ms": 4.2,
    "validation_status": "passed"
}

Data Minimization

Service Health Monitoring

Latency Percentiles

Average latency can hide slow requests. Percentiles provide a more complete view of response-time distribution.

import numpy as np


def calculate_latency_percentiles(
    latency_values
):
    latency_values = np.asarray(
        latency_values,
        dtype=float
    )

    if latency_values.size == 0:

        raise ValueError(
            "Latency observations are required."
        )

    if not np.isfinite(
        latency_values
    ).all():

        raise ValueError(
            "Latency observations must be finite."
        )

    return {
        "p50": float(
            np.percentile(
                latency_values,
                50
            )
        ),
        "p90": float(
            np.percentile(
                latency_values,
                90
            )
        ),
        "p95": float(
            np.percentile(
                latency_values,
                95
            )
        ),
        "p99": float(
            np.percentile(
                latency_values,
                99
            )
        )
    }

Input Schema Monitoring

def calculate_schema_issue_rates(
    monitoring_frame
):
    total_records = len(
        monitoring_frame
    )

    if total_records == 0:

        raise ValueError(
            "Monitoring records are required."
        )

    return {
        "validation_failure_rate": float(
            (
                monitoring_frame[
                    "validation_status"
                ]
                !=
                "passed"
            ).mean()
        ),
        "missing_feature_rate": float(
            monitoring_frame[
                "missing_feature"
            ].mean()
        ),
        "unexpected_feature_rate": float(
            monitoring_frame[
                "unexpected_feature"
            ].mean()
        ),
        "unsupported_category_rate": float(
            monitoring_frame[
                "unsupported_category"
            ].mean()
        )
    }

Training-Serving Skew

Training-serving skew occurs when production inference uses different inputs, feature definitions, preprocessing, or timing semantics from model training.

Types of Drift

Data Drift

Data drift occurs when the production distribution of one or more input features differs from the selected reference distribution.

Concept Drift

Concept drift occurs when the relationship between model inputs and the outcome changes.

Concept drift cannot be established reliably from input distributions alone. Verified outcomes must generally be joined back to earlier predictions before performance changes can be measured.

Choose a Monitoring Baseline

Population Stability Index

Population Stability Index, or PSI, compares the proportions of observations across corresponding reference and current bins.

Here, \(E_i\) is the reference proportion and \(A_i\) is the current proportion in bin \(i\).

import numpy as np


def calculate_psi(
    reference_values,
    current_values,
    number_of_bins=10,
    epsilon=1e-6
):
    reference_values = np.asarray(
        reference_values,
        dtype=float
    )

    current_values = np.asarray(
        current_values,
        dtype=float
    )

    reference_values = reference_values[
        np.isfinite(
            reference_values
        )
    ]

    current_values = current_values[
        np.isfinite(
            current_values
        )
    ]

    if (
        reference_values.size == 0
        or current_values.size == 0
    ):
        raise ValueError(
            "Both datasets require finite values."
        )

    bin_edges = np.unique(
        np.quantile(
            reference_values,
            np.linspace(
                0,
                1,
                number_of_bins + 1
            )
        )
    )

    if bin_edges.size < 2:

        return 0.0

    bin_edges[
        0
    ] = -np.inf

    bin_edges[
        -1
    ] = np.inf

    reference_counts, _ = np.histogram(
        reference_values,
        bins=bin_edges
    )

    current_counts, _ = np.histogram(
        current_values,
        bins=bin_edges
    )

    reference_proportions = (
        reference_counts
        /
        reference_counts.sum()
    )

    current_proportions = (
        current_counts
        /
        current_counts.sum()
    )

    reference_proportions = np.clip(
        reference_proportions,
        epsilon,
        None
    )

    current_proportions = np.clip(
        current_proportions,
        epsilon,
        None
    )

    return float(
        np.sum(
            (
                current_proportions
                -
                reference_proportions
            )
            *
            np.log(
                current_proportions
                /
                reference_proportions
            )
        )
    )

Kolmogorov-Smirnov Test

The two-sample Kolmogorov-Smirnov test compares empirical cumulative distributions for a continuous feature.

from scipy.stats import ks_2samp


def calculate_ks_drift(
    reference_values,
    current_values
):
    result = ks_2samp(
        reference_values,
        current_values
    )

    return {
        "statistic": float(
            result.statistic
        ),
        "p_value": float(
            result.pvalue
        )
    }

Categorical Drift

import pandas as pd


def compare_category_distributions(
    reference_values,
    current_values
):
    reference_distribution = (
        pd.Series(
            reference_values
        )
        .fillna(
            "__MISSING__"
        )
        .value_counts(
            normalize=True
        )
    )

    current_distribution = (
        pd.Series(
            current_values
        )
        .fillna(
            "__MISSING__"
        )
        .value_counts(
            normalize=True
        )
    )

    categories = sorted(
        set(
            reference_distribution.index
        )
        |
        set(
            current_distribution.index
        )
    )

    comparison = pd.DataFrame(
        {
            "category": categories,
            "reference_proportion": [
                float(
                    reference_distribution.get(
                        category,
                        0.0
                    )
                )
                for category in categories
            ],
            "current_proportion": [
                float(
                    current_distribution.get(
                        category,
                        0.0
                    )
                )
                for category in categories
            ]
        }
    )

    comparison[
        "absolute_change"
    ] = (
        comparison[
            "current_proportion"
        ]
        -
        comparison[
            "reference_proportion"
        ]
    ).abs()

    return comparison

Prediction Drift

Prediction drift examines changes in scores, classes, forecasts, or other model outputs before verified outcomes become available.

Delayed Ground Truth

The true outcome is frequently unavailable when the prediction is generated. Monitoring must therefore preserve a controlled join between a historical prediction and its later verified outcome.

Monitor Classification Quality

from sklearn.metrics import accuracy_score
from sklearn.metrics import average_precision_score
from sklearn.metrics import f1_score
from sklearn.metrics import precision_score
from sklearn.metrics import recall_score
from sklearn.metrics import roc_auc_score


def calculate_classification_metrics(
    true_labels,
    probabilities,
    decision_threshold
):
    predicted_labels = (
        probabilities
        >=
        decision_threshold
    ).astype(
        int
    )

    return {
        "accuracy": float(
            accuracy_score(
                true_labels,
                predicted_labels
            )
        ),
        "precision": float(
            precision_score(
                true_labels,
                predicted_labels,
                zero_division=0
            )
        ),
        "recall": float(
            recall_score(
                true_labels,
                predicted_labels,
                zero_division=0
            )
        ),
        "f1": float(
            f1_score(
                true_labels,
                predicted_labels,
                zero_division=0
            )
        ),
        "roc_auc": float(
            roc_auc_score(
                true_labels,
                probabilities
            )
        ),
        "average_precision": float(
            average_precision_score(
                true_labels,
                probabilities
            )
        )
    }

Monitor Regression Quality

import numpy as np
from sklearn.metrics import mean_absolute_error
from sklearn.metrics import mean_squared_error
from sklearn.metrics import r2_score


def calculate_regression_metrics(
    true_values,
    predicted_values
):
    return {
        "mean_absolute_error": float(
            mean_absolute_error(
                true_values,
                predicted_values
            )
        ),
        "root_mean_squared_error": float(
            np.sqrt(
                mean_squared_error(
                    true_values,
                    predicted_values
                )
            )
        ),
        "r_squared": float(
            r2_score(
                true_values,
                predicted_values
            )
        )
    }

Monitor Calibration

A classification model is calibrated when predicted probabilities correspond meaningfully to observed outcome frequencies.

from sklearn.metrics import brier_score_loss


def calculate_brier_score(
    true_labels,
    probabilities
):
    return float(
        brier_score_loss(
            true_labels,
            probabilities
        )
    )

Segment and Group Monitoring

Aggregate metrics can hide poor behavior in particular products, regions, devices, time periods, or other relevant operating groups.

import pandas as pd


def calculate_group_metrics(
    monitoring_frame,
    group_column,
    target_column,
    prediction_column
):
    rows = []

    for group_value, group_data in (
        monitoring_frame.groupby(
            group_column,
            dropna=False
        )
    ):
        rows.append(
            {
                "group": group_value,
                "sample_count": int(
                    len(
                        group_data
                    )
                ),
                "accuracy": float(
                    (
                        group_data[
                            target_column
                        ]
                        ==
                        group_data[
                            prediction_column
                        ]
                    ).mean()
                )
            }
        )

    return pd.DataFrame(
        rows
    )

Business Outcome Monitoring

Monitoring Windows

Design Actionable Alerts

Reduce Alert Fatigue

Incident Response Workflow

When Should a Model Be Retrained?

Rollback and Fallback

Monitoring Dashboard Design

Build a Monitoring Report

import json
from pathlib import Path


def create_monitoring_report(
    service_metrics,
    input_metrics,
    drift_metrics,
    quality_metrics,
    output_path
):
    report = {
        "service": service_metrics,
        "input_quality": input_metrics,
        "drift": drift_metrics,
        "model_quality": quality_metrics
    }

    output_path = Path(
        output_path
    )

    output_path.parent.mkdir(
        parents=True,
        exist_ok=True
    )

    with open(
        output_path,
        "w",
        encoding="utf-8"
    ) as report_file:

        json.dump(
            report,
            report_file,
            indent=4
        )

    return report

Test Monitoring Logic

import numpy as np
import pytest


def test_psi_is_zero_for_identical_data():

    values = np.array(
        [
            1,
            2,
            3,
            4,
            5
        ],
        dtype=float
    )

    result = calculate_psi(
        values,
        values,
        number_of_bins=5
    )

    assert result == pytest.approx(
        0.0
    )


def test_latency_rejects_empty_input():

    with pytest.raises(
        ValueError,
        match="required"
    ):
        calculate_latency_percentiles(
            []
        )


def test_classification_metric_range():

    metrics = calculate_classification_metrics(
        true_labels=np.array(
            [
                0,
                1,
                1,
                0
            ]
        ),
        probabilities=np.array(
            [
                0.1,
                0.8,
                0.7,
                0.2
            ]
        ),
        decision_threshold=0.5
    )

    for metric_name in [
        "accuracy",
        "precision",
        "recall",
        "f1",
        "roc_auc",
        "average_precision"
    ]:
        assert (
            0.0
            <=
            metrics[
                metric_name
            ]
            <=
            1.0
        )

python -m pytest -v

Scheduled Monitoring Pipeline

Monitoring Cost Management

Responsible Model Monitoring

Common Monitoring Mistakes

Recommended Monitoring Workflow

Hands-On Practice

Knowledge Check

Model Monitoring Readiness Checklist

Lesson Summary