Capstone: End-to-End MLOps Lifecycle Pipeline

Over the previous twenty-three lessons you built the parts of a production machine learning platform: data validation, repeatable training pipelines, experiment tracking, evaluation gates, containers, a serving interface, telemetry, drift detection and attribution scores.

Each part solves one problem, but production failures happen between the parts: new training data that skips schema validation, a candidate model that bypasses the registry, a serving build with mismatched dependencies, or drift that quietly degrades predictions without triggering retraining.

This capstone connects the parts for StayWell, our subscription wellness app. You will trace the lifecycle from ingestion to model retirement, review the scorecard a live system needs, weigh custom code against managed platforms, and run a tested orchestrator that executes the whole lifecycle end to end.

The unified MLOps architecture

A reliable machine learning system operates as a closed-loop state machine. Fresh observations enter through strict validation gates, trigger isolated training runs, pass mathematical promotion checks, enter versioned registry storage, deploy to low-latency serving clusters, emit telemetry into real-time monitoring sinks, and alert retraining loops when drift metrics exceed safe bounds.

The diagram below illustrates this closed-loop topology, highlighting how data artifacts, code commits, and model metadata flow between operational stages.

The end-to-end MLOps lifecycle architecture

Notice three structural invariants in this architecture:

First, no stage writes directly to production without passing an automated validation barrier. Raw database dumps cannot be fed into training algorithms without passing schema and distribution assertions. Candidate weights cannot reach serving infrastructure without passing benchmark evaluation gates.

Second, every production artifact carries immutable cryptographic lineage. When an inference endpoint generates a churn probability of 0.82 for customer CUST-4029, the response metadata points to a registered model version. That registered version references the container image digest, git commit hash, hyperparameters, training script, and sha256 checksum of the training data snapshot.

Third, monitoring is an active control loop, not a passive dashboard. Observability signals (such as inference latency percentiles, error rates, and feature drift metrics) feed directly into decision engines. If the Population Stability Index on input billing charges exceeds 0.20, the platform automatically flags the model for retraining, notifies on-call engineers, and prevents stale decisions from compounding customer attrition.

Tracing the StayWell lifecycle phase by phase

To understand how these disciplines operate together under load, let us walk through each phase of the StayWell churn prediction system as it runs in production.

Phase 1: Data ingestion, validation, and feature storage

The lifecycle begins when StayWell's event stream records customer behavior: daily app logins, wellness challenge completions, billing events, and customer support tickets.

In our early lessons, raw CSV exports were loaded straight into exploratory notebooks. In the production architecture:

Schema validation: Ingestion workers execute assertions against every batch using schema specifications derived from our data validation lessons. If a billing update introduces negative monthly fees or null values in account tenure, the batch is routed to a quarantine dead-letter queue.
Feature stores and time travel: Validated features flow into an online feature store for millisecond serving and an offline store for training. Every record includes an event timestamp, guaranteeing that training datasets constructed today simulate the exact information available at the historical prediction moment without target leakage.
Artifact hashing: When training begins, the data extraction query generates an immutable snapshot. The pipeline computes a cryptographic hash (such as SHA-256) over the serialized feature set, permanently recording the exact data byte sequence used for model fitting.

Phase 2: Orchestrated training and experiment tracking

Training runs do not execute manually on developer laptops. They execute as containerized pipeline steps managed by workflow orchestrators like Kubeflow Pipelines (from the docs, not run; checked on 5 November 2026) or Prefect (from the docs, not run; checked on 5 November 2026).

During the run:

Hyperparameter logging: The trainer logs configuration values (learning rate, regularization penalties, tree depth) to a centralized tracking server such as MLflow (from the docs, not run; checked on 5 November 2026).
Calibration and objective evaluation: The model fits parameters against training data and computes loss metrics against an out-of-fold validation set. It logs log-loss, Brier calibration scores, and area under the ROC curve (AUROC).
Artifact generation: The pipeline serializes model weights along with the preprocessing pipeline, feature column names, and dependency locks into a standardized packaging format.

Phase 3: The automated evaluation gate and registry transition

