Data Distribution Shifts

from scipy.stats import ks_2samp
import numpy as np

class StatisticalDriftDetector:
    def __init__(self, reference_data, threshold=0.05):
        self.reference_data = reference_data
        self.threshold = threshold
        
    def detect_drift(self, current_data, feature_names):
        drift_results = {}
        
        for i, feature in enumerate(feature_names):
            ref_feature = self.reference_data[:, i]
            curr_feature = current_data[:, i]
            
            # Kolmogorov-Smirnov test
            statistic, p_value = ks_2samp(ref_feature, curr_feature)
            
            drift_results[feature] = {
                'statistic': statistic,
                'p_value': p_value,
                'drift_detected': p_value < self.threshold,
                'severity': self.calculate_severity(statistic)
            }
            
        return drift_results
    
    def calculate_severity(self, statistic):
        if statistic > 0.3:
            return 'high'
        elif statistic > 0.1:
            return 'medium'
        else:
            return 'low'

class PSIDetector:
    def __init__(self, bins=10):
        self.bins = bins
        self.reference_distribution = None
        
    def fit(self, reference_data):
        """Fit on reference data"""
        self.bin_edges = np.histogram_bin_edges(reference_data, bins=self.bins)
        ref_counts, _ = np.histogram(reference_data, bins=self.bin_edges)
        self.reference_distribution = ref_counts / len(reference_data)
        
    def calculate_psi(self, current_data):
        """Calculate Population Stability Index"""
        curr_counts, _ = np.histogram(current_data, bins=self.bin_edges)
        current_distribution = curr_counts / len(current_data)
        
        # Avoid division by zero
        ref_dist = np.where(self.reference_distribution == 0, 0.0001, 
                           self.reference_distribution)
        curr_dist = np.where(current_distribution == 0, 0.0001, 
                           current_distribution)
        
        psi = np.sum((curr_dist - ref_dist) * np.log(curr_dist / ref_dist))
        
        return {
            'psi': psi,
            'interpretation': self.interpret_psi(psi),
            'action_required': psi > 0.25
        }
    
    def interpret_psi(self, psi):
        if psi < 0.1:
            return 'No significant change'
        elif psi < 0.25:
            return 'Some change detected'
        else:
            return 'Significant change detected'

class ImportanceWeighting:
    def __init__(self):
        self.weight_estimator = None
        
    def estimate_weights(self, X_source, X_target):
        """Estimate importance weights using density ratio estimation"""
        from sklearn.ensemble import RandomForestClassifier
        from sklearn.calibration import CalibratedClassifierCV
        
        # Create binary classification problem
        n_source, n_target = len(X_source), len(X_target)
        X_combined = np.vstack([X_source, X_target])
        y_combined = np.hstack([np.zeros(n_source), np.ones(n_target)])
        
        # Train calibrated classifier
        rf = RandomForestClassifier(n_estimators=100, random_state=42)
        self.weight_estimator = CalibratedClassifierCV(rf, cv=3)
        self.weight_estimator.fit(X_combined, y_combined)
        
        # Calculate importance weights
        source_probs = self.weight_estimator.predict_proba(X_source)[:, 1]
        weights = source_probs / (1 - source_probs + 1e-8)
        
        # Normalize weights
        weights = weights / np.mean(weights)
        weights = np.clip(weights, 0.1, 10)  # Clip extreme weights
        
        return weights
    
    def fit_weighted_model(self, X_source, y_source, X_target, model):
        """Fit model with importance weights"""
        weights = self.estimate_weights(X_source, X_target)
        model.fit(X_source, y_source, sample_weight=weights)
        return model

import torch
import torch.nn as nn

