141：AI产品合规与伦理设计——法律风险防控的系统化方案

# 合规风险评估模型
class ComplianceRiskAssessor:
    def __init__(self):
        self.risk_factors = {
            "data_privacy": self.assess_data_privacy,
            "algorithm_fairness": self.assess_algorithm_fairness,
            "system_security": self.assess_system_security,
            "legal_compliance": self.assess_legal_compliance,
            "ethical_impact": self.assess_ethical_impact
        }
    
    def assess_risk(self, product_details):
        """评估产品合规风险"""
        risk_scores = {}
        for factor, assessor in self.risk_factors.items():
            risk_scores[factor] = assessor(product_details)
        
        # 计算综合风险评分
        overall_risk = sum(risk_scores.values()) / len(risk_scores)
        return {
            "detailed_scores": risk_scores,
            "overall_risk": overall_risk,
            "risk_level": self.get_risk_level(overall_risk)
        }
    
    def assess_data_privacy(self, product_details):
        """评估数据隐私风险"""
        # 实现数据隐私风险评估逻辑
        return 0.8  # 示例值
    
    def assess_algorithm_fairness(self, product_details):
        """评估算法公平性风险"""
        # 实现算法公平性风险评估逻辑
        return 0.6  # 示例值
    
    def assess_system_security(self, product_details):
        """评估系统安全风险"""
        # 实现系统安全风险评估逻辑
        return 0.7  # 示例值
    
    def assess_legal_compliance(self, product_details):
        """评估法律合规风险"""
        # 实现法律合规风险评估逻辑
        return 0.5  # 示例值
    
    def assess_ethical_impact(self, product_details):
        """评估伦理影响风险"""
        # 实现伦理影响风险评估逻辑
        return 0.4  # 示例值
    
    def get_risk_level(self, score):
        """根据评分确定风险等级"""
        if score >= 0.8:
            return "高风险"
        elif score >= 0.6:
            return "中等风险"
        else:
            return "低风险"

# 数据隐私保护实现
import hashlib
import json
from typing import Dict, Any

