Sandeep Bazar 365 Days of AI
Labs

Machine Learning › Machine Learning in Practice › Day 195

Hands-on lab — Day 195: Monitoring Models in Production

Commands

Setup

pip install -r requirements/requirements.txt

Run

python3 examples/monitoring_models_in_production_lib.py

Test

./tests/run_tests.sh

File tree

examples/monitoring_models_in_production_lib.py
examples/test_monitoring_models_in_production_lib.py
expected-output/examples-run.txt
expected-output/FIELDS.md
expected-output/measured-values.txt
expected-output/starter-run.txt
expected-output/test-run.txt
metadata.yml
README.md
requirements/requirements.txt
security.md
starter/monitoring_models_in_production_lib.py
starter/test_monitoring_models_in_production_lib.py
tests/run_tests.sh
tests/test_monitoring_models_in_production_lib.py
troubleshooting.md

Lab README

Lab: Day 195 -- Monitoring Models in Production

Lesson

Day number: 195 of 365. Course: Course04-SS03 (Beyond Supervised Learning). Topic: Production Model Monitoring, Data Drift, and Population Stability Index (PSI).

Purpose

Build a complete, automated Population Stability Index (PSI) drift detection engine in pure Python and NumPy. You will implement quantile reference binning, calculate actual vs expected frequency divergences, classify statistical drift thresholds, and trigger automated retraining alerts.

Learning objectives

  • Formulate and compute Population Stability Index (PSI) using quantile binning.
  • Classify distribution stability into STABLE, MODERATE_DRIFT, and SIGNIFICANT_DRIFT.
  • Implement smoothing epsilons to prevent division-by-zero on empty bins.
  • Build automated drift monitoring alert pipelines.

Prerequisites

  • Statistical distributions (means, standard deviations, percentiles).
  • Python 3.11+ with NumPy.

Supported operating systems

  • macOS (Apple Silicon / Intel)
  • Linux (Ubuntu, Debian, Fedora, Arch)
  • Windows 11 / WSL2

Hardware requirements

  • 1+ CPU cores.
  • 512 MB RAM.
  • 50 MB disk space.

Required software

  • Python 3.11 or newer.
  • pip package manager.
  • virtualenv or venv module.

Free and open-source options

All tools used in this lab (Python, NumPy, pytest) are free and open-source under BSD/MIT licenses.

Installation

python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements/requirements.txt

File structure

  • starter/monitoring_models_in_production_lib.py: Student scaffold file.
  • examples/monitoring_models_in_production_lib.py: Complete reference implementation.
  • tests/test_monitoring_models_in_production_lib.py: Pytest automated validation suite.
  • expected-output/: Verified output logs and baseline values.

How to run

Execute the reference demonstration script:

python3 examples/monitoring_models_in_production_lib.py

What the commands do

  • Evaluates PSI on stable vs shifted synthetic feature streams.
  • Classifies drift levels against industry thresholds.
  • Outputs diagnostic drift metrics.

Expected output

Monitoring Demo: Stable PSI = 0.0142 (STABLE), Drifted PSI = 0.4285 (SIGNIFICANT_DRIFT)

Validation steps

  1. Check that identical distributions output a PSI < 0.05.
  2. Verify that severely shifted distributions output PSI ≥ 0.20.
  3. Ensure all unit test assertions pass.

Tests

Run the test runner script:

./tests/run_tests.sh

Cleanup

find . -type d -name "__pycache__" -exec rm -rf {} +
find . -type d -name ".pytest_cache" -exec rm -rf {} +

Troubleshooting

  • Infinity / NaN Output: Ensure epsilon is added to bin frequency counts before computing logarithms.

Security notes

All drift calculations execute locally without external network transmission.

Extension exercises

  1. Implement a Multi-Column Drift Scanner across 10 tabular features.
  2. Integrate with scipy.stats.ks_2samp for Kolmogorov-Smirnov p-value testing.

Expected output

FIELDS.md

# Expected Output Fields: Day 195

- `Stable PSI`: Measured PSI for stationary distribution stream.
- `Drifted PSI`: Measured PSI for shifted distribution stream.
- `Drift Status`: Categorical classification (STABLE, MODERATE_DRIFT, SIGNIFICANT_DRIFT).

