MLOps & Model Deployment

# model_monitoring.py - Comprehensive monitoring system

import numpy as np
import pandas as pd
from scipy import stats
from typing import Any, Dict
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)

    # Seeded so this example prints the same numbers every run. Without it the
    # simulated drift lands anywhere from -6% to -10%, which straddles the 10%
    # threshold below and makes the reported severity flip between "warning"
    # and "critical" run to run.
    np.random.seed(0)

    # 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)}")

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

import hashlib
import logging
import time
from datetime import datetime
from typing import Dict

import numpy as np
from scipy import stats


class ModelABTest:
    """Run two model versions side by side and compare them on quality."""

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

        Args:
            model_a: Control model (current production)
            model_b: Treatment model (new version)
            experiment: Name of this experiment, used to salt the assignment
            model_a_name: Name for model A
            model_b_name: Name for model B
            traffic_split: Fraction of traffic sent to model B (0.0-1.0)
        """
        self.model_a = model_a
        self.model_b = model_b
        self.experiment = experiment
        self.model_a_name = model_a_name
        self.model_b_name = model_b_name
        self.traffic_split = traffic_split

        # Latency answers "is it fast enough". Correct and scored answer
        # "is it better", which is the question the test exists to settle.
        self.metrics = {
            model_a_name: {'predictions': 0, 'latencies': [], 'correct': 0, 'scored': 0},
            model_b_name: {'predictions': 0, 'latencies': [], 'correct': 0, 'scored': 0},
        }

        self.logger = logging.getLogger(__name__)

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

        The experiment name is hashed along with the user id. Hashing the user
        id alone looks equivalent but is not: every experiment would then split
        the population exactly the same way, so the same users sit in treatment
        forever and effects from separate experiments pile onto one group.
        """
        digest = hashlib.md5(
            f"{self.experiment}:{user_id}".encode(), usedforsecurity=False
        ).hexdigest()
        assignment = (int(digest, 16) % 10_000) / 10_000

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

    def predict(self, features, user_id: str) -> Dict:
        """
        Make a prediction with the variant assigned to this user.

        Returns:
            The prediction plus metadata about which model served it
        """
        variant = self._assign_variant(user_id)
        model = self.model_b if variant == self.model_b_name else self.model_a

        start = time.perf_counter()
        prediction = model.predict(features)
        latency = time.perf_counter() - start

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

        # Per request, so debug: at 1k requests/s this line at INFO is a
        # log bill, not an insight.
        self.logger.debug(
            f"user={user_id}, variant={variant}, latency={latency * 1000:.2f}ms"
        )

        return {
            # float(), not the raw array: model.predict returns an ndarray,
            # which is not JSON serialisable and cannot leave an API.
            'prediction': float(prediction[0]),
            'variant': variant,
            'latency_ms': latency * 1000,
            'timestamp': datetime.now().isoformat()
        }

    def record_outcome(self, user_id: str, was_correct: bool) -> None:
        """
        Record whether a served prediction turned out to be right.

        Ground truth almost never arrives with the request: it shows up when
        the user clicks, the order ships, or a human labels the case. Route it
        back through the same assignment function so it lands on the variant
        that actually served the prediction.
        """
        variant = self._assign_variant(user_id)
        self.metrics[variant]['scored'] += 1
        if was_correct:
            self.metrics[variant]['correct'] += 1

    def get_metrics(self) -> Dict:
        """Summarise each variant."""
        results = {}

        for variant_name, data in self.metrics.items():
            if data['predictions'] == 0:
                continue

            results[variant_name] = {
                'total_predictions': data['predictions'],
                'accuracy': data['correct'] / data['scored'] if data['scored'] else None,
                'scored': data['scored'],
                '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

    def compare_accuracy(self) -> Dict:
        """
        Test whether the accuracy difference between variants is real.

        A chi-squared test on the correct/incorrect counts, which is the
        comparison the rollout decision hinges on. Testing latency instead
        answers a different question: at a few hundred samples any consistent
        speed difference is significant, and it says nothing about quality.
        """
        a = self.metrics[self.model_a_name]
        b = self.metrics[self.model_b_name]

        table = [
            [a['correct'], a['scored'] - a['correct']],
            [b['correct'], b['scored'] - b['correct']],
        ]

        chi2, p_value, _, _ = stats.chi2_contingency(table)

        return {
            'accuracy_control': a['correct'] / a['scored'],
            'accuracy_treatment': b['correct'] / b['scored'],
            'difference_pct_points': (b['correct'] / b['scored'] - a['correct'] / a['scored']) * 100,
            'chi2': chi2,
            'p_value': p_value,
            'significant': p_value < 0.05,
        }


# Example: running an A/B test
if __name__ == "__main__":
    import joblib

    logging.basicConfig(level=logging.INFO)
    rng = np.random.default_rng(0)

    model_a = joblib.load('models/production_v1.joblib')   # current
    model_b = joblib.load('models/candidate_v2.joblib')    # new

    ab_test = ModelABTest(
        model_a=model_a,
        model_b=model_b,
        experiment="ranker-v2-2026-08",
        model_a_name="v1_production",
        model_b_name="v2_candidate",
        traffic_split=0.5
    )

    for i in range(1000):
        user_id = f"user_{i % 100}"          # 100 unique users
        features = rng.normal(size=(1, 10))
        label = float(features[0, 0] > 0)     # the outcome observed later

        result = ab_test.predict(features, user_id=user_id)
        ab_test.record_outcome(user_id, was_correct=result['prediction'] == label)

    print("\nA/B Test Results:")
    print("-" * 60)
    for variant, data in ab_test.get_metrics().items():
        print(f"\n{variant}:")
        print(f"  Predictions: {data['total_predictions']:,}")
        print(f"  Accuracy:    {data['accuracy']:.3f} over {data['scored']:,} scored")
        print(f"  Avg latency: {data['avg_latency_ms']:.2f}ms")
        print(f"  P95 latency: {data['p95_latency_ms']:.2f}ms")

    verdict = ab_test.compare_accuracy()
    print("\nDoes the accuracy difference hold up?")
    print(f"  Control accuracy:   {verdict['accuracy_control']:.3f}")
    print(f"  Treatment accuracy: {verdict['accuracy_treatment']:.3f}")
    print(f"  Difference:         {verdict['difference_pct_points']:+.1f} points")
    print(f"  chi2={verdict['chi2']:.4f}, p={verdict['p_value']:.4f}")
    print(f"  Significant:        {verdict['significant']}")