MLOps & Model Deployment

MLOps Lifecycle:

┌─────────────────────────────────────────────────────────────────┐
│                        MLOps Pipeline                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  1. DATA COLLECTION                                             │
│     ├─ Collect training data                                    │
│     ├─ Validate data quality                                    │
│     └─ Version datasets                                         │
│                                                                 │
│  2. MODEL DEVELOPMENT                                           │
│     ├─ Feature engineering                                      │
│     ├─ Model training                                           │
│     ├─ Hyperparameter tuning                                    │
│     └─ Model evaluation                                         │
│                                                                 │
│  3. MODEL VALIDATION                                            │
│     ├─ Performance metrics                                      │
│     ├─ A/B testing                                              │
│     └─ Business metrics validation                              │
│                                                                 │
│  4. DEPLOYMENT                                                  │
│     ├─ Package model                                            │
│     ├─ Deploy to production                                     │
│     ├─ Canary/blue-green deployment                             │
│     └─ Health checks                                            │
│                                                                 │
│  5. MONITORING                                                  │
│     ├─ Performance tracking                                     │
│     ├─ Data drift detection                                     │
│     ├─ Model drift detection                                    │
│     └─ Cost monitoring                                          │
│                                                                 │
│  6. MAINTENANCE                                                 │
│     ├─ Model retraining                                         │
│     ├─ Incident response                                        │
│     ├─ Rollback if needed                                       │
│     └─ Continuous improvement                                   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

# model_monitoring.py - Comprehensive monitoring system

import numpy as np
import pandas as pd
from scipy import stats
from typing import Dict, Tuple
from datetime import datetime, timedelta
import logging


class ModelMonitor:
    """Monitor model performance and detect drift."""

    def __init__(self, training_stats: dict):
        """
        Initialize with baseline statistics from training data.

        Args:
            training_stats: Dict with feature means, stds, distributions
        """
        self.training_stats = training_stats
        self.logger = logging.getLogger(__name__)

    def detect_data_drift(
        self,
        production_data: pd.DataFrame,
        threshold: float = 0.05
    ) -> Dict[str, any]:
        """
        Detect data drift using statistical tests.

        Args:
            production_data: Recent production data
            threshold: P-value threshold for drift detection

        Returns:
            Drift report with alerts
        """
        drift_report = {
            'timestamp': datetime.now().isoformat(),
            'samples_analyzed': len(production_data),
            'features_with_drift': [],
            'drift_scores': {}
        }

        for feature in production_data.columns:
            # Get training distribution statistics
            if feature not in self.training_stats:
                continue

            train_mean = self.training_stats[feature]['mean']
            train_std = self.training_stats[feature]['std']

            # Production statistics
            prod_data = production_data[feature].dropna()
            prod_mean = prod_data.mean()
            prod_std = prod_data.std()

            # Kolmogorov-Smirnov test
            # Tests if distributions are significantly different
            training_samples = np.random.normal(
                train_mean,
                train_std,
                size=len(prod_data)
            )

            ks_statistic, p_value = stats.ks_2samp(
                training_samples,
                prod_data
            )

            drift_report['drift_scores'][feature] = {
                'train_mean': float(train_mean),
                'prod_mean': float(prod_mean),
                'mean_shift_pct': abs(prod_mean - train_mean) / abs(train_mean) * 100,
                'ks_statistic': float(ks_statistic),
                'p_value': float(p_value),
                'has_drift': p_value < threshold
            }

            if p_value < threshold:
                drift_report['features_with_drift'].append(feature)
                self.logger.warning(
                    f"Data drift detected in '{feature}': "
                    f"p-value={p_value:.4f}, "
                    f"mean shift={drift_report['drift_scores'][feature]['mean_shift_pct']:.1f}%"
                )

        # Overall drift assessment
        drift_report['has_significant_drift'] = len(drift_report['features_with_drift']) > 0
        drift_report['drift_severity'] = self._assess_drift_severity(drift_report)

        return drift_report

    def _assess_drift_severity(self, drift_report: dict) -> str:
        """Assess overall drift severity."""
        num_drifted = len(drift_report['features_with_drift'])

        if num_drifted == 0:
            return "none"
        elif num_drifted <= 2:
            return "low"
        elif num_drifted <= 5:
            return "medium"
        else:
            return "high"

    def detect_model_drift(
        self,
        recent_metrics: pd.DataFrame,
        baseline_metric: str = 'accuracy',
        threshold_pct: float = 5.0
    ) -> Dict[str, any]:
        """
        Detect model performance drift.

        Args:
            recent_metrics: DataFrame with timestamp and metric columns
            baseline_metric: Metric to monitor (accuracy, precision, etc.)
            threshold_pct: Alert if metric drops by this percentage

        Returns:
            Performance drift report
        """
        if len(recent_metrics) < 2:
            return {"status": "insufficient_data"}

        # Calculate baseline (e.g., 30-day average)
        cutoff_date = datetime.now() - timedelta(days=30)
        baseline_data = recent_metrics[
            recent_metrics['timestamp'] < cutoff_date
        ]

        if len(baseline_data) == 0:
            baseline_performance = recent_metrics[baseline_metric].mean()
        else:
            baseline_performance = baseline_data[baseline_metric].mean()

        # Recent performance (last 7 days)
        recent_cutoff = datetime.now() - timedelta(days=7)
        recent_data = recent_metrics[
            recent_metrics['timestamp'] >= recent_cutoff
        ]

        if len(recent_data) == 0:
            return {"status": "no_recent_data"}

        recent_performance = recent_data[baseline_metric].mean()

        # Calculate drift
        performance_change_pct = (
            (recent_performance - baseline_performance) / baseline_performance * 100
        )

        has_drift = abs(performance_change_pct) > threshold_pct
        is_degradation = performance_change_pct < 0

        report = {
            'timestamp': datetime.now().isoformat(),
            'metric': baseline_metric,
            'baseline_performance': float(baseline_performance),
            'recent_performance': float(recent_performance),
            'change_pct': float(performance_change_pct),
            'has_drift': has_drift,
            'is_degradation': is_degradation,
            'severity': 'critical' if is_degradation and abs(performance_change_pct) > 10 else
                       'warning' if is_degradation and abs(performance_change_pct) > 5 else
                       'info'
        }

        if has_drift:
            direction = "degraded" if is_degradation else "improved"
            self.logger.warning(
                f"Model performance {direction} by {abs(performance_change_pct):.1f}%: "
                f"{baseline_performance:.3f} → {recent_performance:.3f}"
            )

        return report


