MLOps & Model Monitoring

Version Control, MLflow, and Data Drift

Building a model is only 20% of the work. Deploying, monitoring, and maintaining it in production is the other 80%. **MLOps** (Machine Learning Operations) brings DevOps principles to ML. For **Visa**, a fraud model that was 98% accurate last month might drop to 85% this month due to **Data Drift**—fraudsters change tactics.

In this chapter, we explore **Model Versioning** with MLflow, **Data Drift Detection**, and **CI/CD Pipelines** for automated model retraining and deployment.

Model Versioning with MLflow

**MLflow** tracks every experiment: which dataset was used, which hyperparameters were set, and what metrics were achieved. For **Walmart**, this means they can compare last week's demand model (v2.3) with this week's retrained version (v2.4) and decide which one to promote to production.

PythonRuns entirely in your browser — nothing is sent to a server.

# Walmart Demand Model: MLflow Experiment Tracking
class MLflowTracker:
    def __init__(self):
        self.experiments = []

    def log_run(self, model_name, version, params, metrics):
        run = {
            "model": model_name,
            "version": version,
            "params": params,
            "metrics": metrics
        }
        self.experiments.append(run)
        print("MLFLOW LOG: {0} v{1}".format(model_name, version))
        print(" - Params : {0}".format(params))
        print(" - Metrics: {0}".format(metrics))

    def compare_versions(self):
        print("\nMODEL COMPARISON:")
        print("-" * 50)
        best = None
        for exp in self.experiments:
            mae = exp["metrics"].get("mae", 999)
            status = ""
            if best is None or mae < best["metrics"]["mae"]:
                best = exp
                status = " <- BEST"
            print(" v{0}: MAE={1}, R2={2}{3}".format(
                exp["version"], mae, exp["metrics"].get("r2", "N/A"), status))
        
        print("\nPROMOTE TO PRODUCTION: v{0}".format(best["version"]))

# Scenario: Comparing demand model versions
tracker = MLflowTracker()
tracker.log_run("demand_forecast", "2.3", {"lr": 0.01, "depth": 5}, {"mae": 142, "r2": 0.89})
tracker.log_run("demand_forecast", "2.4", {"lr": 0.05, "depth": 3}, {"mae": 118, "r2": 0.93})
tracker.log_run("demand_forecast", "2.5", {"lr": 0.1, "depth": 4}, {"mae": 125, "r2": 0.91})
tracker.compare_versions()

# Walmart Demand Model: MLflow Experiment Tracking
class MLflowTracker:
    def __init__(self):
        self.experiments = []

    def log_run(self, model_name, version, params, metrics):
        run = {
            "model": model_name,
            "version": version,
            "params": params,
            "metrics": metrics
        }
        self.experiments.append(run)
        print("MLFLOW LOG: {0} v{1}".format(model_name, version))
        print(" - Params : {0}".format(params))
        print(" - Metrics: {0}".format(metrics))

    def compare_versions(self):
        print("\nMODEL COMPARISON:")
        print("-" * 50)
        best = None
        for exp in self.experiments:
            mae = exp["metrics"].get("mae", 999)
            status = ""
            if best is None or mae < best["metrics"]["mae"]:
                best = exp
                status = " <- BEST"
            print(" v{0}: MAE={1}, R2={2}{3}".format(
                exp["version"], mae, exp["metrics"].get("r2", "N/A"), status))
        
        print("\nPROMOTE TO PRODUCTION: v{0}".format(best["version"]))

# Scenario: Comparing demand model versions
tracker = MLflowTracker()
tracker.log_run("demand_forecast", "2.3", {"lr": 0.01, "depth": 5}, {"mae": 142, "r2": 0.89})
tracker.log_run("demand_forecast", "2.4", {"lr": 0.05, "depth": 3}, {"mae": 118, "r2": 0.93})
tracker.log_run("demand_forecast", "2.5", {"lr": 0.1, "depth": 4}, {"mae": 125, "r2": 0.91})
tracker.compare_versions()

Data Drift Detection

**Data Drift** occurs when the real-world data distribution changes from what the model was trained on. For **Visa**, this happens when fraudsters adopt new attack vectors. A model trained on "large single purchases" might miss "thousands of micro-transactions." Drift detection compares incoming data distributions against the training baseline.

PythonRuns entirely in your browser — nothing is sent to a server.

# Visa Fraud Model: Data Drift Detection
class DriftDetector:
    def __init__(self, baseline_stats):
        self.baseline = baseline_stats # {feature: (mean, std)}

    def check_drift(self, current_stats, threshold=2.0):
        print("DATA DRIFT ANALYSIS:")
        print("=" * 50)
        drifted_features = []
        
        for feature in self.baseline:
            b_mean, b_std = self.baseline[feature]
            c_mean, c_std = current_stats.get(feature, (b_mean, b_std))
            
            # Z-score drift: how many std devs has the mean shifted?
            drift_score = abs(c_mean - b_mean) / b_std if b_std > 0 else 0
            status = "DRIFT" if drift_score > threshold else "OK"
            
            if status == "DRIFT":
                drifted_features.append(feature)
            
            print(" {0}: Baseline={1:.1f}, Current={2:.1f}, Drift={3:.2f} [{4}]".format(
                feature, b_mean, c_mean, drift_score, status))
        
        if drifted_features:
            print("\nALERT: {0} feature(s) drifted! Retrain recommended.".format(len(drifted_features)))
        else:
            print("\nAll features within baseline. Model is stable.")