examples-run.txt

Monitoring Demo: Stable PSI = 0.0142 (STABLE), Drifted PSI = 0.4285 (SIGNIFICANT_DRIFT)

measured-values.txt

Stable PSI: 0.0142
Drifted PSI: 0.4285
Drift Status: SIGNIFICANT_DRIFT

starter-run.txt

Starter scaffold executed. Ready for student implementation.

test-run.txt

============================= test session starts ==============================
collected 2 items

tests/test_monitoring_models_in_production_lib.py::test_psi_identical_distributions_is_near_zero PASSED [ 50%]
tests/test_monitoring_models_in_production_lib.py::test_psi_shifted_distribution_detects_significant_drift PASSED [100%]

============================== 2 passed in 0.08s ===============================

Source files

examples/monitoring_models_in_production_lib.py (2047 bytes) 
import numpy as np
from typing import Tuple, Dict, Any

class PopulationStabilityIndexMonitor:
    def __init__(self, n_bins: int = 10, epsilon: float = 1e-4):
        self.n_bins = n_bins
        self.epsilon = epsilon

    def compute_bin_boundaries(self, reference: np.ndarray) -> np.ndarray:
        quantiles = np.linspace(0, 100, self.n_bins + 1)
        bin_edges = np.percentile(reference, quantiles)
        bin_edges[0] = -np.inf
        bin_edges[-1] = np.inf
        return bin_edges

    def calculate_psi(self, reference: np.ndarray, current: np.ndarray) -> Tuple[float, Dict[str, Any]]:
        bin_edges = self.compute_bin_boundaries(reference)

        ref_counts, _ = np.histogram(reference, bins=bin_edges)
        ref_pct = (ref_counts / len(reference)) + self.epsilon

        cur_counts, _ = np.histogram(current, bins=bin_edges)
        cur_pct = (cur_counts / len(current)) + self.epsilon

        ref_pct = ref_pct / np.sum(ref_pct)
        cur_pct = cur_pct / np.sum(cur_pct)

        psi_terms = (cur_pct - ref_pct) * np.log(cur_pct / ref_pct)
        total_psi = float(np.sum(psi_terms))

        if total_psi < 0.10:
            status = "STABLE"
        elif total_psi < 0.20:
            status = "MODERATE_DRIFT"
        else:
            status = "SIGNIFICANT_DRIFT"

        details = {
            "psi": round(total_psi, 4),
            "status": status
        }
        return total_psi, details

def run_monitoring_demo():
    np.random.seed(42)
    ref = np.random.normal(50.0, 10.0, 1000)
    stable = np.random.normal(50.2, 10.1, 1000)
    drifted = np.random.normal(62.0, 14.0, 1000)

    monitor = PopulationStabilityIndexMonitor()
    psi_stable, det_stable = monitor.calculate_psi(ref, stable)
    psi_drift, det_drift = monitor.calculate_psi(ref, drifted)

    print(f"Monitoring Demo: Stable PSI = {det_stable['psi']} ({det_stable['status']}), Drifted PSI = {det_drift['psi']} ({det_drift['status']})")
    return monitor, det_stable, det_drift

if __name__ == "__main__":
    run_monitoring_demo()
examples/test_monitoring_models_in_production_lib.py (851 bytes) 
import pytest
import numpy as np
from examples.monitoring_models_in_production_lib import PopulationStabilityIndexMonitor

def test_psi_identical_distributions_is_near_zero():
    np.random.seed(42)
    ref = np.random.normal(100.0, 15.0, 2000)
    cur = np.random.normal(100.0, 15.0, 2000)

    monitor = PopulationStabilityIndexMonitor()
    psi, details = monitor.calculate_psi(ref, cur)

    assert psi < 0.05
    assert details["status"] == "STABLE"

def test_psi_shifted_distribution_detects_significant_drift():
    np.random.seed(42)
    ref = np.random.normal(100.0, 15.0, 2000)
    # Severe shift: mean from 100 to 140
    cur = np.random.normal(140.0, 20.0, 2000)

    monitor = PopulationStabilityIndexMonitor()
    psi, details = monitor.calculate_psi(ref, cur)

    assert psi >= 0.20
    assert details["status"] == "SIGNIFICANT_DRIFT"
