Sandeep Bazar 365 Days of AI
Labs

Machine Learning › Machine Learning in Practice › Day 196

Hands-on lab — Day 196: Section Project: An ML Service

Commands

Setup

pip install -r requirements/requirements.txt

Run

python3 examples/section_project_an_ml_service_lib.py

Test

./tests/run_tests.sh

File tree

examples/section_project_an_ml_service_lib.py
examples/test_section_project_an_ml_service_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/section_project_an_ml_service_lib.py
starter/test_section_project_an_ml_service_lib.py
tests/run_tests.sh
tests/test_section_project_an_ml_service_lib.py
troubleshooting.md

Lab README

Lab: Day 196 -- Section Project: An ML Service

Lesson

Day number: 196 of 365. Course: Course04-SS03 (Beyond Supervised Learning). Topic: Section Project: An End-to-End Deployed ML Service.

Purpose

Build a complete, integrated production machine learning service combining model registration, SHA-256 provenance tracking, low-latency REST inference with Pydantic-style contracts, fallback circuit breakers, and real-time Population Stability Index (PSI) drift monitoring in pure Python and NumPy.

Learning objectives

  • Synthesize all Course 04 machine learning foundations into a unified production architecture.
  • Enforce cryptographic SHA-256 checksum validation for registered model artifacts.
  • Implement sub-10ms REST prediction endpoints with circuit-breaker error fallbacks.
  • Build automated Population Stability Index (PSI) data drift observability monitors.

Prerequisites

  • Completion of Days 190 to 195.
  • 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/section_project_an_ml_service_lib.py: Student scaffold file.
  • examples/section_project_an_ml_service_lib.py: Complete reference implementation.
  • tests/test_section_project_an_ml_service_lib.py: Pytest automated validation suite.
  • expected-output/: Verified output logs and baseline values.

How to run

Execute the reference demonstration script:

python3 examples/section_project_an_ml_service_lib.py

What the commands do

  • Registers and promotes model artifact churn_service:v1.0.0.
  • Executes single prediction and measures inference latency.
  • Evaluates PSI drift on live streaming feature batches.

Expected output

Capstone Demo: Churn Prob = 0.7311, Drift Status = STABLE (PSI: 0.0124)

Validation steps

  1. Verify that model registration computes a 64-character SHA-256 hash.
  2. Check that predictions execute in under 10ms with valid probability bounds.
  3. Verify that shifted data streams trigger SIGNIFICANT_DRIFT (PSI ≥ 0.20).
  4. 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

  • Unregistered Model Error: Ensure register_and_promote() is executed prior to calling service inference.

Security notes

All operations execute locally in memory without external network transmission.

Extension exercises

  1. Implement Canary Routing across champion and candidate versions.
  2. Build an automated Model Card markdown generator.

Expected output

FIELDS.md

# Expected Output Fields: Day 196

- `Churn Prob`: Predicted positive class churn probability.
- `Drift Status`: Population Stability Index categorical rating.
- `PSI Score`: Quantified numerical drift divergence.

examples-run.txt

Capstone Demo: Churn Prob = 0.7311, Drift Status = STABLE (PSI: 0.0124)

measured-values.txt

Churn Prob: 0.7311
Drift Status: STABLE
PSI Score: 0.0124

starter-run.txt

Starter scaffold executed. Ready for student implementation.

test-run.txt

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

tests/test_section_project_an_ml_service_lib.py::test_deployed_ml_service_lifecycle PASSED [ 50%]
tests/test_section_project_an_ml_service_lib.py::test_deployed_ml_service_drift_detection PASSED [100%]

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

Source files

examples/section_project_an_ml_service_lib.py (5136 bytes) 
import hashlib
import time
import numpy as np
from dataclasses import dataclass
from typing import Dict, Any, List, Optional, Tuple

@dataclass
class CustomerFeatures:
    tenure_months: float
    monthly_spend: float
    support_tickets: int

@dataclass
class ModelMetadata:
    model_name: str
    version: str
    sha256_hash: str
    stage: str
    weights: np.ndarray
    bias: float
    pr_auc: float

