⚖️ AI Ethics & Governance

import numpy as np
import pandas as pd
from sklearn.metrics import confusion_matrix

class BiasDetector:
    def __init__(self, protected_attributes):
        self.protected_attributes = protected_attributes
        self.bias_metrics = {}
        
    def detect_bias(self, data, predictions, labels):
        """Detect various types of bias in model predictions"""
        results = {}
        
        for attribute in self.protected_attributes:
            groups = data[attribute].unique()
            
            # Calculate metrics for each group
            group_metrics = {}
            for group in groups:
                mask = data[attribute] == group
                group_pred = predictions[mask]
                group_label = labels[mask]
                
                # Basic metrics
                group_metrics[group] = {
                    'size': mask.sum(),
                    'positive_rate': group_pred.mean(),
                    'true_positive_rate': self.tpr(group_label, group_pred),
                    'false_positive_rate': self.fpr(group_label, group_pred),
                    'precision': self.precision(group_label, group_pred),
                    'accuracy': (group_pred == group_label).mean()
                }
            
            # Calculate fairness metrics
            results[attribute] = {
                'group_metrics': group_metrics,
                'demographic_parity': self.demographic_parity(group_metrics),
                'equal_opportunity': self.equal_opportunity(group_metrics),
                'equalized_odds': self.equalized_odds(group_metrics),
                'disparate_impact': self.disparate_impact(group_metrics)
            }
        
        return results
    
    def demographic_parity(self, group_metrics):
        """Difference in positive prediction rates"""
        rates = [m['positive_rate'] for m in group_metrics.values()]
        return max(rates) - min(rates)
    
    def equal_opportunity(self, group_metrics):
        """Difference in true positive rates"""
        tprs = [m['true_positive_rate'] for m in group_metrics.values()]
        return max(tprs) - min(tprs)
    
    def equalized_odds(self, group_metrics):
        """Difference in TPR and FPR"""
        tprs = [m['true_positive_rate'] for m in group_metrics.values()]
        fprs = [m['false_positive_rate'] for m in group_metrics.values()]
        return max(max(tprs) - min(tprs), max(fprs) - min(fprs))
    
    def disparate_impact(self, group_metrics):
        """Ratio of positive rates (80% rule)"""
        rates = [m['positive_rate'] for m in group_metrics.values()]
        if min(rates) > 0:
            return min(rates) / max(rates)
        return 0
    
    def generate_report(self, results):
        """Generate bias assessment report"""
        report = []
        
        for attribute, metrics in results.items():
            report.append(f"\n=== {attribute.upper()} ===")
            
            # Group statistics
            for group, stats in metrics['group_metrics'].items():
                report.append(f"\n{group}:")
                report.append(f"  Size: {stats['size']}")
                report.append(f"  Positive Rate: {stats['positive_rate']:.3f}")
                report.append(f"  Accuracy: {stats['accuracy']:.3f}")
            
            # Fairness metrics
            report.append(f"\nFairness Metrics:")
            report.append(f"  Demographic Parity: {metrics['demographic_parity']:.3f}")
            report.append(f"  Equal Opportunity: {metrics['equal_opportunity']:.3f}")
            report.append(f"  Equalized Odds: {metrics['equalized_odds']:.3f}")
            report.append(f"  Disparate Impact: {metrics['disparate_impact']:.3f}")
            
            # Recommendations
            if metrics['disparate_impact'] < 0.8:
                report.append("  ⚠️ WARNING: Fails 80% rule for disparate impact")
            if metrics['demographic_parity'] > 0.1:
                report.append("  ⚠️ WARNING: Significant demographic parity difference")
        
        return "\n".join(report)

# Usage
detector = BiasDetector(['gender', 'race', 'age_group'])
bias_results = detector.detect_bias(data, predictions, labels)
print(detector.generate_report(bias_results))

import numpy as np

