MLOps & Monitoring

import numpy as np
import pandas as pd
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
from scipy import stats

class ModelPerformanceMonitor:
    def __init__(self, model_name: str, baseline_metrics: Dict):
        self.model_name = model_name
        self.baseline_metrics = baseline_metrics
        self.performance_history = []
        self.drift_threshold = 0.05  # 5% degradation threshold
        
    def log_prediction_batch(self, predictions: np.ndarray, 
                           ground_truth: Optional[np.ndarray] = None,
                           prediction_probs: Optional[np.ndarray] = None,
                           metadata: Optional[Dict] = None) -> None:
        """Log a batch of predictions for monitoring"""
        
        timestamp = datetime.now()
        
        batch_metrics = {
            'timestamp': timestamp,
            'batch_size': len(predictions),
            'predictions': predictions.copy(),
            'metadata': metadata or {}
        }
        
        # Add ground truth metrics if available
        if ground_truth is not None:
            batch_metrics.update({
                'ground_truth': ground_truth.copy(),
                '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')
            })
        
        # Add prediction confidence metrics
        if prediction_probs is not None:
            batch_metrics.update({
                'avg_confidence': np.mean(np.max(prediction_probs, axis=1)),
                'confidence_std': np.std(np.max(prediction_probs, axis=1)),
                'entropy': self._calculate_entropy(prediction_probs)
            })
        
        self.performance_history.append(batch_metrics)
        
        # Check for performance drift
        self._check_performance_drift(batch_metrics)
    
    def _calculate_entropy(self, probs: np.ndarray) -> float:
        """Calculate average prediction entropy"""
        entropy = -np.sum(probs * np.log(probs + 1e-8), axis=1)
        return np.mean(entropy)
    
    def _check_performance_drift(self, current_metrics: Dict) -> None:
        """Check if current performance has drifted from baseline"""
        
        if 'accuracy' not in current_metrics:
            return
        
        current_accuracy = current_metrics['accuracy']
        baseline_accuracy = self.baseline_metrics.get('accuracy', 0)
        
        # Check for significant degradation
        degradation = baseline_accuracy - current_accuracy
        
        if degradation > self.drift_threshold:
            alert = {
                'type': 'performance_drift',
                'severity': 'high' if degradation > 0.1 else 'medium',
                'message': f"Model accuracy dropped by {degradation:.2%}",
                'current_accuracy': current_accuracy,
                'baseline_accuracy': baseline_accuracy,
                'timestamp': current_metrics['timestamp']
            }
            
            self._send_alert(alert)
    
    def detect_data_drift(self, current_features: np.ndarray, 
                         reference_features: np.ndarray,
                         method: str = 'ks_test') -> Dict:
        """Detect feature drift using statistical tests"""
        
        drift_results = {}
        
        for feature_idx in range(current_features.shape[1]):
            current_feature = current_features[:, feature_idx]
            reference_feature = reference_features[:, feature_idx]
            
            if method == 'ks_test':
                # Kolmogorov-Smirnov test
                statistic, p_value = stats.ks_2samp(reference_feature, current_feature)
                drift_detected = p_value < 0.05
                
            elif method == 'psi':
                # Population Stability Index
                psi_score = self._calculate_psi(reference_feature, current_feature)
                drift_detected = psi_score > 0.2  # PSI > 0.2 indicates significant drift
                statistic, p_value = psi_score, None
                
            drift_results[f'feature_{feature_idx}'] = {
                'drift_detected': drift_detected,
                'statistic': statistic,
                'p_value': p_value,
                'method': method
            }
        
        # Overall drift assessment
        features_with_drift = sum(1 for result in drift_results.values() 
                                if result['drift_detected'])
        drift_percentage = features_with_drift / len(drift_results)
        
        if drift_percentage > 0.3:  # More than 30% of features show drift
            self._send_alert({
                'type': 'data_drift',
                'severity': 'high',
                'message': f"Data drift detected in {drift_percentage:.1%} of features",
                'features_affected': features_with_drift,
                'total_features': len(drift_results)
            })
        
        return {
            'overall_drift_detected': drift_percentage > 0.3,
            'drift_percentage': drift_percentage,
            'feature_results': drift_results
        }
    
    def _calculate_psi(self, reference: np.ndarray, current: np.ndarray, 
                      bins: int = 10) -> float:
        """Calculate Population Stability Index"""
        
        # Create bins based on reference distribution
        _, bin_edges = np.histogram(reference, bins=bins)
        
        # Calculate distributions
        ref_hist, _ = np.histogram(reference, bins=bin_edges)
        cur_hist, _ = np.histogram(current, bins=bin_edges)
        
        # Convert to percentages
        ref_pct = ref_hist / len(reference)
        cur_pct = cur_hist / len(current)
        
        # Calculate PSI
        psi = np.sum((cur_pct - ref_pct) * np.log((cur_pct + 1e-8) / (ref_pct + 1e-8)))
        
        return psi