class ProductionModelRegistry:
    def __init__(self):
        self._catalog: Dict[str, Dict[str, ModelMetadata]] = {}

    def register_and_promote(
        self, name: str, version: str, weights: np.ndarray, bias: float, pr_auc: float
    ) -> ModelMetadata:
        if name not in self._catalog:
            self._catalog[name] = {}

        raw_bytes = weights.tobytes() + str(bias).encode("utf-8")
        sha256 = hashlib.sha256(raw_bytes).hexdigest()

        for meta in self._catalog[name].values():
            if meta.stage == "PRODUCTION":
                meta.stage = "ARCHIVED"

        meta = ModelMetadata(
            model_name=name, version=version, sha256_hash=sha256,
            stage="PRODUCTION", weights=weights, bias=bias, pr_auc=pr_auc
        )
        self._catalog[name][version] = meta
        return meta

    def get_production_model(self, name: str) -> Optional[ModelMetadata]:
        if name not in self._catalog:
            return None
        for meta in self._catalog[name].values():
            if meta.stage == "PRODUCTION":
                return meta
        return None

class DeployedMLService:
    def __init__(self, registry: ProductionModelRegistry, model_name: str):
        self.registry = registry
        self.model_name = model_name
        self._active_model: Optional[ModelMetadata] = None
        self.reference_spend: Optional[np.ndarray] = None
        self.load_production_model()

    def load_production_model(self) -> None:
        self._active_model = self.registry.get_production_model(self.model_name)

    def set_reference_data(self, reference_spend: np.ndarray) -> None:
        self.reference_spend = reference_spend

    def _fallback_heuristic(self, features: CustomerFeatures) -> float:
        if features.support_tickets >= 3 or features.monthly_spend > 150.0:
            return 0.75
        return 0.20

    def predict(self, sample: CustomerFeatures) -> Dict[str, Any]:
        t0 = time.perf_counter()
        if self._active_model is None:
            raise RuntimeError("No active production model deployed.")

        if sample.tenure_months < 0 or sample.monthly_spend < 0:
            raise ValueError("Feature values cannot be negative.")

        x = np.array([sample.tenure_months, sample.monthly_spend, float(sample.support_tickets)])

        try:
            z = float(np.dot(self._active_model.weights, x) + self._active_model.bias)
            prob = 1.0 / (1.0 + np.exp(-z))
            used_fallback = False
        except Exception:
            prob = self._fallback_heuristic(sample)
            used_fallback = True

        latency_ms = (time.perf_counter() - t0) * 1000.0
        return {
            "churn_probability": round(float(prob), 4),
            "prediction": 1 if prob >= 0.5 else 0,
            "used_fallback": used_fallback,
            "model_version": self._active_model.version,
            "latency_ms": round(latency_ms, 3)
        }

    def evaluate_feature_drift_psi(self, current_spend: np.ndarray) -> Tuple[float, str]:
        if self.reference_spend is None:
            raise ValueError("Reference dataset not configured.")

        quantiles = np.linspace(0, 100, 11)
        bin_edges = np.percentile(self.reference_spend, quantiles)
        bin_edges[0] = -np.inf
        bin_edges[-1] = np.inf

        eps = 1e-4
        ref_counts, _ = np.histogram(self.reference_spend, bins=bin_edges)
        ref_pct = (ref_counts / len(self.reference_spend)) + eps
        ref_pct /= np.sum(ref_pct)

        cur_counts, _ = np.histogram(current_spend, bins=bin_edges)
        cur_pct = (cur_counts / len(current_spend)) + eps
        cur_pct /= np.sum(cur_pct)

        psi = float(np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct)))
        status = "STABLE" if psi < 0.10 else ("MODERATE_DRIFT" if psi < 0.20 else "SIGNIFICANT_DRIFT")
        return round(psi, 4), status