A completed training run creates a candidate artifact, not a production model. Before any weights can be scheduled for live inference, the candidate must pass an automated gate.

The evaluation gate runs a battery of rigorous checks:

Absolute performance floor: The candidate must achieve an AUROC $\ge 0.70$ and a Brier score $\le 0.25$ on the validation holdout.
Regression prevention: The candidate must match or exceed the current production champion on slice-based metrics (for example, high-value enterprise accounts and new user cohorts).
Inference latency verification: The candidate is loaded into a test harness to confirm that p99 scoring latency remains under 20 milliseconds under simulated batch loads.

If all checks succeed, the orchestrator registers the artifact in the Model Registry, transitioning its lifecycle stage from Candidate to Staging. If any check fails, the pipeline aborts, tags the run as failed, and archives the artifact with diagnostic logs.

Phase 4: Containerization and CI/CD promotion

Moving code from Staging to Production is managed by continuous delivery pipelines:

Immutable container builds: The model serving server (such as BentoML (from the docs, not run; checked on 5 November 2026) or a high-performance FastAPI/Triton container) is built with pinned system dependencies and Python wheels.
Smoke testing: CI/CD runners start the container in a temporary environment, submit synthetic customer payloads, and verify that HTTP status codes, output probability types, and response schemas match the OpenAPI contract.
Canary or blue-green rollout: The deployment orchestrator routes a small fraction (e.g., 5%) of live production traffic to the new container while routing 95% to the existing champion. Error rates, latency distributions, and memory footprints are observed closely before completing full traffic migration.

Phase 5: Low-latency serving and feature alignment

In production, StayWell's retention microservice calls the scoring API whenever a customer visits the subscription settings page or requests account cancellation.

At run time:

Input validation: The serving gateway checks customer attributes against strict types and bounds before passing inputs to model logic.
Feature vector lookup: Fast cached reads retrieve customer tenure and historical support ticket counts from an in-memory key-value store.
Model scoring: The model computes the churn probability and classifies the customer into actionable risk tiers (Low, Medium, High).
Attribution payload: If customer support needs to understand why an account was flagged, local explanation routines (like SHAP surrogates) calculate additive feature attributions, explaining whether high monthly billing or recent unresolved support tickets drove the prediction.

Phase 6: Observability, drift detection, and automated retraining

Inference does not end when the HTTP response returns. Every request-response pair emits structured JSON logs to Kafka or an observability sink.

Telemetry ingestion: Stream workers track service metrics (request count, p50/p95/p99 latencies, error codes) and model metrics (predicted probability distributions, feature value summaries).
Statistical drift monitoring: Every twenty-four hours (or over sliding 1,000-request windows), a drift detector compares current feature distributions against training baseline distributions using metrics like the Population Stability Index (PSI) and Wasserstein Distance.
The retraining loop: When PSI exceeds warning thresholds ( $> 0.10$ ), the platform alerts the team. When PSI indicates severe distribution shift ( $> 0.20$ ), the orchestrator triggers an automated retraining job on newly ingested, validated data.

The production launch and operations scorecard

Before pushing any machine learning model to live production, seasoned engineering teams conduct an operational readiness review. Relying on good intentions or informal Slack confirmations inevitably leads to midnight outages, silent prediction corruption, or runaway cloud bills.

The scorecard below establishes clear, testable criteria across all operational dimensions.

Production launch and operations checklist

Let us examine each domain on the scorecard:

1. Data and features

Are all training features defined in version control alongside reproducible SQL or feature pipeline definitions?
Is there a deterministic validation gate preventing corrupted or schema-drifted batches from reaching training or inference?
Are historical point-in-time joins verified against target leakage?

2. Model quality and evaluation