# Example usage
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    # Training data statistics (save these after training)
    training_stats = {
        'age': {'mean': 35.5, 'std': 12.3},
        'income': {'mean': 55000, 'std': 25000},
        'clicks': {'mean': 45, 'std': 20}
    }

    # Initialize monitor
    monitor = ModelMonitor(training_stats)

    # Simulate production data (with drift)
    production_data = pd.DataFrame({
        'age': np.random.normal(40, 15, 1000),  # Drifted: older users
        'income': np.random.normal(55000, 25000, 1000),  # No drift
        'clicks': np.random.normal(30, 18, 1000)  # Drifted: fewer clicks
    })

    # Detect data drift
    drift_report = monitor.detect_data_drift(production_data)
    print(f"\nData Drift Report:")
    print(f"  Severity: {drift_report['drift_severity']}")
    print(f"  Features with drift: {drift_report['features_with_drift']}")

    # Simulate performance metrics
    metrics = pd.DataFrame({
        'timestamp': pd.date_range(end=datetime.now(), periods=60, freq='D'),
        'accuracy': np.concatenate([
            np.random.normal(0.85, 0.02, 40),  # Baseline
            np.random.normal(0.78, 0.03, 20)   # Recent degradation
        ])
    })

    # Detect model drift
    perf_report = monitor.detect_model_drift(metrics, baseline_metric='accuracy')
    print(f"\nModel Performance Drift:")
    print(f"  Severity: {perf_report.get('severity', 'N/A')}")
    print(f"  Change: {perf_report.get('change_pct', 0):.1f}%")
    print(f"  Action needed: {perf_report.get('has_drift', False)}")

# .github/workflows/ml-pipeline.yml - GitHub Actions CI/CD

name: ML Model CI/CD

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]
  schedule:
    - cron: '0 2 * * 0'  # Weekly retraining on Sunday 2 AM

env:
  MODEL_VERSION: ${{ github.sha }}
  PYTHON_VERSION: '3.11'