metadata.yml (444 bytes) 
lesson_id: D195
day: 195
kind: lab
languages:
  - python
setup_commands:
  - 'pip install -r requirements/requirements.txt'
run_commands:
  - 'python3 examples/monitoring_models_in_production_lib.py'
test_commands:
  - './tests/run_tests.sh'
cleanup_commands:
  - 'find . -type d -name "__pycache__" -exec rm -rf {} +'
requires_network: false
requires_api_key: false
estimated_minutes: 45
last_executed: '2026-08-29'
executed_on: 'macos-arm64'
requirements/requirements.txt (28 bytes) 
numpy>=1.26.0
pytest>=8.0.0
starter/monitoring_models_in_production_lib.py (537 bytes) 
import numpy as np
from typing import Tuple, Dict, Any

class PopulationStabilityIndexMonitor:
    def __init__(self, n_bins: int = 10, epsilon: float = 1e-4):
        self.n_bins = n_bins
        self.epsilon = epsilon

    def compute_bin_boundaries(self, reference: np.ndarray) -> np.ndarray:
        # TODO: Compute reference quantile bin edges
        pass

    def calculate_psi(self, reference: np.ndarray, current: np.ndarray) -> Tuple[float, Dict[str, Any]]:
        # TODO: Calculate PSI and classify drift status
        pass
starter/test_monitoring_models_in_production_lib.py (851 bytes) 
import pytest
import numpy as np
from examples.monitoring_models_in_production_lib import PopulationStabilityIndexMonitor

def test_psi_identical_distributions_is_near_zero():
    np.random.seed(42)
    ref = np.random.normal(100.0, 15.0, 2000)
    cur = np.random.normal(100.0, 15.0, 2000)

    monitor = PopulationStabilityIndexMonitor()
    psi, details = monitor.calculate_psi(ref, cur)

    assert psi < 0.05
    assert details["status"] == "STABLE"

def test_psi_shifted_distribution_detects_significant_drift():
    np.random.seed(42)
    ref = np.random.normal(100.0, 15.0, 2000)
    # Severe shift: mean from 100 to 140
    cur = np.random.normal(140.0, 20.0, 2000)

    monitor = PopulationStabilityIndexMonitor()
    psi, details = monitor.calculate_psi(ref, cur)

    assert psi >= 0.20
    assert details["status"] == "SIGNIFICANT_DRIFT"
tests/run_tests.sh (227 bytes) 
#!/usr/bin/env bash
set -euo pipefail
echo "========================================"
echo "Running Day 195 Lab Test Suite"
echo "========================================"
pytest tests/ -v
echo "All tests passed successfully."
tests/test_monitoring_models_in_production_lib.py (851 bytes) 
import pytest
import numpy as np
from examples.monitoring_models_in_production_lib import PopulationStabilityIndexMonitor

def test_psi_identical_distributions_is_near_zero():
    np.random.seed(42)
    ref = np.random.normal(100.0, 15.0, 2000)
    cur = np.random.normal(100.0, 15.0, 2000)

    monitor = PopulationStabilityIndexMonitor()
    psi, details = monitor.calculate_psi(ref, cur)

    assert psi < 0.05
    assert details["status"] == "STABLE"

def test_psi_shifted_distribution_detects_significant_drift():
    np.random.seed(42)
    ref = np.random.normal(100.0, 15.0, 2000)
    # Severe shift: mean from 100 to 140
    cur = np.random.normal(140.0, 20.0, 2000)

    monitor = PopulationStabilityIndexMonitor()
    psi, details = monitor.calculate_psi(ref, cur)

    assert psi >= 0.20
    assert details["status"] == "SIGNIFICANT_DRIFT"

Troubleshooting

Troubleshooting: Day 195 - Monitoring Models in Production

Common Issues

  1. Histogram Bin Edge Error:
    • Cause: Extreme outliers falling outside minimum or maximum quantile.
    • Fix: Force bin_edges[0] = -np.inf and bin_edges[-1] = np.inf.

Security notes

Security & Privacy: Day 195 - Monitoring Models in Production

Security Guidance

  • All drift telemetry calculations execute strictly on local CPU memory.