class DomainAdversarialNet(nn.Module):
    def __init__(self, input_size, hidden_size, num_classes):
        super().__init__()
        self.feature_extractor = nn.Sequential(
            nn.Linear(input_size, hidden_size),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(hidden_size, hidden_size),
            nn.ReLU()
        )
        
        self.classifier = nn.Linear(hidden_size, num_classes)
        self.domain_classifier = nn.Sequential(
            nn.Linear(hidden_size, hidden_size // 2),
            nn.ReLU(),
            nn.Linear(hidden_size // 2, 2)  # Source vs Target
        )
        
    def forward(self, x, alpha=1.0):
        features = self.feature_extractor(x)
        
        # Prediction for main task
        class_output = self.classifier(features)
        
        # Domain classification with gradient reversal
        reversed_features = GradientReversalLayer.apply(features, alpha)
        domain_output = self.domain_classifier(reversed_features)
        
        return class_output, domain_output

class GradientReversalLayer(torch.autograd.Function):
    @staticmethod
    def forward(ctx, x, alpha):
        ctx.alpha = alpha
        return x.view_as(x)
    
    @staticmethod
    def backward(ctx, grad_output):
        return grad_output.neg() * ctx.alpha, None

import asyncio
from datetime import datetime, timedelta
import pandas as pd
from typing import Dict, List, Any

class DriftDetectionPipeline:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.detectors = self._initialize_detectors()
        self.alert_manager = AlertManager(config['alerting'])
        self.model_manager = ModelManager(config['models'])
        
    def _initialize_detectors(self):
        return {
            'statistical': StatisticalDriftDetector(),
            'psi': PSIDetector(),
            'adversarial': AdversarialDriftDetector()
        }
    
    async def monitor_predictions(self, batch_data: pd.DataFrame):
        """Monitor incoming prediction requests for drift"""
        drift_results = {}
        
        # Run detection methods in parallel
        tasks = [
            self._detect_statistical_drift(batch_data),
            self._detect_psi_drift(batch_data),
            self._detect_adversarial_drift(batch_data)
        ]
        
        results = await asyncio.gather(*tasks)
        drift_results = dict(zip(['statistical', 'psi', 'adversarial'], results))
        
        # Aggregate results and make decisions
        overall_drift = self._aggregate_drift_signals(drift_results)
        
        if overall_drift['action_required']:
            await self._handle_drift_detection(overall_drift, batch_data)
            
        return overall_drift
    
    async def _detect_statistical_drift(self, data: pd.DataFrame):
        """Statistical drift detection"""
        reference_data = await self._get_reference_data()
        return self.detectors['statistical'].detect_drift(
            data.values, reference_data.values
        )
    
    def _aggregate_drift_signals(self, results: Dict):
        """Combine multiple drift detection signals"""
        drift_scores = []
        confidence_scores = []
        
        for method, result in results.items():
            if result.get('drift_detected', False):
                drift_scores.append(result.get('severity_score', 0.5))
                confidence_scores.append(result.get('confidence', 0.5))
        
        if not drift_scores:
            return {'action_required': False, 'confidence': 1.0}
            
        avg_drift_score = np.mean(drift_scores)
        avg_confidence = np.mean(confidence_scores)
        
        return {
            'action_required': avg_drift_score > 0.3 and avg_confidence > 0.7,
            'drift_score': avg_drift_score,
            'confidence': avg_confidence,
            'details': results
        }
    
    async def _handle_drift_detection(self, drift_result: Dict, data: pd.DataFrame):
        """Handle detected drift"""
        severity = self._calculate_severity(drift_result)
        
        if severity == 'high':
            # Immediate model rollback or traffic reduction
            await self.model_manager.reduce_traffic(percentage=50)
            await self.alert_manager.send_alert('CRITICAL', drift_result)
            
        elif severity == 'medium':
            # Schedule model retraining
            await self.model_manager.schedule_retraining(data)
            await self.alert_manager.send_alert('WARNING', drift_result)
            
        # Log for analysis
        await self._log_drift_event(drift_result, data)

class AdaptiveModelManager:
    def __init__(self, config):
        self.models = {}  # Multiple model versions
        self.traffic_split = {}
        self.performance_tracker = PerformanceTracker()
        
    async def handle_drift(self, drift_info: Dict, new_data: pd.DataFrame):
        """Adaptive response to detected drift"""
        
        if drift_info['severity'] == 'high':
            # Immediate adaptation
            await self._emergency_response(drift_info)
            
        elif drift_info['severity'] == 'medium':
            # Gradual adaptation
            await self._gradual_adaptation(new_data)
            
        else:
            # Monitor and collect data
            await self._data_collection_mode(new_data)
    
    async def _emergency_response(self, drift_info):
        """Immediate response to severe drift"""
        # Reduce traffic to affected model
        current_traffic = self.traffic_split.get('main_model', 100)
        new_traffic = max(current_traffic * 0.5, 10)  # Reduce by 50%, min 10%
        
        await self.update_traffic_split({
            'main_model': new_traffic,
            'fallback_model': 100 - new_traffic
        })
        
        # Start emergency retraining
        await self.start_emergency_retraining()
    
    async def _gradual_adaptation(self, new_data):
        """Gradual model adaptation"""
        # Online learning approach
        if hasattr(self.models['main_model'], 'partial_fit'):
            # Incremental learning
            await self._incremental_update(new_data)
        else:
            # Batch retraining with new data
            await self._schedule_batch_retraining(new_data)
    
    async def _incremental_update(self, new_data):
        """Incremental model update"""
        # Get labels for recent predictions (delayed feedback)
        labels = await self._get_delayed_labels(new_data)
        
        if len(labels) > 0:
            # Partial fit with importance weighting
            weights = self._calculate_importance_weights(new_data)
            self.models['main_model'].partial_fit(
                new_data.values, labels, sample_weight=weights
            )
            
            # Validate updated model
            validation_score = await self._validate_model_update()
            if validation_score < self.config['min_performance_threshold']:
                await self._rollback_model_update()
    
    async def update_traffic_split(self, new_split: Dict[str, float]):
        """Update traffic routing between models"""
        self.traffic_split = new_split
        
        # Update load balancer configuration
        await self._update_load_balancer(new_split)
        
        # Log traffic split change
        await self._log_traffic_change(new_split)