class DifferentialPrivacy:
    def __init__(self, epsilon=1.0, delta=1e-5):
        """
        epsilon: privacy budget (lower = more private)
        delta: probability of privacy breach
        """
        self.epsilon = epsilon
        self.delta = delta
    
    def add_laplace_noise(self, data, sensitivity):
        """Add Laplace noise for differential privacy"""
        scale = sensitivity / self.epsilon
        noise = np.random.laplace(0, scale, data.shape)
        return data + noise
    
    def add_gaussian_noise(self, data, sensitivity):
        """Add Gaussian noise for (ε,δ)-differential privacy"""
        sigma = sensitivity * np.sqrt(2 * np.log(1.25 / self.delta)) / self.epsilon
        noise = np.random.normal(0, sigma, data.shape)
        return data + noise
    
    def private_mean(self, data, lower_bound, upper_bound):
        """Calculate differentially private mean"""
        # Clip data to bounds
        clipped = np.clip(data, lower_bound, upper_bound)
        
        # Calculate sensitivity
        sensitivity = (upper_bound - lower_bound) / len(data)
        
        # Add noise to mean
        true_mean = np.mean(clipped)
        private_mean = self.add_laplace_noise(true_mean, sensitivity)
        
        return private_mean
    
    def private_histogram(self, data, bins):
        """Create differentially private histogram"""
        # Create histogram
        hist, edges = np.histogram(data, bins=bins)
        
        # Add noise (sensitivity = 1 for counting queries)
        private_hist = self.add_laplace_noise(hist, sensitivity=1)
        
        # Ensure non-negative counts
        private_hist = np.maximum(private_hist, 0)
        
        return private_hist, edges

# Usage
dp = DifferentialPrivacy(epsilon=0.1, delta=1e-5)
private_avg_age = dp.private_mean(ages, lower_bound=0, upper_bound=120)
print(f"Private average age: {private_avg_age:.1f}")

class ModelCard:
    def __init__(self, model_name, version):
        self.model_name = model_name
        self.version = version
        self.sections = {}
    
    def add_model_details(self, details):
        """Add basic model information"""
        self.sections['model_details'] = {
            'name': self.model_name,
            'version': self.version,
            'type': details.get('type', 'Classification'),
            'architecture': details.get('architecture'),
            'training_date': details.get('training_date'),
            'developers': details.get('developers', []),
            'contact': details.get('contact')
        }
    
    def add_intended_use(self, use_cases):
        """Document intended use cases"""
        self.sections['intended_use'] = {
            'primary_uses': use_cases.get('primary', []),
            'primary_users': use_cases.get('users', []),
            'out_of_scope': use_cases.get('out_of_scope', [])
        }
    
    def add_performance_metrics(self, metrics):
        """Add model performance metrics"""
        self.sections['metrics'] = {
            'overall': metrics.get('overall', {}),
            'subgroup': metrics.get('subgroup', {}),
            'confidence_intervals': metrics.get('confidence_intervals', {})
        }
    
    def add_ethical_considerations(self, ethics):
        """Document ethical considerations"""
        self.sections['ethics'] = {
            'bias_testing': ethics.get('bias_testing', {}),
            'fairness_metrics': ethics.get('fairness_metrics', {}),
            'privacy_measures': ethics.get('privacy_measures', []),
            'potential_harms': ethics.get('potential_harms', []),
            'mitigation_strategies': ethics.get('mitigation', [])
        }
    
    def add_limitations(self, limitations):
        """Document known limitations"""
        self.sections['limitations'] = limitations
    
    def generate_card(self, format='markdown'):
        """Generate the model card"""
        if format == 'markdown':
            return self._generate_markdown()
        elif format == 'json':
            return json.dumps(self.sections, indent=2)
        elif format == 'html':
            return self._generate_html()
    
    def _generate_markdown(self):
        """Generate markdown format model card"""
        md = []
        md.append(f"# Model Card: {self.model_name} v{self.version}")
        md.append("")
        
        # Model Details
        if 'model_details' in self.sections:
            md.append("## Model Details")
            details = self.sections['model_details']
            for key, value in details.items():
                if value:
                    md.append(f"- **{key.replace('_', ' ').title()}**: {value}")
            md.append("")
        
        # Intended Use
        if 'intended_use' in self.sections:
            md.append("## Intended Use")
            use = self.sections['intended_use']
            md.append("### Primary Uses")
            for item in use.get('primary_uses', []):
                md.append(f"- {item}")
            md.append("### Out of Scope")
            for item in use.get('out_of_scope', []):
                md.append(f"- ❌ {item}")
            md.append("")
        
        # Performance Metrics
        if 'metrics' in self.sections:
            md.append("## Performance Metrics")
            metrics = self.sections['metrics']
            if 'overall' in metrics:
                md.append("### Overall Performance")
                for metric, value in metrics['overall'].items():
                    md.append(f"- {metric}: {value}")
            md.append("")
        
        # Ethical Considerations
        if 'ethics' in self.sections:
            md.append("## Ethical Considerations")
            ethics = self.sections['ethics']
            if 'potential_harms' in ethics:
                md.append("### Potential Harms")
                for harm in ethics['potential_harms']:
                    md.append(f"- ⚠️ {harm}")
            if 'mitigation_strategies' in ethics:
                md.append("### Mitigation Strategies")
                for strategy in ethics['mitigation_strategies']:
                    md.append(f"- ✅ {strategy}")
            md.append("")
        
        # Limitations
        if 'limitations' in self.sections:
            md.append("## Limitations")
            for limitation in self.sections['limitations']:
                md.append(f"- {limitation}")
        
        return "\n".join(md)