jobs:
  data-validation:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install dependencies
        run: |
          pip install great-expectations pandas

      - name: Validate training data
        run: |
          python scripts/validate_data.py \
            --data-path data/training.csv \
            --expectations expectations/training_data.json

      - name: Upload data profile
        uses: actions/upload-artifact@v4
        with:
          name: data-profile
          path: reports/data_profile.html

  train-model:
    needs: data-validation
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install dependencies
        run: |
          pip install -r requirements.txt

      - name: Train model
        run: |
          python train.py \
            --data data/training.csv \
            --output models/model_${{ env.MODEL_VERSION }}.joblib \
            --metrics-output metrics.json

      - name: Upload model artifact
        uses: actions/upload-artifact@v4
        with:
          name: trained-model
          path: |
            models/model_${{ env.MODEL_VERSION }}.joblib
            metrics.json

  test-model:
    needs: train-model
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Download model
        uses: actions/download-artifact@v4
        with:
          name: trained-model

      - name: Run model tests
        run: |
          # Unit tests for model code
          pytest tests/test_model.py -v

          # Model performance tests
          python tests/test_model_performance.py \
            --model models/model_${{ env.MODEL_VERSION }}.joblib \
            --test-data data/test.csv \
            --min-accuracy 0.80

      - name: Check model size
        run: |
          SIZE=$(stat -f%z models/model_${{ env.MODEL_VERSION }}.joblib)
          MAX_SIZE=$((100 * 1024 * 1024))  # 100MB
          if [ $SIZE -gt $MAX_SIZE ]; then
            echo "Model too large: ${SIZE} bytes"
            exit 1
          fi

  evaluate-model:
    needs: test-model
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Download model
        uses: actions/download-artifact@v4
        with:
          name: trained-model

      - name: Compare with production model
        run: |
          # Download current production model
          aws s3 cp s3://models/production/model.joblib models/prod_model.joblib

          # Compare metrics
          python scripts/compare_models.py \
            --new-model models/model_${{ env.MODEL_VERSION }}.joblib \
            --prod-model models/prod_model.joblib \
            --test-data data/test.csv \
            --output comparison.json

      - name: Check if better than production
        run: |
          # Parse comparison results
          NEW_ACC=$(jq '.new_model.accuracy' comparison.json)
          PROD_ACC=$(jq '.prod_model.accuracy' comparison.json)

          echo "New model accuracy: $NEW_ACC"
          echo "Production model accuracy: $PROD_ACC"

          # Require 2% improvement to deploy
          if (( $(echo "$NEW_ACC < $PROD_ACC * 1.02" | bc -l) )); then
            echo "New model not significantly better. Skipping deployment."
            exit 1
          fi

  deploy-staging:
    needs: evaluate-model
    runs-on: ubuntu-latest
    environment: staging
    steps:
      - uses: actions/checkout@v4

      - name: Download model
        uses: actions/download-artifact@v4
        with:
          name: trained-model

      - name: Upload to model registry
        run: |
          aws s3 cp models/model_${{ env.MODEL_VERSION }}.joblib \
            s3://models/staging/model_${{ env.MODEL_VERSION }}.joblib

      - name: Deploy to staging
        run: |
          # Update Kubernetes deployment
          kubectl set image deployment/ml-api \
            ml-api=your-registry/ml-api:${{ env.MODEL_VERSION }} \
            --namespace=staging

          # Wait for rollout
          kubectl rollout status deployment/ml-api --namespace=staging

      - name: Run integration tests
        run: |
          python tests/integration_tests.py \
            --api-url https://staging-api.yourcompany.com \
            --test-data data/test.csv

  deploy-production:
    needs: deploy-staging
    runs-on: ubuntu-latest
    environment: production
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4

      - name: Download model
        uses: actions/download-artifact@v4
        with:
          name: trained-model

      - name: Upload to model registry
        run: |
          # Tag as production
          aws s3 cp models/model_${{ env.MODEL_VERSION }}.joblib \
            s3://models/production/model_${{ env.MODEL_VERSION }}.joblib

          aws s3 cp models/model_${{ env.MODEL_VERSION }}.joblib \
            s3://models/production/model.joblib  # Latest production

      - name: Canary deployment
        run: |
          # Deploy to 10% of traffic first
          kubectl set image deployment/ml-api \
            ml-api=your-registry/ml-api:${{ env.MODEL_VERSION }} \
            --namespace=production

          kubectl patch deployment ml-api \
            -p '{"spec":{"replicas":1}}' \
            --namespace=production

          # Wait and monitor
          sleep 300

      - name: Check canary metrics
        run: |
          # Query metrics from last 5 minutes
          python scripts/check_canary_metrics.py \
            --duration 5m \
            --error-threshold 0.05

      - name: Full deployment
        run: |
          # Scale to full deployment
          kubectl patch deployment ml-api \
            -p '{"spec":{"replicas":10}}' \
            --namespace=production

          kubectl rollout status deployment/ml-api --namespace=production

      - name: Update model metadata
        run: |
          # Record deployment in model registry
          python scripts/update_model_registry.py \
            --version ${{ env.MODEL_VERSION }} \
            --status deployed \
            --metrics metrics.json