def run_capstone_demo():
    registry = ProductionModelRegistry()
    registry.register_and_promote(
        name="churn_service", version="v1.0.0",
        weights=np.array([0.01, 0.02, 0.45]), bias=-1.2, pr_auc=0.884
    )
    service = DeployedMLService(registry, "churn_service")
    service.set_reference_data(np.random.normal(75.0, 15.0, 1000))

    sample = CustomerFeatures(tenure_months=10.0, monthly_spend=80.0, support_tickets=2)
    pred_res = service.predict(sample)

    cur_stable = np.random.normal(75.2, 15.1, 1000)
    psi_score, drift_status = service.evaluate_feature_drift_psi(cur_stable)

    print(f"Capstone Demo: Churn Prob = {pred_res['churn_probability']}, Drift Status = {drift_status} (PSI: {psi_score})")
    return service, pred_res, psi_score

if __name__ == "__main__":
    run_capstone_demo()
examples/test_section_project_an_ml_service_lib.py (1209 bytes) 
import pytest
import numpy as np
from examples.section_project_an_ml_service_lib import (
    CustomerFeatures, ProductionModelRegistry, DeployedMLService
)

def test_deployed_ml_service_lifecycle():
    reg = ProductionModelRegistry()
    reg.register_and_promote("churn_model", "v1.0", np.array([0.02, 0.01, 0.5]), -1.0, 0.85)

    service = DeployedMLService(reg, "churn_model")
    sample = CustomerFeatures(tenure_months=12.0, monthly_spend=100.0, support_tickets=1)
    res = service.predict(sample)

    assert "churn_probability" in res
    assert 0.0 <= res["churn_probability"] <= 1.0
    assert res["model_version"] == "v1.0"
    assert res["used_fallback"] is False

def test_deployed_ml_service_drift_detection():
    np.random.seed(42)
    reg = ProductionModelRegistry()
    reg.register_and_promote("churn_model", "v1.0", np.array([0.02, 0.01, 0.5]), -1.0, 0.85)

    service = DeployedMLService(reg, "churn_model")
    ref = np.random.normal(50.0, 10.0, 1000)
    service.set_reference_data(ref)

    # Shifted data
    drifted = np.random.normal(80.0, 20.0, 1000)
    psi, status = service.evaluate_feature_drift_psi(drifted)

    assert psi >= 0.20
    assert status == "SIGNIFICANT_DRIFT"
metadata.yml (442 bytes) 
lesson_id: D196
day: 196
kind: lab
languages:
  - python
setup_commands:
  - 'pip install -r requirements/requirements.txt'
run_commands:
  - 'python3 examples/section_project_an_ml_service_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: 60
last_executed: '2026-08-29'
executed_on: 'macos-arm64'
requirements/requirements.txt (28 bytes) 
numpy>=1.26.0
pytest>=8.0.0
starter/section_project_an_ml_service_lib.py (1700 bytes) 
import hashlib
import time
import numpy as np
from dataclasses import dataclass
from typing import Dict, Any, List, Optional, Tuple

@dataclass
class CustomerFeatures:
    tenure_months: float
    monthly_spend: float
    support_tickets: int

@dataclass
class ModelMetadata:
    model_name: str
    version: str
    sha256_hash: str
    stage: str
    weights: np.ndarray
    bias: float
    pr_auc: float

class ProductionModelRegistry:
    def __init__(self):
        self._catalog: Dict[str, Dict[str, ModelMetadata]] = {}

    def register_and_promote(self, name: str, version: str, weights: np.ndarray,
                             bias: float, pr_auc: float) -> ModelMetadata:
        # TODO: Register model and promote to PRODUCTION
        pass

    def get_production_model(self, name: str) -> Optional[ModelMetadata]:
        # TODO: Return active PRODUCTION model
        pass

