Ética e Segurança - IA Responsável em Produção

import pandas as pd
import numpy as np
from sklearn.metrics import confusion_matrix, classification_report
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime
import json
import warnings
from typing import Dict, List, Tuple, Any
import hashlib
from collections import defaultdict

class EthicalAIAuditor:
    """Sistema completo de auditoria ética para modelos de IA"""
    
    def __init__(self, model_name: str = "Modelo_IA"):
        self.model_name = model_name
        self.audit_log = []
        self.fairness_metrics = {}
        self.privacy_checks = {}
        self.transparency_score = 0
        self.bias_analysis = {}
        
        print(f"🔍 Iniciando Auditoria Ética para: {model_name}")
    
    def check_data_privacy(self, data: pd.DataFrame, 
                          sensitive_columns: List[str] = None) -> Dict:
        """Verifica conformidade com privacidade de dados"""
        print("🔒 Verificando Privacidade de Dados...")
        
        privacy_report = {
            'timestamp': datetime.now().isoformat(),
            'total_records': len(data),
            'columns_analyzed': list(data.columns),
            'privacy_issues': [],
            'recommendations': [],
            'compliance_score': 0
        }
        
        # Detectar colunas potencialmente sensíveis
        if not sensitive_columns:
            sensitive_patterns = [
                'cpf', 'rg', 'email', 'phone', 'telefone', 'celular',
                'name', 'nome', 'address', 'endereco', 'birthday', 
                'nascimento', 'age', 'idade', 'salary', 'salario'
            ]
            
            sensitive_columns = []
            for col in data.columns:
                if any(pattern in col.lower() for pattern in sensitive_patterns):
                    sensitive_columns.append(col)
        
        # Verificar dados pessoais não anonimizados
        if sensitive_columns:
            for col in sensitive_columns:
                if col in data.columns:
                    unique_ratio = data[col].nunique() / len(data)
                    
                    if unique_ratio > 0.8:  # Alta cardinalidade sugere dados não anonimizados
                        privacy_report['privacy_issues'].append({
                            'column': col,
                            'issue': 'Possível dado pessoal não anonimizado',
                            'severity': 'HIGH',
                            'unique_ratio': round(unique_ratio, 3)
                        })
                        
                        privacy_report['recommendations'].append(
                            f"Considere anonimizar ou remover a coluna '{col}'"
                        )
        
        # Verificar duplicatas (possível re-identificação)
        duplicate_count = data.duplicated().sum()
        if duplicate_count > 0:
            privacy_report['privacy_issues'].append({
                'issue': 'Registros duplicados encontrados',
                'count': int(duplicate_count),
                'severity': 'MEDIUM'
            })
        
        # Verificar se há hashing ou criptografia em dados sensíveis
        hashed_columns = []
        for col in data.select_dtypes(include=['object']).columns:
            sample_values = data[col].dropna().head(100).astype(str)
            # Verificar se parecem hashes (32+ chars hexadecimais)
            hash_like = sample_values.str.match(r'^[a-f0-9]{32,}$')
            if hash_like.sum() / len(sample_values) > 0.8:
                hashed_columns.append(col)
        
        if hashed_columns:
            privacy_report['recommendations'].append(
                f"Colunas aparentemente hasheadas encontradas: {hashed_columns} - ✅ Boa prática!"
            )
        
        # Calcular score de compliance
        issues_count = len(privacy_report['privacy_issues'])
        high_severity = sum(1 for issue in privacy_report['privacy_issues'] 
                           if issue.get('severity') == 'HIGH')
        
        if issues_count == 0:
            privacy_report['compliance_score'] = 100
        else:
            privacy_report['compliance_score'] = max(0, 100 - (high_severity * 30) - (issues_count * 10))
        
        self.privacy_checks = privacy_report
        return privacy_report
    
    def analyze_bias(self, data: pd.DataFrame, 
                    predictions: np.array, 
                    true_labels: np.array,
                    protected_attributes: List[str]) -> Dict:
        """Análise abrangente de viés algorítmico"""
        print("⚖️ Analisando Viés Algorítmico...")
        
        bias_report = {
            'timestamp': datetime.now().isoformat(),
            'protected_attributes': protected_attributes,
            'bias_metrics': {},
            'fairness_assessment': {},
            'recommendations': []
        }
        
        for attr in protected_attributes:
            if attr not in data.columns:
                print(f"⚠️ Atributo protegido '{attr}' não encontrado nos dados")
                continue
            
            attr_analysis = {
                'attribute': attr,
                'groups': {},
                'fairness_metrics': {},
                'bias_detected': False
            }
            
            # Analisar cada grupo do atributo protegido
            unique_values = data[attr].unique()
            
            for value in unique_values:
                mask = data[attr] == value
                group_predictions = predictions[mask]
                group_true_labels = true_labels[mask]
                
                if len(group_predictions) == 0:
                    continue
                
                # Métricas por grupo
                group_metrics = {
                    'count': len(group_predictions),
                    'positive_rate': np.mean(group_predictions),
                    'accuracy': np.mean(group_predictions == group_true_labels),
                    'precision': self._calculate_precision(group_predictions, group_true_labels),
                    'recall': self._calculate_recall(group_predictions, group_true_labels)
                }
                
                attr_analysis['groups'][str(value)] = group_metrics
            
            # Calcular métricas de fairness
            groups = list(attr_analysis['groups'].keys())
            if len(groups) >= 2:
                # Paridade demográfica
                positive_rates = [attr_analysis['groups'][g]['positive_rate'] for g in groups]
                demographic_parity = max(positive_rates) - min(positive_rates)
                
                # Igualdade de oportunidades  
                recalls = [attr_analysis['groups'][g]['recall'] for g in groups 
                          if attr_analysis['groups'][g]['recall'] is not None]
                equalized_odds = max(recalls) - min(recalls) if recalls else 0
                
                attr_analysis['fairness_metrics'] = {
                    'demographic_parity_difference': round(demographic_parity, 4),
                    'equalized_odds_difference': round(equalized_odds, 4)
                }
                
                # Detectar viés (thresholds baseados em literatura)
                if demographic_parity > 0.1:  # 10% de diferença
                    attr_analysis['bias_detected'] = True
                    bias_report['recommendations'].append(
                        f"VIÉS DETECTADO em '{attr}': Diferença de {demographic_parity:.1%} "
                        f"na taxa de predições positivas entre grupos"
                    )
                
                if equalized_odds > 0.1:
                    attr_analysis['bias_detected'] = True
                    bias_report['recommendations'].append(
                        f"VIÉS DETECTADO em '{attr}': Diferença de {equalized_odds:.1%} "
                        f"no recall entre grupos"
                    )
            
            bias_report['bias_metrics'][attr] = attr_analysis
        
        # Avaliação geral de fairness
        bias_detected_count = sum(1 for attr_data in bias_report['bias_metrics'].values() 
                                 if attr_data.get('bias_detected', False))
        
        bias_report['fairness_assessment'] = {
            'overall_fair': bias_detected_count == 0,
            'biased_attributes': bias_detected_count,
            'total_attributes_checked': len(protected_attributes),
            'fairness_score': max(0, 100 - (bias_detected_count * 25))
        }
        
        self.bias_analysis = bias_report
        return bias_report
    
    def _calculate_precision(self, predictions, true_labels):
        """Calcula precisão tratando casos especiais"""
        try:
            tp = np.sum((predictions == 1) & (true_labels == 1))
            fp = np.sum((predictions == 1) & (true_labels == 0))
            return tp / (tp + fp) if (tp + fp) > 0 else None
        except:
            return None
    
    def _calculate_recall(self, predictions, true_labels):
        """Calcula recall tratando casos especiais"""
        try:
            tp = np.sum((predictions == 1) & (true_labels == 1))
            fn = np.sum((predictions == 0) & (true_labels == 1))
            return tp / (tp + fn) if (tp + fn) > 0 else None
        except:
            return None
    
    def check_model_transparency(self, model, feature_names: List[str] = None) -> Dict:
        """Avalia transparência e explicabilidade do modelo"""
        print("🔍 Avaliando Transparência do Modelo...")
        
        transparency_report = {
            'timestamp': datetime.now().isoformat(),
            'model_type': type(model).__name__,
            'interpretability_score': 0,
            'transparency_features': [],
            'recommendations': []
        }
        
        # Verificar se o modelo tem feature importance
        if hasattr(model, 'feature_importances_'):
            transparency_report['transparency_features'].append('Feature Importance Disponível')
            transparency_report['interpretability_score'] += 30
            
            if feature_names:
                importances = model.feature_importances_
                feature_importance = dict(zip(feature_names, importances))
                transparency_report['feature_importance'] = {
                    name: round(float(imp), 4) 
                    for name, imp in sorted(feature_importance.items(), 
                                          key=lambda x: x[1], reverse=True)[:10]
                }
        
        # Verificar se é um modelo "white-box" (interpretável)
        interpretable_models = [
            'LinearRegression', 'LogisticRegression', 'DecisionTreeClassifier',
            'DecisionTreeRegressor', 'LinearSVC'
        ]
        
        if any(model_name in type(model).__name__ for model_name in interpretable_models):
            transparency_report['transparency_features'].append('Modelo Interpretável por Natureza')
            transparency_report['interpretability_score'] += 40
        
        # Verificar disponibilidade de coeficientes
        if hasattr(model, 'coef_'):
            transparency_report['transparency_features'].append('Coeficientes Disponíveis')
            transparency_report['interpretability_score'] += 20
        
        # Verificar se tem métodos de explicação
        explanation_methods = ['predict_proba', 'decision_function']
        available_methods = [method for method in explanation_methods 
                           if hasattr(model, method)]
        
        if available_methods:
            transparency_report['transparency_features'].append(
                f'Métodos de Explicação: {available_methods}'
            )
            transparency_report['interpretability_score'] += 10
        
        # Recomendações baseadas no score
        if transparency_report['interpretability_score'] < 50:
            transparency_report['recommendations'].extend([
                "Considere usar técnicas de XAI (Explainable AI) como SHAP ou LIME",
                "Documente as decisões do modelo de forma clara",
                "Implemente logs de auditoria para decisões críticas"
            ])
        
        self.transparency_score = transparency_report['interpretability_score']
        return transparency_report
    
    def security_assessment(self, model, test_data: pd.DataFrame) -> Dict:
        """Avaliação de segurança do modelo"""
        print("🛡️ Realizando Avaliação de Segurança...")
        
        security_report = {
            'timestamp': datetime.now().isoformat(),
            'vulnerabilities_found': [],
            'security_score': 100,
            'robustness_tests': {},
            'recommendations': []
        }
        
        try:
            # Teste de robustez com dados adversariais simples
            original_predictions = model.predict(test_data.select_dtypes(include=[np.number]))
            
            # Adicionar ruído gaussiano
            noise_level = 0.1
            noisy_data = test_data.select_dtypes(include=[np.number]) + \
                        np.random.normal(0, noise_level, test_data.select_dtypes(include=[np.number]).shape)
            
            noisy_predictions = model.predict(noisy_data)
            
            # Calcular estabilidade
            prediction_changes = np.mean(original_predictions != noisy_predictions)
            
            security_report['robustness_tests']['noise_stability'] = {
                'noise_level': noise_level,
                'prediction_change_rate': round(float(prediction_changes), 4),
                'robust': prediction_changes < 0.05  # Menos de 5% de mudanças
            }
            
            if prediction_changes > 0.1:  # Mais de 10% de mudanças
                security_report['vulnerabilities_found'].append(
                    'Modelo sensível a ruído - possível vulnerabilidade a ataques adversariais'
                )
                security_report['security_score'] -= 30
            
        except Exception as e:
            security_report['robustness_tests']['error'] = f"Erro no teste: {str(e)}"
        
        # Verificar overfitting (indicador indireto de segurança)
        if hasattr(model, 'score'):
            try:
                train_score = getattr(model, '_train_score', None)
                if train_score and hasattr(model, 'score'):
                    # Em um cenário real, você teria dados de validação separados
                    score_diff = abs(train_score - 0.85)  # Placeholder
                    
                    if score_diff > 0.2:  # Grande diferença sugere overfitting
                        security_report['vulnerabilities_found'].append(
                            'Possível overfitting detectado - pode vazar informações de treinamento'
                        )
                        security_report['security_score'] -= 20
            except:
                pass
        
        # Recomendações de segurança
        security_report['recommendations'].extend([
            "Implemente rate limiting na API de produção",
            "Use validação rigorosa de inputs",
            "Configure logs de segurança detalhados",
            "Estabeleça monitoramento de anomalias",
            "Considere usar técnicas de differential privacy"
        ])
        
        return security_report
    
    def generate_ethics_checklist(self) -> Dict:
        """Gera checklist completo de ética"""
        print("📋 Gerando Checklist de Ética...")
        
        checklist = {
            'timestamp': datetime.now().isoformat(),
            'model': self.model_name,
            'categories': {
                'Transparência e Explicabilidade': [
                    {'item': 'Modelo é interpretável ou tem ferramentas de explicação', 'status': 'pending'},
                    {'item': 'Decisões importantes podem ser explicadas aos usuários', 'status': 'pending'},
                    {'item': 'Processo de treinamento é documentado', 'status': 'pending'},
                    {'item': 'Limitações do modelo são claramente comunicadas', 'status': 'pending'}
                ],
                'Equidade e Não-Discriminação': [
                    {'item': 'Análise de viés foi realizada em grupos protegidos', 'status': 'pending'},
                    {'item': 'Métricas de fairness atendem padrões aceitáveis', 'status': 'pending'},
                    {'item': 'Dados de treinamento são representativos', 'status': 'pending'},
                    {'item': 'Sistema não perpetua discriminação histórica', 'status': 'pending'}
                ],
                'Privacidade e Proteção de Dados': [
                    {'item': 'Dados pessoais são adequadamente protegidos', 'status': 'pending'},
                    {'item': 'Princípios de minimização de dados são aplicados', 'status': 'pending'},
                    {'item': 'Consentimento apropriado foi obtido', 'status': 'pending'},
                    {'item': 'Conformidade com LGPD/GDPR verificada', 'status': 'pending'}
                ],
                'Segurança e Robustez': [
                    {'item': 'Modelo é robusto contra ataques adversariais', 'status': 'pending'},
                    {'item': 'Sistemas de monitoramento estão implementados', 'status': 'pending'},
                    {'item': 'Planos de contingência existem para falhas', 'status': 'pending'},
                    {'item': 'Atualizações de segurança são aplicadas regularmente', 'status': 'pending'}
                ],
                'Responsabilidade e Governança': [
                    {'item': 'Responsabilidades estão claramente definidas', 'status': 'pending'},
                    {'item': 'Processo de auditoria regular está estabelecido', 'status': 'pending'},
                    {'item': 'Mecanismos de feedback e correção existem', 'status': 'pending'},
                    {'item': 'Impactos sociais foram considerados', 'status': 'pending'}
                ]
            },
            'overall_compliance': 0
        }
        
        return checklist
    
    def generate_comprehensive_report(self) -> Dict:
        """Gera relatório completo de auditoria ética"""
        print("📄 Gerando Relatório Completo de Auditoria Ética...")
        
        report = {
            'audit_metadata': {
                'model_name': self.model_name,
                'audit_date': datetime.now().isoformat(),
                'auditor_version': '1.0.0',
                'audit_id': hashlib.md5(f"{self.model_name}_{datetime.now()}".encode()).hexdigest()[:12]
            },
            'executive_summary': {},
            'detailed_findings': {
                'privacy_assessment': self.privacy_checks,
                'bias_analysis': self.bias_analysis,
                'transparency_score': self.transparency_score,
                'security_findings': {}
            },
            'recommendations': {
                'high_priority': [],
                'medium_priority': [],
                'low_priority': []
            },
            'compliance_scores': {},
            'action_plan': []
        }
        
        # Calcular scores gerais
        privacy_score = self.privacy_checks.get('compliance_score', 0) if self.privacy_checks else 0
        fairness_score = self.bias_analysis.get('fairness_assessment', {}).get('fairness_score', 0) if self.bias_analysis else 0
        
        overall_score = np.mean([privacy_score, fairness_score, self.transparency_score])
        
        report['executive_summary'] = {
            'overall_ethics_score': round(overall_score, 1),
            'privacy_compliance': privacy_score,
            'fairness_assessment': fairness_score,
            'transparency_level': self.transparency_score,
            'recommendation_count': len(self.privacy_checks.get('recommendations', [])) + \
                                  len(self.bias_analysis.get('recommendations', [])),
            'risk_level': 'LOW' if overall_score >= 80 else 'MEDIUM' if overall_score >= 60 else 'HIGH'
        }
        
        print(f"✅ Auditoria Concluída! Score Geral de Ética: {overall_score:.1f}/100")
        print(f"🎯 Nível de Risco: {report['executive_summary']['risk_level']}")
        
        return report

