Explainable AI & Monitoring

// Comprehensive Explainable AI Framework
import shap
import lime
import numpy as np
import pandas as pd
from typing import Dict, List, Optional, Union, Any
from sklearn.metrics import accuracy_score, precision_score, recall_score

class ExplainableAIFramework:
    def __init__(self, model, config: Dict):
        self.model = model
        self.config = config

        # Initialize explainability methods
        self.shap_explainer = None
        self.lime_explainer = None
        self.attention_analyzer = AttentionAnalyzer()

        # Monitoring components
        self.drift_detector = ModelDriftDetector()
        self.fairness_analyzer = FairnessAnalyzer()
        self.uncertainty_estimator = UncertaintyEstimator()

        # Compliance framework
        self.audit_logger = AuditLogger()
        self.model_card_generator = ModelCardGenerator()

    def initialize_explainers(self, X_train: np.ndarray, X_background: np.ndarray = None):
        """Initialize SHAP and LIME explainers"""
        # SHAP explainer initialization
        if hasattr(self.model, 'predict_proba'):
            self.shap_explainer = shap.Explainer(
                self.model.predict_proba,
                X_background if X_background is not None else X_train[:100]
            )
        else:
            self.shap_explainer = shap.Explainer(self.model.predict, X_train[:100])

        # LIME explainer for tabular data
        if self.config.get('data_type') == 'tabular':
            self.lime_explainer = lime.lime_tabular.LimeTabularExplainer(
                X_train,
                feature_names=self.config.get('feature_names'),
                class_names=self.config.get('class_names'),
                mode='classification' if hasattr(self.model, 'predict_proba') else 'regression'
            )

    def explain_prediction(self, instance: np.ndarray,
                          methods: List[str] = ['shap', 'lime']) -> Dict:
        """Generate comprehensive explanation for a single prediction"""
        explanations = {
            'prediction': self.model.predict([instance])[0],
            'confidence': self.get_prediction_confidence(instance),
            'explanations': {},
            'uncertainty': self.uncertainty_estimator.estimate(instance)
        }

        # SHAP explanation
        if 'shap' in methods and self.shap_explainer:
            shap_values = self.shap_explainer([instance])
            explanations['explanations']['shap'] = {
                'values': shap_values.values[0],
                'base_value': shap_values.base_values[0],
                'feature_importance': self.rank_feature_importance(shap_values.values[0])
            }

        # LIME explanation
        if 'lime' in methods and self.lime_explainer:
            lime_explanation = self.lime_explainer.explain_instance(
                instance,
                self.model.predict_proba if hasattr(self.model, 'predict_proba') else self.model.predict,
                num_features=self.config.get('num_features', 10)
            )
            explanations['explanations']['lime'] = {
                'local_explanation': lime_explanation.as_list(),
                'score': lime_explanation.score
            }

        # Counterfactual explanation
        if 'counterfactual' in methods:
            counterfactuals = self.generate_counterfactuals(instance)
            explanations['explanations']['counterfactual'] = counterfactuals

        return explanations

    def generate_counterfactuals(self, instance: np.ndarray,
                               num_counterfactuals: int = 3) -> List[Dict]:
        """Generate counterfactual explanations"""
        counterfactuals = []
        original_prediction = self.model.predict([instance])[0]

        for i in range(num_counterfactuals):
            # Perturbation-based counterfactual generation
            perturbed_instance = self.perturb_instance(instance, perturbation_strength=0.1 * (i + 1))
            new_prediction = self.model.predict([perturbed_instance])[0]

            if new_prediction != original_prediction:
                counterfactuals.append({
                    'instance': perturbed_instance,
                    'prediction': new_prediction,
                    'changes': self.compute_feature_changes(instance, perturbed_instance),
                    'distance': np.linalg.norm(instance - perturbed_instance)
                })

        return sorted(counterfactuals, key=lambda x: x['distance'])

    def monitor_model_performance(self, X_new: np.ndarray, y_true: np.ndarray = None) -> Dict:
        """Comprehensive model monitoring"""
        monitoring_results = {
            'timestamp': pd.Timestamp.now(),
            'data_statistics': self.compute_data_statistics(X_new),
            'drift_analysis': {},
            'performance_metrics': {},
            'fairness_metrics': {}
        }

        # Data drift detection
        drift_results = self.drift_detector.detect_drift(X_new)
        monitoring_results['drift_analysis'] = {
            'statistical_drift': drift_results['statistical_tests'],
            'distribution_shift': drift_results['distribution_metrics'],
            'feature_drift': drift_results['feature_level_drift']
        }

        # Performance monitoring (if ground truth available)
        if y_true is not None:
            y_pred = self.model.predict(X_new)
            monitoring_results['performance_metrics'] = {
                'accuracy': accuracy_score(y_true, y_pred),
                'precision': precision_score(y_true, y_pred, average='weighted'),
                'recall': recall_score(y_true, y_pred, average='weighted'),
                'prediction_distribution': np.bincount(y_pred) / len(y_pred)
            }

        # Fairness analysis
        if self.config.get('protected_attributes'):
            fairness_metrics = self.fairness_analyzer.analyze_fairness(
                X_new, self.model.predict(X_new), self.config['protected_attributes']
            )
            monitoring_results['fairness_metrics'] = fairness_metrics

        # Log monitoring results
        self.audit_logger.log_monitoring_event(monitoring_results)

        return monitoring_results