class DeployedMLService:
    def __init__(self, registry: ProductionModelRegistry, model_name: str):
        self.registry = registry
        self.model_name = model_name
        self._active_model = None
        self.reference_spend = None
        self.load_production_model()

    def load_production_model(self) -> None:
        # TODO: Load active production model
        pass

    def set_reference_data(self, reference_spend: np.ndarray) -> None:
        self.reference_spend = reference_spend

    def predict(self, sample: CustomerFeatures) -> Dict[str, Any]:
        # TODO: Execute prediction with circuit breaker fallback
        pass

    def evaluate_feature_drift_psi(self, current_spend: np.ndarray) -> Tuple[float, str]:
        # TODO: Calculate PSI drift against reference data
        pass
starter/test_section_project_an_ml_service_lib.py (1209 bytes) 
import pytest
import numpy as np
from examples.section_project_an_ml_service_lib import (
    CustomerFeatures, ProductionModelRegistry, DeployedMLService
)

def test_deployed_ml_service_lifecycle():
    reg = ProductionModelRegistry()
    reg.register_and_promote("churn_model", "v1.0", np.array([0.02, 0.01, 0.5]), -1.0, 0.85)

    service = DeployedMLService(reg, "churn_model")
    sample = CustomerFeatures(tenure_months=12.0, monthly_spend=100.0, support_tickets=1)
    res = service.predict(sample)

    assert "churn_probability" in res
    assert 0.0 <= res["churn_probability"] <= 1.0
    assert res["model_version"] == "v1.0"
    assert res["used_fallback"] is False

def test_deployed_ml_service_drift_detection():
    np.random.seed(42)
    reg = ProductionModelRegistry()
    reg.register_and_promote("churn_model", "v1.0", np.array([0.02, 0.01, 0.5]), -1.0, 0.85)

    service = DeployedMLService(reg, "churn_model")
    ref = np.random.normal(50.0, 10.0, 1000)
    service.set_reference_data(ref)

    # Shifted data
    drifted = np.random.normal(80.0, 20.0, 1000)
    psi, status = service.evaluate_feature_drift_psi(drifted)

    assert psi >= 0.20
    assert status == "SIGNIFICANT_DRIFT"
tests/run_tests.sh (227 bytes) 
#!/usr/bin/env bash
set -euo pipefail
echo "========================================"
echo "Running Day 196 Lab Test Suite"
echo "========================================"
pytest tests/ -v
echo "All tests passed successfully."
tests/test_section_project_an_ml_service_lib.py (1209 bytes) 
import pytest
import numpy as np
from examples.section_project_an_ml_service_lib import (
    CustomerFeatures, ProductionModelRegistry, DeployedMLService
)

def test_deployed_ml_service_lifecycle():
    reg = ProductionModelRegistry()
    reg.register_and_promote("churn_model", "v1.0", np.array([0.02, 0.01, 0.5]), -1.0, 0.85)

    service = DeployedMLService(reg, "churn_model")
    sample = CustomerFeatures(tenure_months=12.0, monthly_spend=100.0, support_tickets=1)
    res = service.predict(sample)

    assert "churn_probability" in res
    assert 0.0 <= res["churn_probability"] <= 1.0
    assert res["model_version"] == "v1.0"
    assert res["used_fallback"] is False

def test_deployed_ml_service_drift_detection():
    np.random.seed(42)
    reg = ProductionModelRegistry()
    reg.register_and_promote("churn_model", "v1.0", np.array([0.02, 0.01, 0.5]), -1.0, 0.85)

    service = DeployedMLService(reg, "churn_model")
    ref = np.random.normal(50.0, 10.0, 1000)
    service.set_reference_data(ref)

    # Shifted data
    drifted = np.random.normal(80.0, 20.0, 1000)
    psi, status = service.evaluate_feature_drift_psi(drifted)

    assert psi >= 0.20
    assert status == "SIGNIFICANT_DRIFT"

Troubleshooting

Troubleshooting: Day 196 - Section Project: An ML Service

Common Issues

  1. Unconfigured Reference Data:
    • Cause: Calling evaluate_feature_drift_psi() before set_reference_data().
    • Fix: Initialize reference baseline dataset before evaluating streaming drift.

Security notes

Security & Privacy: Day 196 - Section Project: An ML Service

Security Guidance

  • All calculations execute strictly on local CPU memory.