class DataPrivacyGuard:
    def __init__(self, config_file: str):
        self.privacy_rules = self.load_privacy_rules(config_file)
        self.data_mapping = {}
    
    def load_privacy_rules(self, config_file: str) -> Dict[str, str]:
        """加载隐私保护规则"""
        with open(config_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    
    def anonymize_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """数据匿名化处理"""
        anonymized_data = {}
        for field, value in data.items():
            if field in self.privacy_rules:
                rule = self.privacy_rules[field]
                anonymized_value = self.apply_rule(value, rule, field)
                anonymized_data[field] = anonymized_value
            else:
                anonymized_data[field] = value
        return anonymized_data
    
    def apply_rule(self, value: Any, rule: str, field: str) -> Any:
        """应用隐私保护规则"""
        if rule == "mask":
            if isinstance(value, str) and len(value) > 4:
                masked_value = "****" + value[-4:]
                # 保存映射关系用于后续去匿名化（仅在必要时）
                self.data_mapping[masked_value] = value
                return masked_value
            return "****"
        elif rule == "hash":
            if isinstance(value, str):
                hashed_value = hashlib.sha256(value.encode()).hexdigest()
                self.data_mapping[hashed_value] = value
                return hashed_value
            return value
        elif rule == "delete":
            return None
        elif rule == "generalize":
            # 实现数据泛化，如将具体年龄转换为年龄段
            if isinstance(value, int) and field == "age":
                return f"{value // 10 * 10}-{value // 10 * 10 + 9}"
            return value
        return value
    
    def de-anonymize_data(self, anonymized_data: Dict[str, Any]) -> Dict[str, Any]:
        """数据去匿名化（仅在授权情况下使用）"""
        de_anonymized_data = {}
        for field, value in anonymized_data.items():
            if value in self.data_mapping:
                de_anonymized_data[field] = self.data_mapping[value]
            else:
                de_anonymized_data[field] = value
        return de_anonymized_data

# 算法偏见检测
import pandas as pd
import numpy as np
from sklearn.metrics import confusion_matrix

class FairnessDetector:
    def __init__(self):
        self.metrics = {
            "equal_opportunity": self.calculate_equal_opportunity,
            "demographic_parity": self.calculate_demographic_parity,
            "equalized_odds": self.calculate_equalized_odds,
            "statistical_parity": self.calculate_statistical_parity,
            "disparate_impact": self.calculate_disparate_impact
        }
    
    def detect_bias(self, model, test_data: pd.DataFrame, protected_attributes: list) -> Dict[str, Dict[str, float]]:
        """检测算法偏见"""
        results = {}
        for attr in protected_attributes:
            attr_results = {}
            for metric_name, metric_func in self.metrics.items():
                attr_results[metric_name] = metric_func(model, test_data, attr)
            results[attr] = attr_results
        return results
    
    def calculate_equal_opportunity(self, model, test_data: pd.DataFrame, protected_attr: str) -> float:
        """计算机会均等率"""
        # 实现机会均等率计算逻辑
        # 对于每个受保护属性的群体，计算真阳性率的差异
        groups = test_data[protected_attr].unique()
        tprs = []
        
        for group in groups:
            group_data = test_data[test_data[protected_attr] == group]
            if len(group_data) == 0:
                continue
            
            y_true = group_data['label']
            y_pred = model.predict(group_data.drop(['label', protected_attr], axis=1))
            tn, fp, fn, tp = confusion_matrix(y_true, y_pred).ravel()
            tpr = tp / (tp + fn) if (tp + fn) > 0 else 0
            tprs.append(tpr)
        
        if len(tprs) < 2:
            return 0.0
        
        return max(tprs) - min(tprs)
    
    def calculate_demographic_parity(self, model, test_data: pd.DataFrame, protected_attr: str) -> float:
        """计算人口统计学 parity"""
        # 实现人口统计学 parity 计算逻辑
        groups = test_data[protected_attr].unique()
        positive_rates = []
        
        for group in groups:
            group_data = test_data[test_data[protected_attr] == group]
            if len(group_data) == 0:
                continue
            
            y_pred = model.predict(group_data.drop(['label', protected_attr], axis=1))
            positive_rate = sum(y_pred) / len(y_pred)
            positive_rates.append(positive_rate)
        
        if len(positive_rates) < 2:
            return 0.0
        
        return max(positive_rates) - min(positive_rates)
    
    def calculate_equalized_odds(self, model, test_data: pd.DataFrame, protected_attr: str) -> float:
        """计算等比优势"""
        # 实现等比优势计算逻辑
        groups = test_data[protected_attr].unique()
        tprs = []
        fprs = []
        
        for group in groups:
            group_data = test_data[test_data[protected_attr] == group]
            if len(group_data) == 0:
                continue
            
            y_true = group_data['label']
            y_pred = model.predict(group_data.drop(['label', protected_attr], axis=1))
            tn, fp, fn, tp = confusion_matrix(y_true, y_pred).ravel()
            tpr = tp / (tp + fn) if (tp + fn) > 0 else 0
            fpr = fp / (fp + tn) if (fp + tn) > 0 else 0
            tprs.append(tpr)
            fprs.append(fpr)
        
        if len(tprs) < 2 or len(fprs) < 2:
            return 0.0
        
        tpr_diff = max(tprs) - min(tprs)
        fpr_diff = max(fprs) - min(fprs)
        return max(tpr_diff, fpr_diff)
    
    def calculate_statistical_parity(self, model, test_data: pd.DataFrame, protected_attr: str) -> float:
        """计算统计 parity"""
        # 实现统计 parity 计算逻辑
        return self.calculate_demographic_parity(model, test_data, protected_attr)
    
    def calculate_disparate_impact(self, model, test_data: pd.DataFrame, protected_attr: str) -> float:
        """计算差异影响"""
        # 实现差异影响计算逻辑
        groups = test_data[protected_attr].unique()
        positive_rates = []
        
        for group in groups:
            group_data = test_data[test_data[protected_attr] == group]
            if len(group_data) == 0:
                continue
            
            y_pred = model.predict(group_data.drop(['label', protected_attr], axis=1))
            positive_rate = sum(y_pred) / len(y_pred)
            positive_rates.append(positive_rate)
        
        if len(positive_rates) < 2:
            return 1.0
        
        min_rate = min(positive_rates)
        max_rate = max(positive_rates)
        return min_rate / max_rate if max_rate > 0 else 0.0

# 合规监控与审计系统
import logging
import time
from datetime import datetime
from typing import Dict, List, Any

class ComplianceMonitor:
    def __init__(self, log_file: str):
        self.log_file = log_file
        self.setup_logger()
        self.compliance_events = []
    
    def setup_logger(self):
        """设置日志记录器"""
        logging.basicConfig(
            filename=self.log_file,
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger('compliance_monitor')
    
    def log_event(self, event_type: str, details: Dict[str, Any]):
        """记录合规事件"""
        event = {
            'timestamp': datetime.now().isoformat(),
            'event_type': event_type,
            'details': details
        }
        self.compliance_events.append(event)
        self.logger.info(json.dumps(event))
    
    def monitor_data_access(self, user_id: str, data_type: str, access_action: str):
        """监控数据访问"""
        event_details = {
            'user_id': user_id,
            'data_type': data_type,
            'access_action': access_action,
            'ip_address': self.get_ip_address()
        }
        self.log_event('data_access', event_details)
    
    def monitor_algorithm_decision(self, model_id: str, decision: Any, confidence: float, input_data: Dict[str, Any]):
        """监控算法决策"""
        event_details = {
            'model_id': model_id,
            'decision': decision,
            'confidence': confidence,
            'input_data': self.anonymize_input(input_data)
        }
        self.log_event('algorithm_decision', event_details)
    
    def monitor_compliance_check(self, check_type: str, result: bool, details: Dict[str, Any]):
        """监控合规检查"""
        event_details = {
            'check_type': check_type,
            'result': result,
            'details': details
        }
        self.log_event('compliance_check', event_details)
    
    def get_ip_address(self) -> str:
        """获取IP地址"""
        # 实现获取IP地址的逻辑
        return "127.0.0.1"  # 示例值
    
    def anonymize_input(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """匿名化输入数据"""
        # 实现输入数据匿名化逻辑
        anonymized = {}
        for key, value in input_data.items():
            if key in ['name', 'email', 'phone']:
                anonymized[key] = "[ANONYMIZED]"
            else:
                anonymized[key] = value
        return anonymized
    
    def generate_compliance_report(self, start_time: datetime, end_time: datetime) -> Dict[str, Any]:
        """生成合规报告"""
        filtered_events = [
            event for event in self.compliance_events
            if start_time <= datetime.fromisoformat(event['timestamp']) <= end_time
        ]
        
        report = {
            'report_time': datetime.now().isoformat(),
            'period': {
                'start': start_time.isoformat(),
                'end': end_time.isoformat()
            },
            'event_summary': self.summarize_events(filtered_events),
            'compliance_status': self.assess_compliance_status(filtered_events),
            'recommendations': self.generate_recommendations(filtered_events)
        }
        
        # 保存报告
        report_file = f"compliance_report_{start_time.strftime('%Y%m%d')}_{end_time.strftime('%Y%m%d')}.json"
        with open(report_file, 'w', encoding='utf-8') as f:
            json.dump(report, f, indent=2, ensure_ascii=False)
        
        return report
    
    def summarize_events(self, events: List[Dict[str, Any]]) -> Dict[str, int]:
        """汇总事件"""
        summary = {}
        for event in events:
            event_type = event['event_type']
            if event_type not in summary:
                summary[event_type] = 0
            summary[event_type] += 1
        return summary
    
    def assess_compliance_status(self, events: List[Dict[str, Any]]) -> str:
        """评估合规状态"""
        # 实现合规状态评估逻辑
        compliance_checks = [
            event for event in events
            if event['event_type'] == 'compliance_check'
        ]
        
        if not compliance_checks:
            return "未进行合规检查"
        
        failed_checks = [
            check for check in compliance_checks
            if not check['details']['result']
        ]
        
        if len(failed_checks) == 0:
            return "合规"
        elif len(failed_checks) < len(compliance_checks) / 2:
            return "基本合规"
        else:
            return "不合规"
    
    def generate_recommendations(self, events: List[Dict[str, Any]]) -> List[str]:
        """生成改进建议"""
        # 实现生成改进建议的逻辑
        recommendations = []
        
        # 检查数据访问模式
        data_access_events = [
            event for event in events
            if event['event_type'] == 'data_access'
        ]
        
        if len(data_access_events) > 1000:
            recommendations.append("数据访问频率过高，建议优化访问控制")
        
        # 检查算法决策
        algorithm_events = [
            event for event in events
            if event['event_type'] == 'algorithm_decision'
        ]
        
        low_confidence_decisions = [
            event for event in algorithm_events
            if event['details'].get('confidence', 1.0) < 0.7
        ]
        
        if len(low_confidence_decisions) > 0:
            recommendations.append("存在低置信度决策，建议优化模型")
        
        return recommendations