# Exemplo de uso prático
def demo_ethical_audit():
    """Demonstração do sistema de auditoria ética"""
    print("🎯 DEMO: Sistema de Auditoria Ética para IA")
    print("=" * 60)
    
    # Criar dados de exemplo
    np.random.seed(42)
    n_samples = 1000
    
    # Dados simulados com potencial viés
    data = pd.DataFrame({
        'idade': np.random.randint(18, 80, n_samples),
        'genero': np.random.choice(['M', 'F'], n_samples),
        'raca': np.random.choice(['Branco', 'Negro', 'Pardo', 'Asiático'], n_samples),
        'escolaridade': np.random.choice(['Fundamental', 'Médio', 'Superior'], n_samples),
        'renda': np.random.lognormal(10, 1, n_samples),
        'email': [f"user{i}@exemplo.com" for i in range(n_samples)]  # Dado sensível
    })
    
    # Simular predições com viés (favorece homens brancos)
    predictions = np.zeros(n_samples)
    for i in range(n_samples):
        base_prob = 0.5
        if data.loc[i, 'genero'] == 'M':
            base_prob += 0.2
        if data.loc[i, 'raca'] == 'Branco':
            base_prob += 0.15
        
        predictions[i] = np.random.binomial(1, base_prob)
    
    # Labels verdadeiros (sem viés)
    true_labels = np.random.binomial(1, 0.5, n_samples)
    
    # Executar auditoria
    auditor = EthicalAIAuditor("Modelo_Exemplo_Credito")
    
    # Verificar privacidade
    privacy_report = auditor.check_data_privacy(data, ['email'])
    
    # Analisar viés
    bias_report = auditor.analyze_bias(
        data, predictions, true_labels, 
        protected_attributes=['genero', 'raca']
    )
    
    # Gerar relatório final
    final_report = auditor.generate_comprehensive_report()
    
    return auditor, final_report

if __name__ == "__main__":
    # Executar demo
    auditor, report = demo_ethical_audit()
    
    print(f"\n📊 RESULTADO FINAL:")
    print(f"Score Geral de Ética: {report['executive_summary']['overall_ethics_score']}/100")
    print(f"Nível de Risco: {report['executive_summary']['risk_level']}")
    print(f"Recomendações: {report['executive_summary']['recommendation_count']}")