MLOps Validator — Machine Learning Model Deployment Pipeline

import pytest
import requests
import json
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
import numpy as np

class ModelValidator:
    def __init__(self, model_version: str, baseline_metrics: dict):
        self.model_version = model_version
        self.baseline_accuracy = baseline_metrics['accuracy']
        self.baseline_precision = baseline_metrics['precision']
        self.baseline_f1 = baseline_metrics['f1']
        self.model_url = f"{os.getenv('ML_API_URL')}/api/v1/models/{model_version}/predict"
        self.tolerance = 0.01  # ±1% tolerance
    
    def load_test_data(self, test_set_path: str):
        """Load ground truth test dataset."""
        with open(test_set_path, 'r') as f:
            data = json.load(f)
        return data['features'], data['labels']
    
    def run_inference(self, features: list) -> list:
        """Run inference on model via API."""
        response = requests.post(
            self.model_url,
            json={'features': features, 'batch_size': 100},
            timeout=30
        )
        response.raise_for_status()
        return response.json()['predictions']
    
    def validate_accuracy(self, features: list, ground_truth: list) -> dict:
        """Validate model accuracy against ground truth."""
        predictions = self.run_inference(features)
        
        accuracy = accuracy_score(ground_truth, predictions)
        precision = precision_score(ground_truth, predictions, average='weighted')
        recall = recall_score(ground_truth, predictions, average='weighted')
        f1 = f1_score(ground_truth, predictions, average='weighted')
        
        # Validate within tolerance
        assert abs(accuracy - self.baseline_accuracy) <= self.tolerance, \
            f"Accuracy {accuracy:.3f} differs from baseline {self.baseline_accuracy:.3f} by {abs(accuracy - self.baseline_accuracy):.3f} (max {self.tolerance})"
        
        assert abs(f1 - self.baseline_f1) <= self.tolerance, \
            f"F1 {f1:.3f} differs from baseline {self.baseline_f1:.3f}"
        
        return {
            'accuracy': accuracy,
            'precision': precision,
            'recall': recall,
            'f1': f1,
            'status': 'PASSED'
        }

@pytest.mark.parametrize('model_version', ['v2.5.0', 'v2.5.1', 'v2.6.0'])
def test_model_inference_accuracy(model_version):
    """Test inference accuracy for each model version."""
    # Load baseline metrics from model registry
    baseline = {
        'accuracy': 0.945,
        'precision': 0.943,
        'f1': 0.944
    }
    
    validator = ModelValidator(model_version, baseline)
    
    # Load 10K ground truth test samples
    features, labels = validator.load_test_data('tests/data/test_set_10k.json')
    
    # Validate accuracy
    metrics = validator.validate_accuracy(features, labels)
    
    # Log results
    print(f"Model {model_version} - Accuracy: {metrics['accuracy']:.3f} (baseline: {baseline['accuracy']:.3f})")
    
    assert metrics['status'] == 'PASSED'

import pytest
import pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metrics import (
    DataDriftTable,
    ColumnDriftMetric,
    ColumnMissingValuesMetric,
    ColumnOutliersMetric,
)
import requests

class DataDriftValidator:
    def __init__(self, model_id: str):
        self.model_id = model_id
        self.drift_threshold = 0.1  # 10% drift threshold
        self.monitoring_api = os.getenv('MONITORING_API_URL')
    
    def fetch_production_data(self, hours_back: int = 24) -> pd.DataFrame:
        """Fetch recent production data for drift detection."""
        response = requests.get(
            f"{self.monitoring_api}/api/v1/models/{self.model_id}/production_data",
            params={'hours_back': hours_back}
        )
        response.raise_for_status()
        return pd.DataFrame(response.json()['data'])
    
    def load_reference_data(self) -> pd.DataFrame:
        """Load reference dataset (training data distribution)."""
        return pd.read_csv(f'models/{self.model_id}/reference_data.csv')
    
    def detect_drift(self, reference: pd.DataFrame, current: pd.DataFrame) -> dict:
        """Run Evidently AI drift detection."""
        # Data drift report
        drift_report = Report(metrics=[
            DataDriftPreset(),
            DataQualityPreset(),
        ])
        drift_report.run(reference_data=reference, current_data=current)
        
        # Extract drift results
        drift_results = {}
        for metric in drift_report.metrics:
            if isinstance(metric, DataDriftTable):
                # Check each feature for drift
                for col in reference.columns:
                    col_drift = metric.get_result()['data'].get(col, {}).get('drift_detected', False)
                    statistic = metric.get_result()['data'].get(col, {}).get('statistic', 0)
                    drift_results[col] = {
                        'drift_detected': col_drift,
                        'statistic': statistic
                    }
        
        return drift_results
    
    def validate_data_quality(self, current: pd.DataFrame) -> dict:
        """Check for data quality issues."""
        quality_issues = {}
        
        for col in current.columns:
            missing_pct = current[col].isna().sum() / len(current)
            
            # Flag if missing % > 5%
            if missing_pct > 0.05:
                quality_issues[col] = {'type': 'missing_values', 'percentage': missing_pct}
            
            # Detect outliers for numeric columns
            if current[col].dtype in ['int64', 'float64']:
                Q1 = current[col].quantile(0.25)
                Q3 = current[col].quantile(0.75)
                IQR = Q3 - Q1
                outlier_count = ((current[col] < Q1 - 1.5*IQR) | (current[col] > Q3 + 1.5*IQR)).sum()
                outlier_pct = outlier_count / len(current)
                
                if outlier_pct > 0.05:
                    quality_issues[col] = {'type': 'outliers', 'percentage': outlier_pct}
        
        return quality_issues

@pytest.mark.parametrize('model_id', ['recommendation-model', 'fraud-detection-v3', 'churn-prediction'])
def test_data_drift_detection(model_id):
    """Detect data drift in production model inputs."""
    validator = DataDriftValidator(model_id)
    
    # Fetch data
    reference = validator.load_reference_data()
    current = validator.fetch_production_data(hours_back=24)
    
    # Detect drift
    drift_results = validator.detect_drift(reference, current)
    
    # Check for significant drift
    drift_detected_cols = [col for col, result in drift_results.items() if result['drift_detected']]
    
    # Log findings
    print(f"\nModel: {model_id}")
    print(f"Drift detected in {len(drift_detected_cols)} features: {drift_detected_cols}")
    
    # Validate data quality
    quality_issues = validator.validate_data_quality(current)
    
    if quality_issues:
        print(f"Data quality issues detected: {quality_issues}")
        pytest.fail(f"Data quality issues in {model_id}: {quality_issues}")
    
    # Alert if drift detected in critical features
    critical_features = ['user_id', 'transaction_amount', 'timestamp']
    for col in drift_detected_cols:
        if col in critical_features:
            pytest.warns(UserWarning, f"Drift detected in critical feature: {col}")