# Scenario: Visa's fraud model monthly health check
baseline = {"avg_amount": (150, 50), "tx_per_day": (5, 2), "foreign_ratio": (0.1, 0.05)}
current = {"avg_amount": (145, 55), "tx_per_day": (12, 3), "foreign_ratio": (0.35, 0.1)}
detector = DriftDetector(baseline)
detector.check_drift(current)

# Visa Fraud Model: Data Drift Detection
class DriftDetector:
    def __init__(self, baseline_stats):
        self.baseline = baseline_stats # {feature: (mean, std)}

    def check_drift(self, current_stats, threshold=2.0):
        print("DATA DRIFT ANALYSIS:")
        print("=" * 50)
        drifted_features = []
        
        for feature in self.baseline:
            b_mean, b_std = self.baseline[feature]
            c_mean, c_std = current_stats.get(feature, (b_mean, b_std))
            
            # Z-score drift: how many std devs has the mean shifted?
            drift_score = abs(c_mean - b_mean) / b_std if b_std > 0 else 0
            status = "DRIFT" if drift_score > threshold else "OK"
            
            if status == "DRIFT":
                drifted_features.append(feature)
            
            print(" {0}: Baseline={1:.1f}, Current={2:.1f}, Drift={3:.2f} [{4}]".format(
                feature, b_mean, c_mean, drift_score, status))
        
        if drifted_features:
            print("\nALERT: {0} feature(s) drifted! Retrain recommended.".format(len(drifted_features)))
        else:
            print("\nAll features within baseline. Model is stable.")

# Scenario: Visa's fraud model monthly health check
baseline = {"avg_amount": (150, 50), "tx_per_day": (5, 2), "foreign_ratio": (0.1, 0.05)}
current = {"avg_amount": (145, 55), "tx_per_day": (12, 3), "foreign_ratio": (0.35, 0.1)}
detector = DriftDetector(baseline)
detector.check_drift(current)

CI/CD for Machine Learning

**CI/CD (Continuous Integration / Continuous Deployment)** automates the testing and deployment of ML models. For **Walmart**, when a data scientist pushes a new model version, the pipeline automatically validates it against a test set, checks for performance regression, and deploys it if it passes—all without manual intervention.

PythonRuns entirely in your browser — nothing is sent to a server.

# Walmart ML Pipeline: CI/CD Simulation
class MLCICDPipeline:
    def __init__(self, model_name):
        self.model = model_name
        self.stages = []

    def run_stage(self, stage_name, check_func):
        result = check_func()
        status = "PASS" if result else "FAIL"
        self.stages.append((stage_name, status))
        print("[{0}] {1}".format(status, stage_name))
        return result

    def deploy(self):
        print("ML CI/CD PIPELINE: {0}".format(self.model))
        print("=" * 50)
        
        # Stage 1: Data validation
        s1 = self.run_stage("Data Validation (schema + null check)", lambda: True)
        
        # Stage 2: Model training
        s2 = self.run_stage("Model Training (v2.5)", lambda: True) if s1 else False
        
        # Stage 3: Performance check
        s3 = self.run_stage("Performance Gate (MAE < 150)", lambda: True) if s2 else False
        
        # Stage 4: A/B test simulation
        s4 = self.run_stage("Shadow Deployment (A/B test vs v2.4)", lambda: True) if s3 else False
        
        # Stage 5: Production deployment
        if s4:
            self.run_stage("PRODUCTION DEPLOYMENT", lambda: True)
            print("\nModel {0} v2.5 is now LIVE!".format(self.model))
        else:
            print("\nPIPELINE FAILED. Model not deployed.")

# Scenario: Walmart's automated model deployment
pipeline = MLCICDPipeline("demand_forecast")
pipeline.deploy()

# Walmart ML Pipeline: CI/CD Simulation
class MLCICDPipeline:
    def __init__(self, model_name):
        self.model = model_name
        self.stages = []

    def run_stage(self, stage_name, check_func):
        result = check_func()
        status = "PASS" if result else "FAIL"
        self.stages.append((stage_name, status))
        print("[{0}] {1}".format(status, stage_name))
        return result

    def deploy(self):
        print("ML CI/CD PIPELINE: {0}".format(self.model))
        print("=" * 50)
        
        # Stage 1: Data validation
        s1 = self.run_stage("Data Validation (schema + null check)", lambda: True)
        
        # Stage 2: Model training
        s2 = self.run_stage("Model Training (v2.5)", lambda: True) if s1 else False
        
        # Stage 3: Performance check
        s3 = self.run_stage("Performance Gate (MAE < 150)", lambda: True) if s2 else False
        
        # Stage 4: A/B test simulation
        s4 = self.run_stage("Shadow Deployment (A/B test vs v2.4)", lambda: True) if s3 else False
        
        # Stage 5: Production deployment
        if s4:
            self.run_stage("PRODUCTION DEPLOYMENT", lambda: True)
            print("\nModel {0} v2.5 is now LIVE!".format(self.model))
        else:
            print("\nPIPELINE FAILED. Model not deployed.")

# Scenario: Walmart's automated model deployment
pipeline = MLCICDPipeline("demand_forecast")
pipeline.deploy()

Practice Questions

Question 1

What is 'Data Drift' and why is it dangerous for Visa's fraud model?

It means the database is moving to a new server
It means the real-world data distribution has changed from what the model was trained on, causing accuracy to degrade silently
It means the model is getting faster
It only happens with linear models

Question 2

Why is MLflow valuable for Walmart's data science team?

It makes Python faster
It tracks every experiment's parameters, datasets, and metrics, enabling reproducibility and informed decisions about which model version to promote
It replaces the need for training data
It only works with XGBoost