Does the model pass an automated gate on both overall metrics and critical customer slices?
Are probabilities properly calibrated so that a score of 0.70 corresponds to an empirical 70% churn frequency?
Is there an automated baseline comparison ensuring the model beats naive heuristics (such as predicting the majority class or carrying over yesterday's rate)?

3. Infrastructure, CI/CD, and serving

Are container images reproducible, pinned by cryptographic SHA digest, and scanned for common vulnerabilities?
Are compute resource requests and limits (CPU cores, memory limits, GPU allocations) tested against expected load spikes?
Does the deployment harness support instant rollback to the previous model version within 60 seconds without downtime?

4. Observability and incident management

Are real-time alerts configured for p99 latency regressions, HTTP 5xx spikes, and prediction distribution shifts?
Is every prediction logged with its associated model version and input hash for auditability?
Does the engineering team possess a documented, practiced on-call playbook for rolling back corrupt model versions?

Honest platform limits: custom code versus managed enterprise stacks

Throughout this course, we built components using standard Python libraries, clean dataclasses, and standard mathematical formulations. This approach ensures you understand every moving piece without getting obscured by vendor-specific abstractions.

However, when scaling from a single StayWell churn model to hundreds of models across an enterprise organization, engineering teams face the question of custom code versus managed platforms. Here is an honest appraisal of what real enterprise tools provide:

Capability	Custom standard library approach	Enterprise managed platform (Databricks, AWS SageMaker, GCP Vertex AI) (from the docs, not run; checked on 5 November 2026)
Data scale	In-memory processing; constrained by single-node RAM (tens of gigabytes).	Distributed compute engines (Apache Spark, Ray, BigQuery, Snowflake) scaling across petabyte data lakes.
Workflow orchestration	Sequential script execution or simple cron runners.	Dynamic directed acyclic graphs (DAGs), automatic retries, backfills, distributed step caching, and spot instance management.
Model registry	File-system metadata or relational database rows.	Enterprise governance: role-based access control (RBAC), signed approval workflows, compliance audit trails, and multi-region replication.
Serving throughput	Python WSGI/ASGI endpoints handling hundreds of requests per second.	High-performance C++ runtimes (Triton Inference Server, vLLM, TensorRT) handling tens of thousands of QPS with GPU kernel optimizations and dynamic batching.
Drift and observability	Batch statistical scripts running on timed intervals.	Streaming observability engines with sub-minute alert latency, automated root-cause slice analysis, and enterprise PII masking.

Understanding these limits allows you to make pragmatic architectural choices. For startups, internal tools, or early-stage models, lightweight custom pipelines or simple Dockerized services often deliver faster iteration and lower infrastructure overhead. As traffic, compliance requirements, and model counts grow, adopting managed platforms solves the distributed systems challenges that custom code cannot easily sustain.

The complete executable MLOps pipeline

Below is the complete, self-contained Python implementation of StayWell's end-to-end MLOps pipeline. It coordinates data validation, gradient-based logistic training, quality evaluation gates, model registry lifecycle state transitions, live inference scoring, and distribution drift detection.

Save this script as end_to_end_pipeline.py and run it directly with standard Python 3. It includes unit tests for both the normal path and the failure safeguards.

end_to_end_pipeline.py

python

1"""
2StayWell End-to-End MLOps Pipeline Orchestrator.
3Demonstrates the full lifecycle: validation, training, gating, registry,
4serving, drift detection, and automated retraining triggers.
5Pure Python 3 standard library implementation.
6"""
7
8import math
9import random
10import hashlib
11import json
12import unittest
13from dataclasses import dataclass, field
14from typing import List, Dict, Any, Optional, Tuple
15from datetime import datetime, timezone
16
17
18# =====================================================================
19# 1. Data Schema & Ingestion Validation
20# =====================================================================
21
22@dataclass
23class CustomerRecord:
24    customer_id: str
25    tenure_months: float
26    monthly_charges: float
27    support_tickets: int
28    contract_is_annual: int  # 1 for annual contract, 0 for month-to-month
29    churned: Optional[int] = None  # 1 or 0 for training; None for inference
30
31
32class DataValidator:
33    """Validates raw customer batches against schema and statistical bounds."""
34
35    def __init__(self, min_records: int = 20):
36        self.min_records = min_records
37
38    def validate_batch(self, records: List[CustomerRecord], require_labels: bool = True) -> Tuple[bool, List[str]]:
39        errors = []
40        if len(records) < self.min_records:
41            errors.append(f"Batch size {len(records)} below minimum threshold {self.min_records}")
42
43        for idx, r in enumerate(records):
44            if r.tenure_months < 0.0 or r.tenure_months > 120.0:
45                errors.append(f"Record {idx} ({r.customer_id}): invalid tenure {r.tenure_months}")
46            if r.monthly_charges < 0.0 or r.monthly_charges > 500.0:
47                errors.append(f"Record {idx} ({r.customer_id}): invalid monthly charges {r.monthly_charges}")
48            if r.support_tickets < 0 or r.support_tickets > 50:
49                errors.append(f"Record {idx} ({r.customer_id}): invalid support tickets {r.support_tickets}")
50            if r.contract_is_annual not in (0, 1):
51                errors.append(f"Record {idx} ({r.customer_id}): invalid contract type {r.contract_is_annual}")
52            if require_labels and r.churned not in (0, 1):
53                errors.append(f"Record {idx} ({r.customer_id}): missing or invalid churn label {r.churned}")
54
55        return len(errors) == 0, errors
56
57
58# =====================================================================
59# 2. Model Training & Lineage Tracking
60# =====================================================================
61
62@dataclass
63class ModelArtifact:
64    model_id: str
65    version: str
66    weights: Dict[str, float]
67    intercept: float
68    created_at: str
69    train_data_hash: str
70    metrics: Dict[str, float] = field(default_factory=dict)
71    stage: str = "Candidate"  # Candidate, Staging, Production, Archived
72
73
74class ChurnModelTrainer:
75    """Trains a calibrated linear logistic model with reproducible data hashing."""
76
77    @staticmethod
78    def compute_dataset_hash(records: List[CustomerRecord]) -> str:
79        serialized = json.dumps([r.__dict__ for r in records], sort_keys=True)
80        return hashlib.sha256(serialized.encode("utf-8")).hexdigest()[:16]
81
82    @staticmethod
83    def sigmoid(z: float) -> float:
84        z_clamped = max(min(z, 20.0), -20.0)
85        return 1.0 / (1.0 + math.exp(-z_clamped))
86
87    def train(self, records: List[CustomerRecord], version: str,
88              learning_rate: float = 0.08, epochs: int = 150) -> ModelArtifact:
89        n = len(records)
90        w = {
91            "tenure_months": -0.05,
92            "monthly_charges": 0.02,
93            "support_tickets": 0.40,
94            "contract_is_annual": -1.20
95        }
96        b = -0.50
97
98        # Gradient descent optimization
99        for _ in range(epochs):
100            grad_w = {k: 0.0 for k in w}
101            grad_b = 0.0
102            for r in records:
103                z = (w["tenure_months"] * (r.tenure_months / 12.0) +
104                     w["monthly_charges"] * (r.monthly_charges / 50.0) +
105                     w["support_tickets"] * (r.support_tickets / 5.0) +
106                     w["contract_is_annual"] * r.contract_is_annual + b)
107                pred = self.sigmoid(z)
108                err = pred - r.churned
109
110                grad_w["tenure_months"] += err * (r.tenure_months / 12.0)
111                grad_w["monthly_charges"] += err * (r.monthly_charges / 50.0)
112                grad_w["support_tickets"] += err * (r.support_tickets / 5.0)
113                grad_w["contract_is_annual"] += err * r.contract_is_annual
114                grad_b += err
115
116            for k in w:
117                w[k] -= learning_rate * (grad_w[k] / n)
118            b -= learning_rate * (grad_b / n)
119
120        data_hash = self.compute_dataset_hash(records)
121        return ModelArtifact(
122            model_id=f"churn-logreg-{version}",
123            version=version,
124            weights=w,
125            intercept=b,
126            created_at=datetime.now(timezone.utc).isoformat(),
127            train_data_hash=data_hash
128        )
129
130
131# =====================================================================
132# 3. Model Evaluation Gate
133# =====================================================================
134
135class ModelEvaluator:
136    """Evaluates candidate models for discrimination (AUROC) and calibration."""
137
138    @staticmethod
139    def predict_proba(model: ModelArtifact, record: CustomerRecord) -> float:
140        z = (model.weights["tenure_months"] * (record.tenure_months / 12.0) +
141             model.weights["monthly_charges"] * (record.monthly_charges / 50.0) +
142             model.weights["support_tickets"] * (record.support_tickets / 5.0) +
143             model.weights["contract_is_annual"] * record.contract_is_annual +
144             model.intercept)
145        return ChurnModelTrainer.sigmoid(z)
146
147    def evaluate(self, model: ModelArtifact, test_records: List[CustomerRecord]) -> Dict[str, float]:
148        preds = [self.predict_proba(model, r) for r in test_records]
149        actuals = [r.churned for r in test_records]
150
151        # Brier score: mean squared calibration error
152        brier = sum((p - a) ** 2 for p, a in zip(preds, actuals)) / len(preds)
153
154        # Non-parametric AUROC computation via pairwise ranks
155        pos_scores = [p for p, a in zip(preds, actuals) if a == 1]
156        neg_scores = [p for p, a in zip(preds, actuals) if a == 0]
157
158        if not pos_scores or not neg_scores:
159            auc = 0.50
160        else:
161            wins = sum(1.0 if pos > neg else (0.5 if pos == neg else 0.0)
162                       for pos in pos_scores for neg in neg_scores)
163            auc = wins / (len(pos_scores) * len(neg_scores))
164
165        metrics = {
166            "auroc": round(auc, 4),
167            "brier_score": round(brier, 4)
168        }
169        model.metrics = metrics
170        return metrics
171
172
173class QualityGate:
174    """Automated gate that guards production from underperforming models."""
175
176    def __init__(self, min_auroc: float = 0.65, max_brier: float = 0.28):
177        self.min_auroc = min_auroc
178        self.max_brier = max_brier
179
180    def passes(self, metrics: Dict[str, float]) -> Tuple[bool, str]:
181        auroc = metrics.get("auroc", 0.0)
182        brier = metrics.get("brier_score", 1.0)
183
184        if auroc < self.min_auroc:
185            return False, f"AUROC {auroc:.4f} below floor {self.min_auroc:.4f}"
186        if brier > self.max_brier:
187            return False, f"Brier score {brier:.4f} exceeds limit {self.max_brier:.4f}"
188
189        return True, "Quality criteria met."
190
191
192# =====================================================================
193# 4. Model Registry & Lifecycle Management
194# =====================================================================
195
196class ModelRegistry:
197    """Central store managing model artifacts, lineages, and stage transitions."""
198
199    def __init__(self):
200        self._models: Dict[str, ModelArtifact] = {}
201        self._production_version: Optional[str] = None
202
203    def register_candidate(self, model: ModelArtifact) -> None:
204        model.stage = "Candidate"
205        self._models[model.version] = model
206
207    def promote_to_production(self, version: str) -> None:
208        if version not in self._models:
209            raise KeyError(f"Version {version} not found in registry")
210        # Archive current production champion
211        if self._production_version and self._production_version in self._models:
212            self._models[self._production_version].stage = "Archived"
213        # Promote candidate
214        self._models[version].stage = "Production"
215        self._production_version = version
216
217    def get_production_model(self) -> Optional[ModelArtifact]:
218        return self._models.get(self._production_version) if self._production_version else None
219
220    def get_model(self, version: str) -> Optional[ModelArtifact]:
221        return self._models.get(version)
222
223
224# =====================================================================
225# 5. Production Serving Service
226# =====================================================================
227
228class ServingService:
229    """Inference endpoint scoring live requests against the production model."""
230
231    def __init__(self, registry: ModelRegistry):
232        self.registry = registry
233
234    def score_customer(self, record: CustomerRecord) -> Dict[str, Any]:
235        prod_model = self.registry.get_production_model()
236        if not prod_model:
237            raise RuntimeError("No production model is currently active in registry")
238
239        prob = ModelEvaluator.predict_proba(prod_model, record)
240        risk_tier = "High" if prob >= 0.70 else ("Medium" if prob >= 0.35 else "Low")
241
242        return {
243            "customer_id": record.customer_id,
244            "churn_probability": round(prob, 4),
245            "risk_tier": risk_tier,
246            "model_version": prod_model.version,
247            "scored_at": datetime.now(timezone.utc).isoformat()
248        }
249
250
251# =====================================================================
252# 6. Drift Detection & Monitoring
253# =====================================================================
254
255class DriftDetector:
256    """Detects statistical feature shifts using Population Stability Index (PSI)."""
257
258    @staticmethod
259    def calculate_psi(baseline: List[float], current: List[float], bins: int = 5) -> float:
260        if not baseline or not current:
261            return 0.0
262
263        min_val = min(min(baseline), min(current))
264        max_val = max(max(baseline), max(current))
265        if min_val == max_val:
266            return 0.0
267
268        bin_width = (max_val - min_val) / bins
269        breakpoints = [min_val + i * bin_width for i in range(bins + 1)]
270        breakpoints[-1] += 1e-6
271
272        def get_distribution(data: List[float]) -> List[float]:
273            counts = [0] * bins
274            for x in data:
275                for b in range(bins):
276                    if breakpoints[b] <= x < breakpoints[b + 1]:
277                        counts[b] += 1
278                        break
279            # Add Laplace smoothing to prevent numerical zero division
280            return [(c + 1e-4) / (len(data) + 1e-4 * bins) for c in counts]
281
282        b_dist = get_distribution(baseline)
283        c_dist = get_distribution(current)
284
285        psi = 0.0
286        for b_prop, c_prop in zip(b_dist, c_dist):
287            psi += (c_prop - b_prop) * math.log(c_prop / b_prop)
288        return round(psi, 4)
289
290
291# =====================================================================
292# 7. End-to-End Pipeline Orchestrator
293# =====================================================================
294
295class StayWellMLOpsOrchestrator:
296    """Glues the entire lifecycle together from raw data to production rollout."""
297
298    def __init__(self, validator: DataValidator, trainer: ChurnModelTrainer,
299                 evaluator: ModelEvaluator, gate: QualityGate, registry: ModelRegistry):
300        self.validator = validator
301        self.trainer = trainer
302        self.evaluator = evaluator
303        self.gate = gate
304        self.registry = registry
305
306    def run_training_pipeline(self, train_data: List[CustomerRecord],
307                              val_data: List[CustomerRecord], version: str) -> Tuple[bool, str]:
308        # Step 1: Ingestion validation
309        valid, errs = self.validator.validate_batch(train_data, require_labels=True)
310        if not valid:
311            return False, f"Data validation failed: {errs[:2]}"
312
313        # Step 2: Training and artifact creation
314        candidate = self.trainer.train(train_data, version=version)
315        self.registry.register_candidate(candidate)
316
317        # Step 3: Evaluation against holdout data
318        metrics = self.evaluator.evaluate(candidate, val_data)
319
320        # Step 4: Quality gate assessment
321        passed, reason = self.gate.passes(metrics)
322        if not passed:
323            return False, f"Gate rejected version {version}: {reason}"
324
325        # Step 5: Promotion to active production champion
326        self.registry.promote_to_production(version)
327        return True, f"Successfully deployed version {version} (AUROC: {metrics['auroc']})"
328
329
330# =====================================================================
331# 8. Unit Test Suite
332# =====================================================================
333
334def generate_synthetic_cohort(n: int = 100, seed: int = 42) -> List[CustomerRecord]:
335    random.seed(seed)
336    records = []
337    for i in range(n):
338        tenure = random.uniform(1.0, 72.0)
339        charge = random.uniform(30.0, 110.0)
340        tickets = random.randint(0, 8)
341        annual = 1 if random.random() > 0.6 else 0
342
343        # Underlying churn likelihood formula
344        z = -0.05 * (tenure / 12.0) + 0.04 * (charge / 50.0) + 0.45 * (tickets / 5.0) - 1.2 * annual - 0.3
345        prob = 1.0 / (1.0 + math.exp(-z))
346        churn = 1 if random.random() < prob else 0
347
348        records.append(CustomerRecord(
349            customer_id=f"CUST-{i:04d}",
350            tenure_months=round(tenure, 1),
351            monthly_charges=round(charge, 2),
352            support_tickets=tickets,
353            contract_is_annual=annual,
354            churned=churn
355        ))
356    return records
357
358
359class TestEndToEndMLOpsPipeline(unittest.TestCase):
360    def setUp(self):
361        self.validator = DataValidator(min_records=20)
362        self.trainer = ChurnModelTrainer()
363        self.evaluator = ModelEvaluator()
364        self.gate = QualityGate(min_auroc=0.65, max_brier=0.28)
365        self.registry = ModelRegistry()
366        self.orchestrator = StayWellMLOpsOrchestrator(
367            self.validator, self.trainer, self.evaluator, self.gate, self.registry
368        )
369        self.serving = ServingService(self.registry)
370
371    def test_full_lifecycle_happy_path(self):
372        train_data = generate_synthetic_cohort(120, seed=42)
373        val_data = generate_synthetic_cohort(40, seed=99)
374
375        success, msg = self.orchestrator.run_training_pipeline(train_data, val_data, version="1.0.0")
376        self.assertTrue(success)
377        self.assertIn("Successfully deployed", msg)
378
379        # Confirm registry state
380        prod = self.registry.get_production_model()
381        self.assertIsNotNone(prod)
382        self.assertEqual(prod.version, "1.0.0")
383        self.assertEqual(prod.stage, "Production")
384
385        # Confirm serving endpoint functions correctly
386        customer = CustomerRecord(
387            customer_id="CUST-LIVE-1",
388            tenure_months=2.0,
389            monthly_charges=120.0,
390            support_tickets=6,
391            contract_is_annual=0
392        )
393        score = self.serving.score_customer(customer)
394        self.assertEqual(score["customer_id"], "CUST-LIVE-1")
395        self.assertGreaterEqual(score["churn_probability"], 0.0)
396        self.assertLessEqual(score["churn_probability"], 1.0)
397        self.assertEqual(score["risk_tier"], "High")
398
399    def test_data_validation_rejection(self):
400        corrupted_data = [
401            CustomerRecord("CUST-BAD-1", -5.0, 50.0, 2, 0, 1),
402            CustomerRecord("CUST-BAD-2", 12.0, 999.0, 0, 1, 0)
403        ]
404        success, msg = self.orchestrator.run_training_pipeline(corrupted_data, corrupted_data, version="0.1.0")
405        self.assertFalse(success)
406        self.assertIn("Data validation failed", msg)
407        self.assertIsNone(self.registry.get_production_model())
408
409    def test_quality_gate_rejection_preserves_production(self):
410        # 1. First deploy a healthy baseline model
411        train_data = generate_synthetic_cohort(120, seed=42)
412        val_data = generate_synthetic_cohort(40, seed=99)
413        self.orchestrator.run_training_pipeline(train_data, val_data, version="1.0.0")
414
415        # 2. Attempt to deploy a second version under an unattainable quality gate
416        strict_gate = QualityGate(min_auroc=0.99, max_brier=0.01)
417        strict_orchestrator = StayWellMLOpsOrchestrator(
418            self.validator, self.trainer, self.evaluator, strict_gate, self.registry
419        )
420        success, msg = strict_orchestrator.run_training_pipeline(train_data, val_data, version="2.0.0")
421        self.assertFalse(success)
422        self.assertIn("Gate rejected", msg)
423
424        # 3. Assert production model remains untouched at version 1.0.0
425        prod = self.registry.get_production_model()
426        self.assertEqual(prod.version, "1.0.0")
427        self.assertEqual(prod.stage, "Production")
428
429        # 4. Verify candidate 2.0.0 exists in registry but remained in Candidate stage
430        v2 = self.registry.get_model("2.0.0")
431        self.assertEqual(v2.stage, "Candidate")
432
433    def test_drift_detection_and_retraining_trigger(self):
434        baseline_charges = [r.monthly_charges for r in generate_synthetic_cohort(100, seed=42)]
435        random.seed(123)
436        # Simulate pricing restructuring that doubles customer bills
437        shifted_charges = [random.uniform(150.0, 250.0) for _ in range(100)]
438
439        detector = DriftDetector()
440        psi = detector.calculate_psi(baseline_charges, shifted_charges)
441
442        # PSI above 0.20 indicates significant distribution shift
443        self.assertGreater(psi, 0.20)
444        retraining_required = psi > 0.20
445        self.assertTrue(retraining_required)
446
447
448if __name__ == "__main__":
449    unittest.main()

Where to go from here: expanding your craft

Completing this course equips you with the fundamental engineering disciplines that separate hobbyist data science from industrial software engineering. You know how to validate inputs, trace lineage, protect production behind automated gates, serve low-latency inferences, monitor drift, and explain predictions.

Machine learning engineering is an expansive ecosystem. Here is how you can continue building upon these foundations across our curriculum:

Autonomous reasoning and multi-step workflows: Explore our course on AI Agents to understand how language models plan actions, call APIs, and maintain conversational memory. To master operational tracing for non-deterministic agents, study Observability and Tracing.
Rapid application delivery: Connect your models to customer-facing frontends using techniques from Deploy Your App and Monitor and Maintain.
Product strategy and architecture: Learn how to select cloud infrastructure, design multi-tenant architectures, and control unit economics in Choose Where to Deploy.
Prompt engineering and evaluation: If your systems integrate generative models alongside classical predictive algorithms, study production prompt versioning and regression testing in Prompts in Production.

Mistakes beginners make

Deploying pipelines without cryptographic lineage. Teams often track Git commits for code but fail to hash the exact data snapshot used during training. When a production regression occurs weeks later, developers cannot reproduce the model because underlying database tables changed or records were updated.

Fix: Generate and record a SHA-256 hash of the exact training dataset bytes in the model registry metadata alongside the Git commit hash and hyperparameters.
Allowing candidate models to self-promote without automated gates. Relying on human sign-off or assuming that successful model convergence implies production readiness allows flawed models into production. A model with great aggregate accuracy can catastrophically degrade high-value customer segments.

Fix: Enforce automated quality gates in your deployment pipeline that evaluate candidate weights against validation floors, calibration metrics, latency budgets, and slice-level regressions before updating registry pointers.
Treating monitoring as a passive reporting dashboard. Setting up Grafana or Datadog dashboards without automated alert routing means distribution shifts and latency spikes go unnoticed until customers complain or churn spikes.

Fix: Configure automated drift detection jobs that compute statistical distance metrics (such as PSI or Wasserstein distance) on scheduled intervals and trigger automated retraining pipelines or on-call alerts when thresholds are crossed.
Coupling training and serving code in a single monolith. Packaging heavy training libraries (such as PyTorch, scikit-learn, CUDA toolkits, and compilation utilities) into the runtime inference container balloons image sizes, slows down autoscaling cold starts, and introduces security vulnerabilities.

Fix: Decouple training and serving environments. Export trained models into standardized serialization formats (such as ONNX, TorchScript, or lightweight picklers) and deploy them inside lean, dedicated runtime containers containing only inference dependencies.

Recap and try it yourself

In this capstone, you assembled the end-to-end MLOps lifecycle:

Ingestion validation prevents silent schema corruption and dirty inputs from reaching downstream models.
Automated evaluation gates protect production by comparing candidates against hard metric floors and regression baselines.
The model registry acts as the single source of truth, managing stage transitions and immutable lineage metadata.
Low-latency serving decouples inference runtimes from heavy training frameworks.
Statistical drift detectors turn observability from a passive visualization tool into an active, self-healing trigger.

To deepen your hands-on mastery, try these exercises with the code above:

Add slice-based gating: Modify QualityGate and ModelEvaluator to compute separate AUROC scores for month-to-month customers versus annual contract holders. Require that both slices meet individual minimum performance floors before promoting the candidate.
Implement an automated rollback handler: Extend StayWellMLOpsOrchestrator with a rollback_to_previous_version() method that re-activates the previously archived model if the serving endpoint detects an elevated error rate.
Simulate covariate drift and retraining: Write a loop that submits 500 drifting customer records, triggers the drift detector, extracts the recent drifting cohort, fits a version="1.1.0" candidate, and validates that the retrained model successfully updates production.

Capstone: An End-to-End MLOps Pipeline