# Usage Example
card = ModelCard("CreditRiskModel", "2.0")

card.add_model_details({
    'type': 'Binary Classification',
    'architecture': 'XGBoost',
    'training_date': '2024-03-15',
    'developers': ['AI Team'],
    'contact': 'ai-team@company.com'
})

card.add_intended_use({
    'primary': ['Credit risk assessment for loan applications'],
    'users': ['Credit analysts', 'Loan officers'],
    'out_of_scope': ['Investment advice', 'Criminal background checks']
})

card.add_performance_metrics({
    'overall': {
        'accuracy': 0.89,
        'precision': 0.87,
        'recall': 0.91,
        'f1_score': 0.89
    },
    'subgroup': {
        'gender': {'male': 0.88, 'female': 0.89},
        'age_group': {'<30': 0.86, '30-50': 0.90, '>50': 0.89}
    }
})

card.add_ethical_considerations({
    'bias_testing': {'demographic_parity': 0.03, 'equal_opportunity': 0.02},
    'potential_harms': ['May perpetuate historical lending biases'],
    'mitigation': ['Regular bias audits', 'Human review for edge cases']
})

card.add_limitations([
    'Performance may degrade for applicants with limited credit history',
    'Not validated for small business loans',
    'Requires retraining every 6 months'
])

print(card.generate_card())

import numpy as np
from typing import List, Dict, Tuple