# ab_testing.py - A/B testing framework for ML models

import hashlib
from typing import Optional, Dict
import logging
from datetime import datetime


class ModelABTest:
    """A/B test multiple model versions."""

    def __init__(
        self,
        model_a,
        model_b,
        model_a_name: str = "control",
        model_b_name: str = "treatment",
        traffic_split: float = 0.5
    ):
        """
        Initialize A/B test.

        Args:
            model_a: Control model (current production)
            model_b: Treatment model (new version)
            model_a_name: Name for model A
            model_b_name: Name for model B
            traffic_split: Percentage of traffic to model B (0.0-1.0)
        """
        self.model_a = model_a
        self.model_b = model_b
        self.model_a_name = model_a_name
        self.model_b_name = model_b_name
        self.traffic_split = traffic_split

        # Metrics tracking
        self.metrics = {
            model_a_name: {'predictions': 0, 'latencies': []},
            model_b_name: {'predictions': 0, 'latencies': []}
        }

        self.logger = logging.getLogger(__name__)

    def _assign_variant(self, user_id: str) -> str:
        """
        Consistently assign user to model variant.

        Uses hash of user_id for stable assignment (same user always
        gets same model during test period).
        """
        # Hash user_id to get deterministic assignment
        hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
        assignment = (hash_val % 100) / 100.0

        if assignment < self.traffic_split:
            return self.model_b_name
        else:
            return self.model_a_name

    def predict(self, features, user_id: str) -> Dict:
        """
        Make prediction using assigned model variant.

        Returns:
            Prediction with metadata about which model was used
        """
        import time

        # Assign variant
        variant = self._assign_variant(user_id)

        # Select model
        model = self.model_b if variant == self.model_b_name else self.model_a

        # Predict and measure latency
        start = time.time()
        prediction = model.predict(features)
        latency = time.time() - start

        # Track metrics
        self.metrics[variant]['predictions'] += 1
        self.metrics[variant]['latencies'].append(latency)

        self.logger.info(
            f"Prediction: user={user_id}, variant={variant}, "
            f"latency={latency*1000:.2f}ms"
        )

        return {
            'prediction': prediction,
            'variant': variant,
            'latency_ms': latency * 1000,
            'timestamp': datetime.now().isoformat()
        }

    def get_metrics(self) -> Dict:
        """Get A/B test metrics."""
        import numpy as np

        results = {}

        for variant_name, data in self.metrics.items():
            if data['predictions'] > 0:
                results[variant_name] = {
                    'total_predictions': data['predictions'],
                    'avg_latency_ms': np.mean(data['latencies']) * 1000,
                    'p95_latency_ms': np.percentile(data['latencies'], 95) * 1000,
                    'p99_latency_ms': np.percentile(data['latencies'], 99) * 1000
                }

        return results


# Example: Running A/B test in production
if __name__ == "__main__":
    import joblib
    import numpy as np

    logging.basicConfig(level=logging.INFO)

    # Load models
    model_a = joblib.load('models/production_v1.joblib')  # Current
    model_b = joblib.load('models/candidate_v2.joblib')   # New

    # Initialize A/B test (50% traffic to new model)
    ab_test = ModelABTest(
        model_a=model_a,
        model_b=model_b,
        model_a_name="v1_production",
        model_b_name="v2_candidate",
        traffic_split=0.5
    )

    # Simulate predictions
    for i in range(1000):
        user_id = f"user_{i % 100}"  # 100 unique users
        features = np.random.randn(1, 10)

        result = ab_test.predict(features, user_id=user_id)

    # Analyze results
    metrics = ab_test.get_metrics()

    print("\nA/B Test Results:")
    print("-" * 60)
    for variant, data in metrics.items():
        print(f"\n{variant}:")
        print(f"  Predictions: {data['total_predictions']:,}")
        print(f"  Avg latency: {data['avg_latency_ms']:.2f}ms")
        print(f"  P95 latency: {data['p95_latency_ms']:.2f}ms")
        print(f"  P99 latency: {data['p99_latency_ms']:.2f}ms")

    # Statistical significance test
    from scipy import stats

    latencies_a = ab_test.metrics['v1_production']['latencies']
    latencies_b = ab_test.metrics['v2_candidate']['latencies']

    t_stat, p_value = stats.ttest_ind(latencies_a, latencies_b)

    print(f"\nStatistical test (latency difference):")
    print(f"  T-statistic: {t_stat:.4f}")
    print(f"  P-value: {p_value:.4f}")
    print(f"  Significant: {p_value < 0.05}")