class ModelDriftDetector:
    def __init__(self):
        self.reference_stats = None
        self.drift_threshold = 0.05

    def detect_drift(self, X_new: np.ndarray) -> Dict:
        """Detect various types of model drift"""
        drift_results = {
            'statistical_tests': {},
            'distribution_metrics': {},
            'feature_level_drift': {}
        }

        if self.reference_stats is None:
            self.reference_stats = self.compute_reference_statistics(X_new)
            return drift_results

        # Statistical tests voor drift detection
        from scipy import stats

        for i, feature_name in enumerate(self.config.get('feature_names', [])):
            # Kolmogorov-Smirnov test
            ks_stat, ks_p_value = stats.ks_2samp(
                self.reference_stats['feature_distributions'][i],
                X_new[:, i]
            )

            # Population Stability Index
            psi = self.calculate_psi(
                self.reference_stats['feature_distributions'][i],
                X_new[:, i]
            )

            drift_results['feature_level_drift'][feature_name] = {
                'ks_statistic': ks_stat,
                'ks_p_value': ks_p_value,
                'psi': psi,
                'drift_detected': ks_p_value < self.drift_threshold or psi > 0.2
            }

        return 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_bin_edges(reference, bins=bins)

        # Calculate distributions
        ref_hist, _ = np.histogram(reference, bins=bin_edges)
        cur_hist, _ = np.histogram(current, bins=bin_edges)

        # Normalize to percentages
        ref_pct = ref_hist / len(reference) + 1e-6  # Small epsilon to avoid log(0)
        cur_pct = cur_hist / len(current) + 1e-6

        # Calculate PSI
        psi = np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))

        return psi

class FairnessAnalyzer:
    def analyze_fairness(self, X: np.ndarray, y_pred: np.ndarray,
                        protected_attributes: List[str]) -> Dict:
        """Comprehensive fairness analysis"""
        fairness_metrics = {}

        for attr in protected_attributes:
            attr_index = self.config['feature_names'].index(attr)
            protected_groups = np.unique(X[:, attr_index])

            group_metrics = {}
            for group in protected_groups:
                group_mask = X[:, attr_index] == group
                group_predictions = y_pred[group_mask]

                group_metrics[f'group_{group}'] = {
                    'size': np.sum(group_mask),
                    'positive_rate': np.mean(group_predictions == 1),
                    'prediction_distribution': np.bincount(group_predictions) / len(group_predictions)
                }

            # Calculate fairness metrics
            fairness_metrics[attr] = {
                'group_metrics': group_metrics,
                'demographic_parity': self.calculate_demographic_parity(group_metrics),
                'equalized_odds': self.calculate_equalized_odds(group_metrics),
                'disparate_impact': self.calculate_disparate_impact(group_metrics)
            }

        return fairness_metrics

    def calculate_demographic_parity(self, group_metrics: Dict) -> float:
        """Calculate demographic parity difference"""
        positive_rates = [metrics['positive_rate'] for metrics in group_metrics.values()]
        return max(positive_rates) - min(positive_rates)

# Model Card Generation voor compliance
class ModelCardGenerator:
    def generate_model_card(self, model, training_data: Dict,
                          performance_metrics: Dict, fairness_analysis: Dict) -> Dict:
        """Generate comprehensive model card"""
        model_card = {
            'model_details': {
                'model_type': type(model).__name__,
                'model_version': self.config.get('model_version', '1.0'),
                'training_date': pd.Timestamp.now().isoformat(),
                'intended_use': self.config.get('intended_use', ''),
                'out_of_scope_uses': self.config.get('out_of_scope_uses', [])
            },
            'training_data': {
                'dataset_name': training_data.get('dataset_name'),
                'dataset_size': training_data.get('dataset_size'),
                'feature_statistics': training_data.get('feature_statistics'),
                'preprocessing_steps': training_data.get('preprocessing_steps', [])
            },
            'evaluation_data': training_data.get('evaluation_data', {}),
            'quantitative_analysis': {
                'performance_metrics': performance_metrics,
                'fairness_metrics': fairness_analysis
            },
            'ethical_considerations': {
                'bias_analysis': fairness_analysis,
                'environmental_impact': self.config.get('environmental_impact'),
                'recommendations': self.config.get('ethical_recommendations', [])
            },
            'technical_specifications': {
                'model_architecture': self.config.get('model_architecture'),
                'hardware_requirements': self.config.get('hardware_requirements'),
                'software_dependencies': self.config.get('software_dependencies', [])
            }
        }

        return model_card