class FederatedLearning:
    def __init__(self, num_clients: int, learning_rate: float = 0.01):
        self.num_clients = num_clients
        self.learning_rate = learning_rate
        self.global_model = None
        self.client_models = []
        
    def initialize_global_model(self, model_shape: Tuple):
        """Initialize the global model"""
        self.global_model = np.random.randn(*model_shape) * 0.01
        return self.global_model
    
    def distribute_model(self):
        """Distribute global model to clients"""
        return [self.global_model.copy() for _ in range(self.num_clients)]
    
    def train_on_client(self, client_id: int, client_data: np.ndarray, 
                       client_labels: np.ndarray, epochs: int = 1):
        """Train model on client's local data"""
        local_model = self.client_models[client_id].copy()
        
        for epoch in range(epochs):
            # Simulate local training (simplified)
            predictions = self.forward_pass(client_data, local_model)
            loss = self.compute_loss(predictions, client_labels)
            gradients = self.compute_gradients(client_data, client_labels, local_model)
            
            # Update local model
            local_model -= self.learning_rate * gradients
        
        # Add differential privacy noise
        privacy_noise = self.add_privacy_noise(local_model, epsilon=1.0)
        local_model += privacy_noise
        
        return local_model
    
    def federated_averaging(self, client_updates: List[np.ndarray], 
                          client_weights: List[float] = None):
        """Aggregate client updates using FedAvg"""
        if client_weights is None:
            client_weights = [1.0 / len(client_updates)] * len(client_updates)
        
        # Weighted average of client models
        aggregated_model = np.zeros_like(self.global_model)
        for model, weight in zip(client_updates, client_weights):
            aggregated_model += weight * model
        
        return aggregated_model
    
    def secure_aggregation(self, client_updates: List[np.ndarray]):
        """Secure aggregation with privacy guarantees"""
        # Add masks for secure aggregation
        masks = []
        for i in range(len(client_updates)):
            mask = np.random.randn(*client_updates[0].shape)
            masks.append(mask)
        
        # Masked updates
        masked_updates = []
        for update, mask in zip(client_updates, masks):
            masked_updates.append(update + mask)
        
        # Aggregate masked updates
        aggregated = np.mean(masked_updates, axis=0)
        
        # Remove masks (in real implementation, this is done securely)
        aggregated -= np.mean(masks, axis=0)
        
        return aggregated
    
    def add_privacy_noise(self, model: np.ndarray, epsilon: float):
        """Add differential privacy noise"""
        sensitivity = 1.0  # L2 sensitivity
        scale = sensitivity / epsilon
        noise = np.random.laplace(0, scale, model.shape)
        return noise
    
    def evaluate_fairness(self, test_data: Dict[str, np.ndarray]):
        """Evaluate model fairness across different groups"""
        fairness_metrics = {}
        
        for group_name, group_data in test_data.items():
            predictions = self.forward_pass(group_data['X'], self.global_model)
            accuracy = np.mean(predictions == group_data['y'])
            fairness_metrics[group_name] = {
                'accuracy': accuracy,
                'positive_rate': np.mean(predictions > 0.5)
            }
        
        # Calculate fairness measures
        accuracies = [m['accuracy'] for m in fairness_metrics.values()]
        pos_rates = [m['positive_rate'] for m in fairness_metrics.values()]
        
        fairness_metrics['overall'] = {
            'accuracy_disparity': max(accuracies) - min(accuracies),
            'demographic_parity': max(pos_rates) - min(pos_rates)
        }
        
        return fairness_metrics
    
    def run_federated_round(self, client_data: List[Dict]):
        """Run one round of federated learning"""
        # Distribute current global model
        self.client_models = self.distribute_model()
        
        # Train on each client
        client_updates = []
        for client_id, data in enumerate(client_data):
            local_model = self.train_on_client(
                client_id, 
                data['X'], 
                data['y'],
                epochs=5
            )
            client_updates.append(local_model)
        
        # Aggregate updates
        self.global_model = self.secure_aggregation(client_updates)
        
        return self.global_model

# Usage
fed_learning = FederatedLearning(num_clients=10)
fed_learning.initialize_global_model((100, 10))  # 100 features, 10 classes

# Simulate federated training
for round in range(10):
    print(f"Federated Round {round + 1}")
    global_model = fed_learning.run_federated_round(client_datasets)
    
    # Evaluate fairness
    fairness = fed_learning.evaluate_fairness(test_datasets)
    print(f"Fairness Metrics: {fairness['overall']}")

class ComplianceAutomation:
    def __init__(self, regulations=['GDPR', 'CCPA', 'EU_AI_Act']):
        self.regulations = regulations
        self.checks = self._load_compliance_checks()
        self.audit_log = []
        
    def run_compliance_check(self, ai_system):
        """Run automated compliance checks"""
        results = {
            'timestamp': datetime.now(),
            'system': ai_system.name,
            'version': ai_system.version,
            'checks': {}
        }
        
        for regulation in self.regulations:
            results['checks'][regulation] = self._check_regulation(
                ai_system, regulation
            )
        
        # Generate compliance score
        results['compliance_score'] = self._calculate_score(results['checks'])
        
        # Log results
        self.audit_log.append(results)
        
        return results
    
    def _check_regulation(self, ai_system, regulation):
        """Check compliance with specific regulation"""
        checks = self.checks[regulation]
        results = {}
        
        for check_name, check_func in checks.items():
            try:
                passed, details = check_func(ai_system)
                results[check_name] = {
                    'passed': passed,
                    'details': details,
                    'timestamp': datetime.now()
                }
            except Exception as e:
                results[check_name] = {
                    'passed': False,
                    'error': str(e)
                }
        
        return results
    
    def _load_compliance_checks(self):
        """Load compliance check functions"""
        return {
            'GDPR': {
                'data_minimization': self.check_data_minimization,
                'consent': self.check_consent_mechanism,
                'right_to_explanation': self.check_explainability,
                'data_protection_by_design': self.check_privacy_by_design
            },
            'EU_AI_Act': {
                'risk_assessment': self.check_risk_assessment,
                'human_oversight': self.check_human_oversight,
                'transparency': self.check_transparency,
                'robustness': self.check_robustness
            }
        }
    
    def check_data_minimization(self, ai_system):
        """Check if system follows data minimization principle"""
        features_used = len(ai_system.get_features())
        features_needed = len(ai_system.get_essential_features())
        
        ratio = features_needed / features_used if features_used > 0 else 0
        passed = ratio > 0.8  # At least 80% of features are essential
        
        return passed, {
            'features_used': features_used,
            'features_needed': features_needed,
            'ratio': ratio
        }
    
    def check_human_oversight(self, ai_system):
        """Check for human oversight mechanisms"""
        has_override = ai_system.has_human_override()
        has_monitoring = ai_system.has_monitoring_dashboard()
        has_alerts = ai_system.has_alert_system()
        
        passed = all([has_override, has_monitoring, has_alerts])
        
        return passed, {
            'human_override': has_override,
            'monitoring': has_monitoring,
            'alerts': has_alerts
        }
    
    def generate_compliance_report(self):
        """Generate comprehensive compliance report"""
        report = {
            'summary': self._generate_summary(),
            'detailed_findings': self.audit_log[-1] if self.audit_log else None,
            'recommendations': self._generate_recommendations(),
            'certification_ready': self._check_certification_readiness()
        }
        
        return report

# Usage
compliance = ComplianceAutomation()
results = compliance.run_compliance_check(ai_system)
report = compliance.generate_compliance_report()

if report['certification_ready']:
    print("✅ System ready for certification")
else:
    print("⚠️ Address compliance gaps before certification")

# Demographic Parity
P(Ŷ=1|A=0) = P(Ŷ=1|A=1)

# Equal Opportunity
P(Ŷ=1|Y=1,A=0) = P(Ŷ=1|Y=1,A=1)

# Equalized Odds
P(Ŷ=1|Y=y,A=0) = P(Ŷ=1|Y=y,A=1) for y ∈ {0,1}

# Disparate Impact (80% rule)
P(Ŷ=1|A=0) / P(Ŷ=1|A=1) ≥ 0.8

# Individual Fairness
d(x₁, x₂) small → |f(x₁) - f(x₂)| small

# Counterfactual Fairness
P(Ŷ_A←a = y | A = a', X = x) = P(Ŷ_A←a' = y | A = a', X = x)