163_伦理前沿：AI公平性审计 - 2025年偏见审计工具的独特框架设计与实施指南

AI公平性
├── 群体公平性
│   ├── 统计平等
│   ├── 等错误率
│   ├── 机会平等
│   └── 预测均等
├── 个体公平性
│   ├── 相似性原则
│   └── 一致性原则
└── 社会公平性
    ├── 分配公平
    └── 程序公平

P(\hat{Y}=1|A=a_1) = P(\hat{Y}=1|A=a_2)

社会偏见 → 数据收集 → 代表性/历史/标注/采样偏见 → 模型训练 → 设计/优化/交互偏见 → 系统部署 → 部署/解释/滥用偏见 → 社会影响 → 社会偏见(循环强化)

公平性评估指标
├── 差异指标
│   ├── 差异率 → 群体间比例比较
│   ├── 等机会差异 → 真阳性率差值
│   └── 预测均等差异 → 阳性预测值差值
├── 公平性得分
│   ├── 综合得分 → 多维度评估
│   └── 缓解效率 → 措施有效性
└── 统计测试
    ├── 卡方检验 → 大样本显著性
    └── Fisher精确检验 → 小样本显著性

# 公平性评估指标计算示例
import pandas as pd
import numpy as np
from scipy import stats

def calculate_demographic_parity_ratio(y_true, y_pred, protected_attr):
    """计算不同受保护属性群体间的统计平等比率"""
    df = pd.DataFrame({
        'y_true': y_true,
        'y_pred': y_pred,
        'protected_attr': protected_attr
    })
    
    # 计算每个群体的正预测率
    positive_rates = df.groupby('protected_attr')['y_pred'].mean()
    
    # 计算比率 (最大值/最小值)
    if len(positive_rates) >= 2:
        return positive_rates.max() / positive_rates.min()
    return 1.0

def calculate_equal_opportunity_difference(y_true, y_pred, protected_attr):
    """计算不同受保护属性群体间的等机会差异"""
    df = pd.DataFrame({
        'y_true': y_true,
        'y_pred': y_pred,
        'protected_attr': protected_attr
    })
    
    # 计算每个群体的真阳性率
    tprs = {}
    for attr_value in df['protected_attr'].unique():
        group = df[df['protected_attr'] == attr_value]
        if group['y_true'].sum() > 0:  # 避免除以零
            tpr = (group['y_true'] & group['y_pred']).sum() / group['y_true'].sum()
            tprs[attr_value] = tpr
    
    # 计算差异
    if len(tprs) >= 2:
        return max(tprs.values()) - min(tprs.values())
    return 0.0

def perform_statistical_test(y_pred, protected_attr):
    """对预测结果和受保护属性进行卡方检验"""
    contingency_table = pd.crosstab(protected_attr, y_pred)
    chi2, p_value, dof, expected = stats.chi2_contingency(contingency_table)
    return chi2, p_value

# 多目标优化示例：同时考虑准确性和公平性
import torch
import torch.nn as nn
def multi_objective_loss(y_true, y_pred, protected_attr, alpha=0.5):
    """
    多目标损失函数，结合交叉熵损失和公平性惩罚
    
    参数:
    y_true: 真实标签
    y_pred: 预测概率
    protected_attr: 受保护属性
    alpha: 公平性权重，范围[0,1]
    """
    # 交叉熵损失
    ce_loss = nn.functional.cross_entropy(y_pred, y_true)
    
    # 公平性惩罚：基于群体统计平等
    group_probs = []
    for attr_value in torch.unique(protected_attr):
        mask = (protected_attr == attr_value)
        if mask.sum() > 0:
            group_prob = y_pred[mask][:, 1].mean()  # 正类的平均概率
            group_probs.append(group_prob)
    
    # 计算组间差异作为公平性惩罚
    if len(group_probs) >= 2:
        group_diff = torch.tensor(0.0)
        for i in range(len(group_probs)):
            for j in range(i+1, len(group_probs)):
                group_diff += torch.abs(group_probs[i] - group_probs[j])
        fairness_penalty = group_diff / (len(group_probs) * (len(group_probs) - 1) / 2)
    else:
        fairness_penalty = torch.tensor(0.0)
    
    # 组合损失
    total_loss = (1 - alpha) * ce_loss + alpha * fairness_penalty
    return total_loss

治理层 (政策、合规、风险管理)
  ↑
应用层 (部署审计、运行时监控、反馈)
  ↑
模型层 (训练监控、公平性评估、偏见归因)
  ↑
数据层 (数据收集、多样性分析、质量评估)

class BiasDetectionEngine:
    """偏见检测引擎，用于识别数据和模型中的各种偏见"""
    
    def __init__(self, sensitive_attributes=None):
        """
        初始化偏见检测引擎
        
        参数:
        sensitive_attributes: 需要关注的敏感属性列表
        """
        self.sensitive_attributes = sensitive_attributes or ['gender', 'race', 'age']
    
    def detect_data_bias(self, data, target_variable):
        """
        检测训练数据中的偏见
        
        参数:
        data: 包含特征和目标变量的DataFrame
        target_variable: 目标变量名称
        
        返回:
        包含数据偏见分析结果的字典
        """
        results = {}
        
        # 检测代表性偏见
        for attr in self.sensitive_attributes:
            if attr in data.columns:
                # 计算不同群体的样本分布
                distribution = data[attr].value_counts(normalize=True).to_dict()
                
                # 计算目标变量在不同群体中的分布差异
                target_distributions = {}
                for value in data[attr].unique():
                    mask = (data[attr] == value)
                    if mask.sum() > 0:
                        target_dist = data.loc[mask, target_variable].mean()
                        target_distributions[value] = target_dist
                
                # 计算差异指标
                if len(target_distributions) >= 2:
                    values = list(target_distributions.values())
                    max_diff = max(values) - min(values)
                    
                    results[attr] = {
                        'distribution': distribution,
                        'target_distributions': target_distributions,
                        'max_difference': max_diff,
                        'bias_level': 'high' if max_diff > 0.2 else 'medium' if max_diff > 0.1 else 'low'
                    }
        
        return results
    
    def detect_model_bias(self, model, data, target_variable):
        """
        检测模型中的偏见
        
        参数:
        model: 要评估的模型
        data: 测试数据
        target_variable: 目标变量名称
        
        返回:
        包含模型偏见分析结果的字典
        """
        # 获取特征列（排除目标变量和敏感属性）
        feature_columns = [col for col in data.columns if col not in [target_variable] + self.sensitive_attributes]
        
        # 进行预测
        X_test = data[feature_columns]
        y_pred = model.predict(X_test)
        
        # 计算不同敏感属性的公平性指标
        results = {}
        for attr in self.sensitive_attributes:
            if attr in data.columns:
                # 计算统计平等比率
                demographic_parity = self._calculate_demographic_parity(y_pred, data[attr])
                
                # 计算等机会差异
                equal_opportunity = self._calculate_equal_opportunity(y_pred, data[target_variable], data[attr])
                
                # 计算预测均等差异
                predictive_parity = self._calculate_predictive_parity(y_pred, data[target_variable], data[attr])
                
                results[attr] = {
                    'demographic_parity_ratio': demographic_parity,
                    'equal_opportunity_difference': equal_opportunity,
                    'predictive_parity_difference': predictive_parity,
                    'overall_bias_score': (demographic_parity + equal_opportunity + predictive_parity) / 3
                }
        
        return results
    
    def _calculate_demographic_parity(self, y_pred, protected_attr):
        """计算统计平等比率"""
        import pandas as pd
        df = pd.DataFrame({'y_pred': y_pred, 'protected_attr': protected_attr})
        rates = df.groupby('protected_attr')['y_pred'].mean()
        return rates.max() / rates.min() if len(rates) >= 2 else 1.0
    
    def _calculate_equal_opportunity(self, y_pred, y_true, protected_attr):
        """计算等机会差异"""
        import pandas as pd
        df = pd.DataFrame({'y_pred': y_pred, 'y_true': y_true, 'protected_attr': protected_attr})
        
        tprs = {}
        for value in df['protected_attr'].unique():
            group = df[df['protected_attr'] == value]
            if group['y_true'].sum() > 0:
                tpr = (group['y_true'] & group['y_pred']).sum() / group['y_true'].sum()
                tprs[value] = tpr
        
        return max(tprs.values()) - min(tprs.values()) if len(tprs) >= 2 else 0.0
    
    def _calculate_predictive_parity(self, y_pred, y_true, protected_attr):
        """计算预测均等差异"""
        import pandas as pd
        df = pd.DataFrame({'y_pred': y_pred, 'y_true': y_true, 'protected_attr': protected_attr})
        
        ppvs = {}
        for value in df['protected_attr'].unique():
            group = df[df['protected_attr'] == value]
            if group['y_pred'].sum() > 0:
                ppv = (group['y_true'] & group['y_pred']).sum() / group['y_pred'].sum()
                ppvs[value] = ppv
        
        return max(ppvs.values()) - min(ppvs.values()) if len(ppvs) >= 2 else 0.0

开始 → 数据输入
  ↓
敏感属性识别
  ↓
┌────────────┐    ┌───────────────┐
│数据偏见检测│    │ 模型偏见检测  │
│            │    │               │
│1. 分布分析 │    │1. 模型预测    │
│2. 目标差异 │───→│2. 统计平等计算│
│3. 偏见评级 │    │3. 等机会计算  │
└────┬───────┘    │4. 预测均等计算│
     │            └────────┬──────┘
     └────────────────────┼───┐
                         ↓   │
                  偏见综合分析│
                         ↓   │
                  生成审计报告│
                         ↓
                        结束

class FairnessQuantificationModule:
    """公平性量化模块，用于综合评估AI系统的公平性"""
    
    def __init__(self, metrics_config=None):
        """
        初始化公平性量化模块
        
        参数:
        metrics_config: 指标配置，包括权重和阈值
        """
        self.metrics_config = metrics_config or {
            'demographic_parity': {'weight': 0.3, 'threshold': 1.2},
            'equal_opportunity': {'weight': 0.4, 'threshold': 0.05},
            'predictive_parity': {'weight': 0.3, 'threshold': 0.05}
        }
    
    def calculate_overall_fairness_score(self, bias_results):
        """
        计算系统的总体公平性得分
        
        参数:
        bias_results: 偏见检测结果
        
        返回:
        综合公平性得分和详细评估
        """
        overall_scores = {}
        
        # 对每个敏感属性计算公平性得分
        for attr, metrics in bias_results.items():
            # 计算各项指标的标准化得分（0-1）
            scores = {
                'demographic_parity': self._normalize_demographic_parity(
                    metrics['demographic_parity_ratio'],
                    self.metrics_config['demographic_parity']['threshold']
                ),
                'equal_opportunity': self._normalize_difference(
                    metrics['equal_opportunity_difference'],
                    self.metrics_config['equal_opportunity']['threshold']
                ),
                'predictive_parity': self._normalize_difference(
                    metrics['predictive_parity_difference'],
                    self.metrics_config['predictive_parity']['threshold']
                )
            }
            
            # 计算加权总分
            weighted_score = sum(
                scores[metric] * config['weight']
                for metric, config in self.metrics_config.items()
            )
            
            # 确定公平性等级
            if weighted_score >= 0.8:
                level = 'Excellent'
            elif weighted_score >= 0.6:
                level = 'Good'
            elif weighted_score >= 0.4:
                level = 'Fair'
            else:
                level = 'Poor'
            
            overall_scores[attr] = {
                'detailed_scores': scores,
                'weighted_score': weighted_score,
                'level': level
            }
        
        # 计算总体得分（所有属性的平均值）
        total_weighted_score = sum(score['weighted_score'] for score in overall_scores.values()) / len(overall_scores)
        total_level = self._get_fairness_level(total_weighted_score)
        
        return {
            'attribute_scores': overall_scores,
            'overall_score': total_weighted_score,
            'overall_level': total_level
        }
    
    def _normalize_demographic_parity(self, ratio, threshold):
        """标准化统计平等比率（1表示完全平等，值越大表示越不平等）"""
        if ratio <= 1:
            return 1.0  # 比率<=1视为完全平等
        elif ratio <= threshold:
            # 在1和阈值之间线性映射
            return 1 - (ratio - 1) / (threshold - 1)
        else:
            # 超过阈值，得分下降更快
            return max(0, 0.5 - (ratio - threshold) / threshold)
    
    def _normalize_difference(self, difference, threshold):
        """标准化差异指标（0表示无差异，值越大表示差异越大）"""
        if difference <= 0:
            return 1.0  # 无差异视为完全平等
        elif difference <= threshold:
            # 在0和阈值之间线性映射
            return 1 - difference / threshold
        else:
            # 超过阈值，得分下降更快
            return max(0, 0.5 - (difference - threshold) / threshold)
    
    def _get_fairness_level(self, score):
        """根据得分确定公平性等级"""
        if score >= 0.8:
            return 'Excellent'
        elif score >= 0.6:
            return 'Good'
        elif score >= 0.4:
            return 'Fair'
        else:
            return 'Poor'
    
    def generate_visualization(self, fairness_results):
        """
        生成公平性可视化图表
        
        参数:
        fairness_results: 公平性评估结果
        
        返回:
        可视化图表的HTML代码
        """
        # 这里应该返回实际的可视化代码，例如使用Plotly、Dash等
        # 由于篇幅限制，这里只提供示例框架
        return "<div>Fairness Visualization</div>"

┌───────────────────────────────────────────────────────────────────┐
│ 公平性评分分布                评分区间：0.0-1.0                     │
├─────────────┬──────────────┬───────────────┬──────────────────────┤
│ Excellent   │      Good    │     Fair      │        Poor          │
│ [0.8-1.0]   │   [0.6-0.8)  │   [0.4-0.6)   │      [0.0-0.4)       │
│             │              │               │                      │
│     ████    │     ████     │     ████      │        ████          │
│     ████    │     ████     │     ████      │        ████          │
│     ████    │     ████     │     ████      │        ████          │
│             │              │               │                      │
│ 权重计算：统计平等(0.3)+等机会(0.4)+预测均等(0.3)                   │
└─────────────┴──────────────┴───────────────┴──────────────────────┘

class BiasMitigationStrategies:
    """偏见缓解策略库，提供多种减少AI系统偏见的方法"""
    
    @staticmethod
    def reweighting(data, sensitive_attr, target, method='demographic_parity'):
        """
        数据重加权方法
        
        参数:
        data: 训练数据
        sensitive_attr: 敏感属性列名
        target: 目标变量列名
        method: 公平性目标方法
        
        返回:
        重加权后的数据和权重
        """
        import pandas as pd
        import numpy as np
        
        # 创建交叉表
        cross_tab = pd.crosstab(data[sensitive_attr], data[target])
        
        # 计算理想分布
        if method == 'demographic_parity':
            # 统计平等：不同群体的正结果率相同
            target_dist = data[target].value_counts(normalize=True)
            attr_dist = data[sensitive_attr].value_counts(normalize=True)
            
            # 计算理想分布
            ideal_dist = pd.DataFrame(index=cross_tab.index, columns=cross_tab.columns)
            for attr_val in cross_tab.index:
                for target_val in cross_tab.columns:
                    ideal_dist.loc[attr_val, target_val] = attr_dist[attr_val] * target_dist[target_val]
        else:
            # 其他方法的实现...
            ideal_dist = cross_tab / len(data)
        
        # 计算权重
        actual_dist = cross_tab / len(data)
        weights = pd.Series(index=data.index, dtype=float)
        
        for idx, row in data.iterrows():
            attr_val = row[sensitive_attr]
            target_val = row[target]
            if actual_dist.loc[attr_val, target_val] > 0:
                weights[idx] = ideal_dist.loc[attr_val, target_val] / actual_dist.loc[attr_val, target_val]
            else:
                weights[idx] = 1.0
        
        return data, weights
    
    @staticmethod
    def adversarial_debiasing(X, y, sensitive_attr, model_class, n_epochs=50):
        """
        对抗性去偏方法
        
        参数:
        X: 特征数据
        y: 目标变量
        sensitive_attr: 敏感属性
        model_class: 基础模型类
        n_epochs: 训练轮数
        
        返回:
        去偏后的模型
        """
        import torch
        import torch.nn as nn
        import torch.optim as optim
        
        # 创建预测模型
        predictor = model_class(X.shape[1])
        # 创建对抗模型
        adversary = nn.Sequential(
            nn.Linear(1, 64),  # 输入是预测器的输出
            nn.ReLU(),
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.Linear(32, 1),
            nn.Sigmoid()
        )
        
        # 优化器
        predictor_optimizer = optim.Adam(predictor.parameters(), lr=0.001)
        adversary_optimizer = optim.Adam(adversary.parameters(), lr=0.001)
        
        # 损失函数
        predictor_criterion = nn.BCELoss()
        adversary_criterion = nn.BCELoss()
        
        # 训练循环
        for epoch in range(n_epochs):
            # 前向传播
            y_pred = predictor(X)
            
            # 预测器的损失（既要准确预测目标，又要防止对抗器预测敏感属性）
            predictor_loss = predictor_criterion(y_pred, y) - adversary_criterion(adversary(y_pred), sensitive_attr)
            
            # 更新预测器
            predictor_optimizer.zero_grad()
            predictor_loss.backward(retain_graph=True)
            predictor_optimizer.step()
            
            # 对抗器的损失（尝试从预测中推断敏感属性）
            adversary_loss = adversary_criterion(adversary(y_pred.detach()), sensitive_attr)
            
            # 更新对抗器
            adversary_optimizer.zero_grad()
            adversary_loss.backward()
            adversary_optimizer.step()
        
        return predictor
    
    @staticmethod
    def postprocessing_adjustment(y_pred_proba, sensitive_attr, calibration_method='equal_odds'):
        """
        后处理调整方法
        
        参数:
        y_pred_proba: 预测概率
        sensitive_attr: 敏感属性
        calibration_method: 校准方法
        
        返回:
        调整后的预测结果
        """
        import pandas as pd
        import numpy as np
        
        # 创建数据框
        df = pd.DataFrame({
            'y_pred_proba': y_pred_proba,
            'sensitive_attr': sensitive_attr
        })
        
        # 计算不同群体的校准阈值
        thresholds = {}
        for attr_val in df['sensitive_attr'].unique():
            group = df[df['sensitive_attr'] == attr_val]
            
            if calibration_method == 'equal_odds':
                # 等错误率校准（简化版本）
                # 实际实现需要更复杂的ROC分析
                thresholds[attr_val] = group['y_pred_proba'].median()
            else:
                # 默认阈值
                thresholds[attr_val] = 0.5
        
        # 应用阈值调整
        y_pred_adjusted = np.zeros_like(y_pred_proba)
        for i, (proba, attr_val) in enumerate(zip(y_pred_proba, sensitive_attr)):
            y_pred_adjusted[i] = 1 if proba >= thresholds[attr_val] else 0
        
        return y_pred_adjusted

class DataPreprocessor:
    """数据预处理模块，负责数据的加载、清洗和转换"""
    
    @staticmethod
    def load_data(source, source_type='csv', **kwargs):
        """
        从不同来源加载数据
        
        参数:
        source: 数据源（文件路径、数据库连接字符串等）
        source_type: 数据源类型
        **kwargs: 额外参数
        
        返回:
        加载的数据
        """
        import pandas as pd
        
        if source_type == 'csv':
            return pd.read_csv(source, **kwargs)
        elif source_type == 'excel':
            return pd.read_excel(source, **kwargs)
        elif source_type == 'database':
            import sqlite3
            conn = sqlite3.connect(source)
            query = kwargs.get('query', 'SELECT * FROM data')
            return pd.read_sql(query, conn)
        elif source_type == 'json':
            return pd.read_json(source, **kwargs)
        else:
            raise ValueError(f"不支持的数据源类型: {source_type}")
    
    @staticmethod
    def clean_data(data, sensitive_attributes=None):
        """
        清洗数据，处理缺失值、异常值等
        
        参数:
        data: 输入数据
        sensitive_attributes: 敏感属性列表
        
        返回:
        清洗后的数据
        """
        import pandas as pd
        
        # 复制数据以避免修改原始数据
        cleaned_data = data.copy()
        
        # 处理缺失值
        # 对数值型数据使用中位数填充
        numeric_cols = cleaned_data.select_dtypes(include=['int64', 'float64']).columns
        for col in numeric_cols:
            if cleaned_data[col].isnull().any():
                cleaned_data[col].fillna(cleaned_data[col].median(), inplace=True)
        
        # 对分类型数据使用众数填充
        categorical_cols = cleaned_data.select_dtypes(include=['object', 'category']).columns
        for col in categorical_cols:
            if cleaned_data[col].isnull().any():
                cleaned_data[col].fillna(cleaned_data[col].mode()[0], inplace=True)
        
        # 特殊处理敏感属性
        if sensitive_attributes:
            for attr in sensitive_attributes:
                if attr in cleaned_data.columns:
                    # 确保敏感属性的值是离散的、有意义的
                    if cleaned_data[attr].nunique() > 10:  # 如果唯一值太多
                        # 可以考虑分箱或其他处理方式
                        cleaned_data[attr] = pd.qcut(cleaned_data[attr], q=5)
        
        # 移除重复行
        cleaned_data.drop_duplicates(inplace=True)
        
        return cleaned_data
    
    @staticmethod
    def encode_categorical(data, categorical_columns=None, encoding_type='onehot'):
        """
        对分类型特征进行编码
        
        参数:
        data: 输入数据
        categorical_columns: 要编码的分类型列名列表
        encoding_type: 编码类型
        
        返回:
        编码后的数据
        """
        import pandas as pd
        from sklearn.preprocessing import LabelEncoder, OneHotEncoder
        
        # 复制数据
        encoded_data = data.copy()
        
        # 如果没有指定分类列，自动检测
        if not categorical_columns:
            categorical_columns = encoded_data.select_dtypes(include=['object', 'category']).columns
        
        if encoding_type == 'label':
            # 标签编码
            le = LabelEncoder()
            for col in categorical_columns:
                if col in encoded_data.columns:
                    encoded_data[col] = le.fit_transform(encoded_data[col])
        elif encoding_type == 'onehot':
            # 独热编码
            encoded_data = pd.get_dummies(encoded_data, columns=categorical_columns, drop_first=True)
        else:
            raise ValueError(f"不支持的编码类型: {encoding_type}")
        
        return encoded_data
    
    @staticmethod
    def split_data(data, target_variable, test_size=0.2, random_state=42):
        """
        分割训练集和测试集
        
        参数:
        data: 输入数据
        target_variable: 目标变量列名
        test_size: 测试集比例
        random_state: 随机种子
        
        返回:
        训练集和测试集（X_train, X_test, y_train, y_test）
        """
        from sklearn.model_selection import train_test_split
        
        # 分离特征和目标变量
        X = data.drop(target_variable, axis=1)
        y = data[target_variable]
        
        # 分割数据
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=test_size, random_state=random_state
        )
        
        return X_train, X_test, y_train, y_test

class ModelAdapter:
    """模型适配器模块，支持不同类型的AI模型"""
    
    @staticmethod
    def adapt_model(model, model_type=None):
        """
        适配不同类型的模型
        
        参数:
        model: 原始模型
        model_type: 模型类型（自动检测或手动指定）
        
        返回:
        适配后的统一模型接口
        """
        # 自动检测模型类型
        if model_type is None:
            model_type = ModelAdapter._detect_model_type(model)
        
        # 根据模型类型选择合适的适配器
        if model_type == 'sklearn':
            return SklearnModelAdapter(model)
        elif model_type == 'tensorflow':
            return TensorFlowModelAdapter(model)
        elif model_type == 'pytorch':
            return PyTorchModelAdapter(model)
        elif model_type == 'custom':
            return CustomModelAdapter(model)
        else:
            raise ValueError(f"不支持的模型类型: {model_type}")
    
    @staticmethod
    def _detect_model_type(model):
        """
        自动检测模型类型
        
        参数:
        model: 要检测的模型
        
        返回:
        模型类型
        """
        # 检查是否为sklearn模型
        try:
            import sklearn.base
            if isinstance(model, sklearn.base.BaseEstimator):
                return 'sklearn'
        except ImportError:
            pass
        
        # 检查是否为tensorflow模型
        try:
            import tensorflow as tf
            if isinstance(model, tf.keras.Model):
                return 'tensorflow'
        except ImportError:
            pass
        
        # 检查是否为pytorch模型
        try:
            import torch.nn
            if isinstance(model, torch.nn.Module):
                return 'pytorch'
        except ImportError:
            pass
        
        # 检查是否有必要的方法
        if hasattr(model, 'predict'):
            return 'custom'
        
        raise ValueError("无法识别模型类型，请确保模型有predict方法或手动指定模型类型")


class BaseModelAdapter:
    """模型适配器基类"""
    
    def __init__(self, model):
        self.model = model
    
    def predict(self, X):
        """预测方法"""
        raise NotImplementedError
    
    def predict_proba(self, X):
        """预测概率方法"""
        raise NotImplementedError


class SklearnModelAdapter(BaseModelAdapter):
    """sklearn模型适配器"""
    
    def predict(self, X):
        """调用sklearn模型的predict方法"""
        return self.model.predict(X)
    
    def predict_proba(self, X):
        """调用sklearn模型的predict_proba方法"""
        if hasattr(self.model, 'predict_proba'):
            return self.model.predict_proba(X)
        elif hasattr(self.model, 'decision_function'):
            # 对于SVC等模型，使用decision_function
            import numpy as np
            scores = self.model.decision_function(X)
            # 转换为概率（简化处理）
            if len(scores.shape) == 1:
                # 二分类情况
                from sklearn.preprocessing import MinMaxScaler
                scaler = MinMaxScaler()
                prob_pos = scaler.fit_transform(scores.reshape(-1, 1)).flatten()
                return np.vstack([1 - prob_pos, prob_pos]).T
            else:
                # 多分类情况
                from sklearn.preprocessing import softmax
                return softmax(scores, axis=1)
        else:
            raise ValueError("该模型不支持概率预测")


class TensorFlowModelAdapter(BaseModelAdapter):
    """TensorFlow模型适配器"""
    
    def predict(self, X):
        """调用TensorFlow模型的predict方法并转换结果"""
        import numpy as np
        predictions = self.model.predict(X)
        # 对于分类问题，返回类别而不是概率
        if len(predictions.shape) > 1 and predictions.shape[1] > 1:
            return np.argmax(predictions, axis=1)
        else:
            # 对于二分类或回归问题
            return (predictions > 0.5).astype(int).flatten() if predictions.shape[1] == 1 else predictions
    
    def predict_proba(self, X):
        """获取预测概率"""
        import numpy as np
        predictions = self.model.predict(X)
        # 检查是否已经是概率
        if len(predictions.shape) > 1:
            # 对于多分类，可能已经是softmax输出
            if predictions.shape[1] > 1:
                return predictions
            else:
                # 对于二分类，补充负类概率
                prob_pos = predictions.flatten()
                return np.vstack([1 - prob_pos, prob_pos]).T
        else:
            raise ValueError("该模型不直接输出概率")


class PyTorchModelAdapter(BaseModelAdapter):
    """PyTorch模型适配器"""
    
    def predict(self, X):
        """调用PyTorch模型进行预测"""
        import torch
        import numpy as np
        
        # 确保模型在评估模式
        self.model.eval()
        
        # 转换为tensor（如果需要）
        if not isinstance(X, torch.Tensor):
            X_tensor = torch.tensor(X.values if hasattr(X, 'values') else X, dtype=torch.float32)
        else:
            X_tensor = X
        
        # 进行预测
        with torch.no_grad():
            outputs = self.model(X_tensor)
            
        # 转换为numpy数组
        if isinstance(outputs, torch.Tensor):
            outputs = outputs.numpy()
        
        # 返回预测类别
        if len(outputs.shape) > 1 and outputs.shape[1] > 1:
            return np.argmax(outputs, axis=1)
        else:
            return (outputs > 0.5).astype(int).flatten() if outputs.ndim > 1 else (outputs > 0.5).astype(int)
    
    def predict_proba(self, X):
        """获取预测概率"""
        import torch
        import numpy as np
        import torch.nn.functional as F
        
        # 确保模型在评估模式
        self.model.eval()
        
        # 转换为tensor（如果需要）
        if not isinstance(X, torch.Tensor):
            X_tensor = torch.tensor(X.values if hasattr(X, 'values') else X, dtype=torch.float32)
        else:
            X_tensor = X
        
        # 进行预测
        with torch.no_grad():
            outputs = self.model(X_tensor)
        
        # 应用softmax获取概率
        if isinstance(outputs, torch.Tensor):
            probabilities = F.softmax(outputs, dim=1).numpy()
        else:
            # 如果输出已经是概率（不太可能）
            probabilities = outputs
        
        return probabilities


class CustomModelAdapter(BaseModelAdapter):
    """自定义模型适配器"""
    
    def predict(self, X):
        """调用自定义模型的predict方法"""
        return self.model.predict(X)
    
    def predict_proba(self, X):
        """尝试获取概率预测"""
        if hasattr(self.model, 'predict_proba'):
            return self.model.predict_proba(X)
        elif hasattr(self.model, 'predict'):
            # 如果只有predict方法，尝试生成伪概率
            import numpy as np
            predictions = self.model.predict(X)
            # 将类别预测转换为one-hot编码作为伪概率
            if len(np.unique(predictions)) <= 2:
                # 二分类
                prob_pos = np.zeros_like(predictions, dtype=float)
                prob_pos[predictions == 1] = 1.0
                return np.vstack([1 - prob_pos, prob_pos]).T
            else:
                # 多分类
                n_classes = len(np.unique(predictions))
                n_samples = len(predictions)
                probabilities = np.zeros((n_samples, n_classes))
                for i, pred in enumerate(predictions):
                    probabilities[i, int(pred)] = 1.0
                return probabilities
        else:
            raise ValueError("该模型不支持预测")

class VisualizationModule:
    """可视化模块，生成公平性分析的各种图表"""
    
    @staticmethod
    def plot_demographic_parity(data, sensitive_attr, target_variable, title="统计平等分析"):
        """
        绘制统计平等分析图表
        
        参数:
        data: 包含敏感属性和目标变量的数据
        sensitive_attr: 敏感属性列名
        target_variable: 目标变量列名
        title: 图表标题
        
        返回:
        图表对象或HTML代码
        """
        import pandas as pd
        import plotly.express as px
        import plotly.graph_objects as go
        
        # 计算不同群体的正结果率
        parity_data = data.groupby(sensitive_attr)[target_variable].mean().reset_index()
        parity_data.columns = [sensitive_attr, 'positive_rate']
        
        # 创建柱状图
        fig = px.bar(
            parity_data, 
            x=sensitive_attr, 
            y='positive_rate',
            title=title,
            labels={'positive_rate': '正结果率'},
            color_discrete_sequence=px.colors.qualitative.Set1
        )
        
        # 添加参考线（总体正结果率）
        overall_positive_rate = data[target_variable].mean()
        fig.add_hline(
            y=overall_positive_rate, 
            line_dash="dash", 
            line_color="gray",
            annotation_text=f"总体正结果率: {overall_positive_rate:.3f}",
            annotation_position="top right"
        )
        
        # 更新布局
        fig.update_layout(
            font_family="Arial",
            font_size=14,
            height=400,
            margin=dict(l=40, r=40, t=40, b=40)
        )
        
        # 返回HTML
        return fig.to_html(full_html=False)
    
    @staticmethod
    def plot_roc_curves(data, sensitive_attr, y_true, y_pred_proba, title="不同群体的ROC曲线"):
        """
        绘制不同群体的ROC曲线
        
        参数:
        data: 包含敏感属性的数据
        sensitive_attr: 敏感属性列名
        y_true: 真实标签
        y_pred_proba: 预测概率
        title: 图表标题
        
        返回:
        图表对象或HTML代码
        """
        from sklearn.metrics import roc_curve, auc
        import plotly.graph_objects as go
        import numpy as np
        
        fig = go.Figure()
        
        # 添加对角线（随机猜测）
        fig.add_trace(go.Scatter(
            x=np.linspace(0, 1, 100),
            y=np.linspace(0, 1, 100),
            mode='lines',
            line=dict(dash='dash', color='gray'),
            name='随机猜测'
        ))
        
        # 为每个群体绘制ROC曲线
        for attr_value in data[sensitive_attr].unique():
            mask = data[sensitive_attr] == attr_value
            if mask.sum() > 0:
                fpr, tpr, _ = roc_curve(y_true[mask], y_pred_proba[mask])
                roc_auc = auc(fpr, tpr)
                
                fig.add_trace(go.Scatter(
                    x=fpr,
                    y=tpr,
                    mode='lines',
                    name=f'{sensitive_attr}={attr_value} (AUC={roc_auc:.3f})'
                ))
        
        # 更新布局
        fig.update_layout(
            title=title,
            xaxis_title='假正率 (FPR)',
            yaxis_title='真正率 (TPR)',
            font_family="Arial",
            font_size=14,
            height=500,
            width=700,
            legend_title="群体",
            margin=dict(l=40, r=40, t=40, b=40)
        )
        
        return fig.to_html(full_html=False)
    
    @staticmethod
    def plot_confusion_matrices(data, sensitive_attr, y_true, y_pred, title="不同群体的混淆矩阵"):
        """
        绘制不同群体的混淆矩阵
        
        参数:
        data: 包含敏感属性的数据
        sensitive_attr: 敏感属性列名
        y_true: 真实标签
        y_pred: 预测标签
        title: 图表标题
        
        返回:
        图表对象或HTML代码
        """
        from sklearn.metrics import confusion_matrix
        import plotly.subplots as sp
        import plotly.graph_objects as go
        import numpy as np
        
        # 获取所有群体
        attr_values = data[sensitive_attr].unique()
        n_groups = len(attr_values)
        
        # 创建子图
        fig = sp.make_subplots(
            rows=1, 
            cols=n_groups, 
            subplot_titles=[f'{sensitive_attr}={value}' for value in attr_values]
        )
        
        # 为每个群体绘制混淆矩阵
        for i, attr_value in enumerate(attr_values):
            mask = data[sensitive_attr] == attr_value
            if mask.sum() > 0:
                cm = confusion_matrix(y_true[mask], y_pred[mask])
                
                # 计算百分比
                cm_percent = cm / cm.sum() * 100
                
                # 创建文本注释
                text = [[f'{cm[i, j]}<br>({cm_percent[i, j]:.1f}%)' for j in range(cm.shape[1])] 
                        for i in range(cm.shape[0])]
                
                # 添加热力图
                fig.add_trace(
                    go.Heatmap(
                        z=cm,
                        text=text,
                        texttemplate='%{text}',
                        textfont=dict(size=14),
                        colorscale='Blues',
                        showscale=False
                    ),
                    row=1,
                    col=i+1
                )
        
        # 更新布局
        fig.update_layout(
            title=title,
            font_family="Arial",
            font_size=14,
            height=400,
            width=300 * n_groups,
            margin=dict(l=40, r=40, t=40, b=40)
        )
        
        # 更新坐标轴标签
        for i in range(n_groups):
            fig.update_xaxes(title_text='预测标签', row=1, col=i+1)
            fig.update_yaxes(title_text='真实标签', row=1, col=i+1)
        
        return fig.to_html(full_html=False)
    
    @staticmethod
    def plot_fairness_metrics(fairness_results, title="公平性指标分析"):
        """
        绘制公平性指标分析图表
        
        参数:
        fairness_results: 公平性分析结果
        title: 图表标题
        
        返回:
        图表对象或HTML代码
        """
        import plotly.graph_objects as go
        import plotly.subplots as sp
        import numpy as np
        
        # 获取敏感属性列表
        attributes = list(fairness_results['attribute_scores'].keys())
        
        # 为每个指标创建子图
        fig = sp.make_subplots(
            rows=1,
            cols=3,
            subplot_titles=["统计平等", "等机会", "预测均等"],
            shared_yaxes=True
        )
        
        # 为每个属性收集指标数据
        demographic_parity_scores = []
        equal_opportunity_scores = []
        predictive_parity_scores = []
        
        for attr in attributes:
            scores = fairness_results['attribute_scores'][attr]['detailed_scores']
            demographic_parity_scores.append(scores['demographic_parity'])
            equal_opportunity_scores.append(scores['equal_opportunity'])
            predictive_parity_scores.append(scores['predictive_parity'])
        
        # 添加统计平等指标
        fig.add_trace(
            go.Bar(
                x=attributes,
                y=demographic_parity_scores,
                name="统计平等",
                marker_color='royalblue'
            ),
            row=1,
            col=1
        )
        
        # 添加等机会指标
        fig.add_trace(
            go.Bar(
                x=attributes,
                y=equal_opportunity_scores,
                name="等机会",
                marker_color='lightcoral'
            ),
            row=1,
            col=2
        )
        
        # 添加预测均等指标
        fig.add_trace(
            go.Bar(
                x=attributes,
                y=predictive_parity_scores,
                name="预测均等",
                marker_color='lightgreen'
            ),
            row=1,
            col=3
        )
        
        # 添加参考线
        for i in range(3):
            fig.add_hline(y=0.8, line_dash="dash", line_color="green", row=1, col=i+1)
            fig.add_hline(y=0.6, line_dash="dash", line_color="orange", row=1, col=i+1)
            fig.add_hline(y=0.4, line_dash="dash", line_color="red", row=1, col=i+1)
        
        # 更新布局
        fig.update_layout(
            title=title,
            font_family="Arial",
            font_size=14,
            height=400,
            width=1200,
            margin=dict(l=40, r=40, t=40, b=40),
            legend_title="指标"
        )
        
        # 更新y轴范围
        fig.update_yaxes(range=[0, 1], row=1, col=1)
        fig.update_yaxes(range=[0, 1], row=1, col=2)
        fig.update_yaxes(range=[0, 1], row=1, col=3)
        
        return fig.to_html(full_html=False)


class ReportGenerator:
    """报告生成模块，创建全面的审计报告"""
    
    @staticmethod
    def generate_html_report(audit_results, title="AI公平性审计报告"):
        """
        生成HTML格式的审计报告
        
        参数:
        audit_results: 审计结果
        title: 报告标题
        
        返回:
        HTML报告内容
        """
        html_content = f"""
        <!DOCTYPE html>
        <html>
        <head>
            <title>{title}</title>
            <style>
                body {{
                    font-family: Arial, sans-serif;
                    line-height: 1.6;
                    color: #333;
                    max-width: 1200px;
                    margin: 0 auto;
                    padding: 20px;
                }}
                h1, h2, h3, h4 {{
                    color: #2c3e50;
                }}
                .summary-box {{
                    background-color: #f8f9fa;
                    border-left: 4px solid #3498db;
                    padding: 15px;
                    margin-bottom: 20px;
                    border-radius: 5px;
                }}
                .issue-high {{
                    background-color: #fee;
                    border-left: 4px solid #e74c3c;
                    padding: 15px;
                    margin-bottom: 10px;
                    border-radius: 5px;
                }}
                .issue-medium {{
                    background-color: #fff8e6;
                    border-left: 4px solid #f39c12;
                    padding: 15px;
                    margin-bottom: 10px;
                    border-radius: 5px;
                }}
                .recommendation {{
                    background-color: #e8f4f8;
                    border-left: 4px solid #2980b9;
                    padding: 15px;
                    margin-bottom: 10px;
                    border-radius: 5px;
                }}
                table {{
                    width: 100%;
                    border-collapse: collapse;
                    margin: 20px 0;
                }}
                th, td {{
                    border: 1px solid #ddd;
                    padding: 8px 12px;
                    text-align: left;
                }}
                th {{
                    background-color: #f2f2f2;
                }}
                tr:nth-child(even) {{
                    background-color: #f9f9f9;
                }}
                .chart-container {{
                    margin: 20px 0;
                    padding: 15px;
                    border: 1px solid #e0e0e0;
                    border-radius: 5px;
                }}
                .footer {{
                    margin-top: 50px;
                    padding-top: 20px;
                    border-top: 1px solid #e0e0e0;
                    font-size: 0.9em;
                    color: #777;
                    text-align: center;
                }}
            </style>
            <script src="https://cdn.plot.ly/plotly-latest.min.js"></script>
        </head>
        <body>
            <h1>{title}</h1>
            
            <h2>1. 审计摘要</h2>
            <div class="summary-box">
                <h3>总体评估: {audit_results['audit_summary']['overall_assessment']}</h3>
                <table>
                    <tr>
                        <th>问题类型</th>
                        <th>数量</th>
                    </tr>
                    <tr>
                        <td>高严重性问题</td>
                        <td>{audit_results['audit_summary']['high_severity_issues']}</td>
                    </tr>
                    <tr>
                        <td>中等严重性问题</td>
                        <td>{audit_results['audit_summary']['medium_severity_issues']}</td>
                    </tr>
                    <tr>
                        <td>总问题数</td>
                        <td>{audit_results['audit_summary']['total_issues']}</td>
                    </tr>
                </table>
            </div>
        """
        
        # 添加数据审计部分
        if audit_results['data_audit']:
            html_content += """
            <h2>2. 数据审计结果</h2>
            <h3>2.1 数据偏见分析</h3>
            """
            
            # 添加数据偏见问题
            if 'issues' in audit_results['data_audit'] and audit_results['data_audit']['issues']:
                for issue in audit_results['data_audit']['issues']:
                    issue_class = 'issue-high' if issue['severity'] == 'high' else 'issue-medium'
                    html_content += f"""
                    <div class="{issue_class}">
                        <strong>问题:</strong> {issue['issue']} ({issue['attribute']})<br>
                        <strong>严重性:</strong> {issue['severity']}<br>
                        <strong>差异程度:</strong> {issue['max_difference']:.3f}
                    </div>
                    """
            else:
                html_content += "<p>未发现严重的数据偏见问题。</p>"
        
        # 添加模型审计部分
        if audit_results['model_audit']:
            html_content += """
            <h2>3. 模型审计结果</h2>
            """
            
            # 添加模型偏见问题
            if 'issues' in audit_results['model_audit'] and audit_results['model_audit']['issues']:
                for issue in audit_results['model_audit']['issues']:
                    issue_class = 'issue-high' if issue['severity'] == 'high' else 'issue-medium'
                    html_content += f"""
                    <div class="{issue_class}">
                        <strong>问题:</strong> {issue['issue']} ({issue['attribute']})<br>
                        <strong>严重性:</strong> {issue['severity']}<br>
                        <strong>公平性得分:</strong> {issue['score']:.3f}
                    </div>
                    """
            else:
                html_content += "<p>未发现严重的模型偏见问题。</p>"
            
            # 添加公平性得分详情
            if 'fairness_scores' in audit_results['model_audit']:
                html_content += """
                <h3>3.1 公平性得分详情</h3>
                <table>
                    <tr>
                        <th>敏感属性</th>
                        <th>统计平等</th>
                        <th>等机会</th>
                        <th>预测均等</th>
                        <th>加权得分</th>
                        <th>等级</th>
                    </tr>
                """
                
                for attr, scores in audit_results['model_audit']['fairness_scores']['attribute_scores'].items():
                    html_content += f"""
                    <tr>
                        <td>{attr}</td>
                        <td>{scores['detailed_scores']['demographic_parity']:.3f}</td>
                        <td>{scores['detailed_scores']['equal_opportunity']:.3f}</td>
                        <td>{scores['detailed_scores']['predictive_parity']:.3f}</td>
                        <td>{scores['weighted_score']:.3f}</td>
                        <td>{scores['level']}</td>
                    </tr>
                    """
                
                html_content += """
                </table>
                <p><strong>总体公平性得分:</strong> {audit_results['model_audit']['fairness_scores']['overall_score']:.3f}</p>
                <p><strong>总体公平性等级:</strong> {audit_results['model_audit']['fairness_scores']['overall_level']}</p>
                """
            
            # 添加可视化图表
            if 'visualizations' in audit_results:
                html_content += """
                <h3>3.2 可视化分析</h3>
                <div class="chart-container">
                    {audit_results['visualizations']}
                </div>
                """
        
        # 添加改进建议部分
        if audit_results['mitigation_recommendations']:
            html_content += """
            <h2>4. 改进建议</h2>
            """
            
            for rec in audit_results['mitigation_recommendations']:
                html_content += f"""
                <div class="recommendation">
                    <strong>类型:</strong> {rec['type']}<br>
                    {rec.get('strategy', '') and f"<strong>策略:</strong> {rec['strategy']}<br>"}
                    <strong>描述:</strong> {rec.get('description', rec.get('recommendation', ''))}<br>
                    <strong>优先级:</strong> {rec['priority']}<br>
                    <strong>预计工作量:</strong> {rec['estimated_effort']}
                </div>
                """
        
        # 添加页脚
        html_content += """
            <div class="footer">
                <p>报告生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
            </div>
        </body>
        </html>
        """
        
        # 导入datetime模块用于时间戳
        from datetime import datetime
        
        return html_content
    
    @staticmethod
    def save_report(report_content, output_file):
        """
        保存报告到文件
        
        参数:
        report_content: 报告内容
        output_file: 输出文件路径
        """
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(report_content)
        print(f"报告已保存到: {output_file}")

pip install ai-fairness-audit

# 敏感属性配置
sensitive_attributes:
  - name: "gender"
    description: "性别"
    values: ["male", "female", "other"]
  - name: "race"
    description: "种族"
    values: ["white", "black", "asian", "hispanic", "other"]
  - name: "age"
    description: "年龄"
    bins: [(0, 30), (31, 50), (51, 100)]

# 公平性指标配置
fairness_metrics:
  - name: "demographic_parity"
    weight: 0.3
    threshold: 1.2
  - name: "equal_opportunity"
    weight: 0.4
    threshold: 0.05
  - name: "predictive_parity"
    weight: 0.3
    threshold: 0.05

# 审计报告配置
report:
  format: "html"
  output_path: "./reports"
  include_visualizations: true
  include_recommendations: true

from ai_fairness_audit import AuditFramework, DataPreprocessor

# 加载和预处理数据
preprocessor = DataPreprocessor()
data = preprocessor.load_data('loan_data.csv')
cleaned_data = preprocessor.clean_data(data, sensitive_attributes=['gender', 'race', 'age'])

# 初始化审计框架
framework = AuditFramework(config_path='config.yaml')

# 运行数据审计
data_audit_results = framework.run_audit(
    data=cleaned_data,
    target_variable='default',
    audit_type='data_only'
)

# 生成报告
from ai_fairness_audit import ReportGenerator
html_report = ReportGenerator.generate_html_report(data_audit_results, "信贷数据公平性审计报告")
ReportGenerator.save_report(html_report, "./reports/data_audit_report.html")

from sklearn.ensemble import RandomForestClassifier
from ai_fairness_audit import AuditFramework, DataPreprocessor, ModelAdapter

# 准备数据
preprocessor = DataPreprocessor()
data = preprocessor.load_data('loan_data.csv')
cleaned_data = preprocessor.clean_data(data, sensitive_attributes=['gender', 'race', 'age'])
encoded_data = preprocessor.encode_categorical(cleaned_data)
X_train, X_test, y_train, y_test = preprocessor.split_data(encoded_data, 'default')

# 训练模型
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# 初始化审计框架
framework = AuditFramework(config_path='config.yaml')

# 运行综合审计
audit_results = framework.run_audit(
    data=encoded_data,
    model=model,
    target_variable='default',
    audit_type='comprehensive'
)

# 生成报告
from ai_fairness_audit import ReportGenerator
html_report = ReportGenerator.generate_html_report(audit_results, "信贷模型公平性审计报告")
ReportGenerator.save_report(html_report, "./reports/model_audit_report.html")

from ai_fairness_audit import BiasMitigationStrategies

# 加载缓解策略
mitigation = BiasMitigationStrategies()

# 应用数据重加权
data, weights = mitigation.reweighting(
    cleaned_data, 
    sensitive_attr='gender', 
    target='default',
    method='demographic_parity'
)

# 使用加权数据重新训练模型
model_weighted = RandomForestClassifier(n_estimators=100, random_state=42)
model_weighted.fit(X_train, y_train, sample_weight=weights.iloc[X_train.index])

# 审计改进后的模型
improved_results = framework.run_audit(
    data=encoded_data,
    model=model_weighted,
    target_variable='default',
    audit_type='model_only'
)

# 比较改进前后的公平性
print("原始模型公平性得分:", audit_results['model_audit']['fairness_scores']['overall_score'])
print("改进后模型公平性得分:", improved_results['model_audit']['fairness_scores']['overall_score'])

## 5. 前沿技术与未来发展

### 5.1 新兴技术在偏见审计中的应用

2025年，多种前沿技术正在革新AI公平性审计的方式和效果。这些技术不仅提高了审计的准确性和效率，还拓展了审计的适用范围。

#### 5.1.1 可解释AI技术的应用

可解释AI(XAI)技术在偏见审计中的应用越来越广泛，帮助审计人员理解模型决策背后的原因，特别是针对复杂的深度学习模型。

**注意力机制可视化**：通过可视化模型的注意力权重，识别模型在决策过程中过于关注敏感属性的情况。

```python
def analyze_attention_bias(attention_weights, sensitive_attributes, input_tokens):
    """
    分析模型注意力机制中的偏见
    
    参数:
    attention_weights: 模型的注意力权重矩阵
    sensitive_attributes: 敏感属性相关的token列表
    input_tokens: 输入token列表
    
    返回:
    偏见分析结果
    """
    import numpy as np
    
    # 识别敏感属性token的位置
    sensitive_positions = [i for i, token in enumerate(input_tokens) 
                          if any(attr in token.lower() for attr in sensitive_attributes)]
    
    # 计算敏感属性token的平均注意力权重
    if sensitive_positions:
        sensitive_attention = np.mean(attention_weights[:, sensitive_positions], axis=1)
        non_sensitive_attention = np.mean(attention_weights[:, [i for i in range(len(input_tokens)) 
                                                                if i not in sensitive_positions]], axis=1)
        
        # 计算注意力偏差
        attention_bias = np.mean(sensitive_attention) / np.mean(non_sensitive_attention)
        
        return {
            'has_sensitive_tokens': True,
            'attention_bias': attention_bias,
            'sensitive_attention_mean': np.mean(sensitive_attention),
            'non_sensitive_attention_mean': np.mean(non_sensitive_attention)
        }
    else:
        return {
            'has_sensitive_tokens': False,
            'message': '输入中未发现敏感属性相关token'
        }

def causal_bias_analysis(model, data, sensitive_attributes, target_variable):
    """
    使用因果推断分析模型中的偏见
    
    参数:
    model: 要分析的模型
    data: 数据集
    sensitive_attributes: 敏感属性列表
    target_variable: 目标变量
    
    返回:
    因果偏见分析结果
    """
    import dowhy
    import dowhy.datasets
    
    results = {}
    
    for sensitive_attr in sensitive_attributes:
        # 创建因果图
        causal_graph = f"{sensitive_attr}->{target_variable}"
        
        # 构建因果模型
        model_causal = dowhy.CausalModel(
            data=data,
            treatment=sensitive_attr,
            outcome=target_variable,
            graph=causal_graph
        )
        
        # 识别因果效应
        identified_estimand = model_causal.identify_effect()
        
        # 估计因果效应
        estimate = model_causal.estimate_effect(
            identified_estimand,
            method_name="backdoor.propensity_score_matching"
        )
        
        # 测试敏感性
        refute_results = model_causal.refute_estimate(
            identified_estimand, 
            estimate, 
            method_name="random_common_cause"
        )
        
        results[sensitive_attr] = {
            'causal_effect': estimate.value,
            'p_value': estimate.test_stat_significance()['p_value'],
            'refute_result': refute_results.refutation_result
        }
    
    return results

def differential_privacy_audit(model, data, sensitive_attributes, target_variable, epsilon=1.0):
    """
    使用差分隐私技术进行偏见审计
    
    参数:
    model: 要审计的模型
    data: 数据集
    sensitive_attributes: 敏感属性列表
    target_variable: 目标变量
    epsilon: 差分隐私预算
    
    返回:
    差分隐私审计结果
    """
    import numpy as np
    from diffprivlib.models import RandomForestClassifier as DPRandomForest
    
    results = {}
    
    for sensitive_attr in sensitive_attributes:
        # 按照敏感属性分组
        groups = data[sensitive_attr].unique()
        group_results = {}
        
        for group in groups:
            # 提取该群体的数据
            group_data = data[data[sensitive_attr] == group]
            
            # 应用差分隐私保护
            dp_model = DPRandomForest(epsilon=epsilon/len(groups))
            
            # 准备特征和标签
            X = group_data.drop([sensitive_attr, target_variable], axis=1)
            y = group_data[target_variable]
            
            # 训练并评估
            dp_model.fit(X, y)
            y_pred = dp_model.predict(X)
            
            # 计算准确率（添加拉普拉斯噪声）
            accuracy = np.mean(y_pred == y)
            # 添加拉普拉斯噪声
            noise = np.random.laplace(0, 1/epsilon)
            private_accuracy = max(0, min(1, accuracy + noise))
            
            group_results[group] = {'private_accuracy': private_accuracy}
        
        # 计算群体间差异
        accuracies = [v['private_accuracy'] for v in group_results.values()]
        results[sensitive_attr] = {
            'group_results': group_results,
            'max_difference': max(accuracies) - min(accuracies) if accuracies else 0
        }
    
    return results

def federated_bias_audit(clients, global_model, sensitive_attributes, target_variable, audit_config):
    """
    在联邦学习环境中进行偏见审计
    
    参数:
    clients: 客户端列表
    global_model: 全局模型
    sensitive_attributes: 敏感属性列表
    target_variable: 目标变量
    audit_config: 审计配置
    
    返回:
    联邦审计结果
    """
    import numpy as np
    
    # 初始化结果存储
    results = {
        'client_results': {},
        'global_results': {}
    }
    
    # 本地审计
    for client_id, client in clients.items():
        # 获取客户端模型副本
        local_model = client.get_model()
        
        # 获取客户端数据（使用本地训练的数据）
        local_data = client.get_data_sample(limit=audit_config.get('sample_size', 1000))
        
        # 本地进行偏见审计
        client_bias_results = {
            sensitive_attr: {
                'demographic_parity': 0,
                'equal_opportunity': 0
            } for sensitive_attr in sensitive_attributes
        }
        
        # 计算每个敏感属性的公平性指标
        for sensitive_attr in sensitive_attributes:
            # 本地计算（这里简化处理，实际应该调用公平性计算函数）
            # 由于是联邦学习，客户端只计算指标，不共享原始数据
            group_stats = {}
            groups = local_data[sensitive_attr].unique()
            
            for group in groups:
                group_data = local_data[local_data[sensitive_attr] == group]
                X = group_data.drop([sensitive_attr, target_variable], axis=1)
                y = group_data[target_variable]
                
                y_pred = local_model.predict(X)
                y_pred_proba = local_model.predict_proba(X)[:, 1] if hasattr(local_model, 'predict_proba') else y_pred
                
                # 计算群体统计信息
                group_stats[group] = {
                    'positive_rate': np.mean(y_pred),
                    'tpr': np.sum((y_pred == 1) & (y == 1)) / np.sum(y == 1) if np.sum(y == 1) > 0 else 0
                }
            
            # 计算统计平等
            positive_rates = [stats['positive_rate'] for stats in group_stats.values()]
            demographic_parity = 1.0 - (max(positive_rates) - min(positive_rates)) if positive_rates else 1.0
            
            # 计算等机会
            tprs = [stats['tpr'] for stats in group_stats.values() if stats['tpr'] > 0]
            equal_opportunity = 1.0 - (max(tprs) - min(tprs)) if tprs else 1.0
            
            client_bias_results[sensitive_attr] = {
                'demographic_parity': demographic_parity,
                'equal_opportunity': equal_opportunity,
                'group_count': len(groups)
            }
        
        # 添加到结果中
        results['client_results'][client_id] = client_bias_results
    
    # 聚合结果
    for sensitive_attr in sensitive_attributes:
        # 收集所有客户端的指标
        client_dp_scores = [r[sensitive_attr]['demographic_parity'] for r in results['client_results'].values()]
        client_eo_scores = [r[sensitive_attr]['equal_opportunity'] for r in results['client_results'].values()]
        
        # 加权平均（基于群体数量）
        group_counts = [r[sensitive_attr]['group_count'] for r in results['client_results'].values()]
        total_groups = sum(group_counts)
        
        if total_groups > 0:
            weighted_dp = sum(s * c for s, c in zip(client_dp_scores, group_counts)) / total_groups
            weighted_eo = sum(s * c for s, c in zip(client_eo_scores, group_counts)) / total_groups
        else:
            weighted_dp = np.mean(client_dp_scores) if client_dp_scores else 0
            weighted_eo = np.mean(client_eo_scores) if client_eo_scores else 0
        
        results['global_results'][sensitive_attr] = {
            'demographic_parity': weighted_dp,
            'equal_opportunity': weighted_eo,
            'weighted_score': 0.5 * weighted_dp + 0.5 * weighted_eo
        }
    
    return results

def multimodal_bias_audit(text_model, image_model, data, sensitive_attributes):
    """
    多模态模型的偏见审计
    
    参数:
    text_model: 文本模型
    image_model: 图像模型
    data: 多模态数据集
    sensitive_attributes: 敏感属性列表
    
    返回:
    多模态偏见分析结果
    """
    import numpy as np
    from PIL import Image
    
    results = {
        'text_bias': {},
        'image_bias': {},
        'cross_modal_bias': {}
    }
    
    # 文本偏见分析
    for sensitive_attr in sensitive_attributes:
        # 按敏感属性分组
        groups = data[sensitive_attr].unique()
        group_scores = []
        
        for group in groups:
            group_data = data[data[sensitive_attr] == group]
            # 提取文本特征
            text_features = group_data['text'].tolist()
            # 预测并计算得分
            text_predictions = text_model.predict(text_features)
            # 简化为计算正向情感比例（根据应用场景调整）
            positive_rate = np.mean(text_predictions)
            group_scores.append({
                'group': group,
                'positive_rate': positive_rate,
                'count': len(group_data)
            })
        
        # 计算差异
        rates = [s['positive_rate'] for s in group_scores]
        max_diff = max(rates) - min(rates) if rates else 0
        
        results['text_bias'][sensitive_attr] = {
            'group_scores': group_scores,
            'max_difference': max_diff
        }
    
    # 图像偏见分析（类似文本分析）
    for sensitive_attr in sensitive_attributes:
        groups = data[sensitive_attr].unique()
        group_scores = []
        
        for group in groups:
            group_data = data[data[sensitive_attr] == group]
            # 提取图像并预处理
            images = []
            for img_path in group_data['image_path']:
                try:
                    img = Image.open(img_path).resize((224, 224))
                    img_array = np.array(img) / 255.0
                    images.append(img_array)
                except:
                    continue
            
            if images:
                # 预测
                image_predictions = image_model.predict(np.array(images))
                positive_rate = np.mean(image_predictions)
                group_scores.append({
                    'group': group,
                    'positive_rate': positive_rate,
                    'count': len(images)
                })
        
        # 计算差异
        rates = [s['positive_rate'] for s in group_scores]
        max_diff = max(rates) - min(rates) if rates else 0
        
        results['image_bias'][sensitive_attr] = {
            'group_scores': group_scores,
            'max_difference': max_diff
        }
    
    # 跨模态偏见分析
    for sensitive_attr in sensitive_attributes:
        groups = data[sensitive_attr].unique()
        cross_scores = []
        
        for group in groups:
            group_data = data[data[sensitive_attr] == group]
            modal_differences = []
            
            for _, row in group_data.iterrows():
                try:
                    # 文本预测
                    text_pred = text_model.predict([row['text']])[0]
                    
                    # 图像预测
                    img = Image.open(row['image_path']).resize((224, 224))
                    img_array = np.array(img) / 255.0
                    img_pred = image_model.predict(np.array([img_array]))[0]
                    
                    # 计算差异
                    modal_diff = abs(text_pred - img_pred)
                    modal_differences.append(modal_diff)
                except:
                    continue
            
            if modal_differences:
                avg_diff = np.mean(modal_differences)
                cross_scores.append({
                    'group': group,
                    'avg_modal_difference': avg_diff,
                    'count': len(modal_differences)
                })
        
        results['cross_modal_bias'][sensitive_attr] = {
            'group_scores': cross_scores,
            'interpretation': '跨模态差异小表示模态间一致性高，差异大可能暗示某种形式的偏见'
        }
    
    return results

def compliance_checker(audit_results, industry_standards):
    """
    根据行业标准检查审计结果的合规性
    
    参数:
    audit_results: 偏见审计结果
    industry_standards: 行业标准要求
    
    返回:
    合规检查报告
    """
    compliance_report = {
        'overall_compliance': True,
        'detailed_checks': {},
        'recommendations': []
    }
    
    # 检查每个敏感属性的合规性
    for sensitive_attr, standards in industry_standards.items():
        attr_checks = {}
        
        # 获取该属性的审计结果
        if sensitive_attr in audit_results['model_audit']['fairness_scores']['attribute_scores']:
            scores = audit_results['model_audit']['fairness_scores']['attribute_scores'][sensitive_attr]
            
            # 检查每个指标
            for metric, standard_threshold in standards.items():
                if metric in scores['detailed_scores']:
                    actual_score = scores['detailed_scores'][metric]
                    is_compliant = actual_score >= standard_threshold
                    
                    attr_checks[metric] = {
                        'actual_score': actual_score,
                        'standard_threshold': standard_threshold,
                        'is_compliant': is_compliant
                    }
                    
                    if not is_compliant:
                        compliance_report['overall_compliance'] = False
                        compliance_report['recommendations'].append({
                            'attribute': sensitive_attr,
                            'metric': metric,
                            'issue': f'指标得分 {actual_score:.3f} 低于标准阈值 {standard_threshold}',
                            'suggestion': f'考虑使用偏见缓解技术改进{metric}指标'
                        })
                else:
                    attr_checks[metric] = {
                        'status': 'missing',
                        'message': f'未找到{metric}指标的审计结果',
                        'is_compliant': False
                    }
                    compliance_report['overall_compliance'] = False
        else:
            attr_checks['general'] = {
                'status': 'missing',
                'message': f'未找到{sensitive_attr}的审计结果',
                'is_compliant': False
            }
            compliance_report['overall_compliance'] = False
            compliance_report['recommendations'].append({
                'attribute': sensitive_attr,
                'issue': f'未审计{sensitive_attr}的公平性',
                'suggestion': '请使用适当的敏感属性配置重新运行审计'
            })
        
        compliance_report['detailed_checks'][sensitive_attr] = attr_checks
    
    # 检查总体合规性
    if compliance_report['overall_compliance']:
        compliance_report['compliance_status'] = '通过'
        compliance_report['certification_ready'] = True
    else:
        compliance_report['compliance_status'] = '需要改进'
        compliance_report['certification_ready'] = False
    
    return compliance_report

class RealTimeFairnessMonitor:
    """
    实时公平性监控系统
    """
    
    def __init__(self, model, sensitive_attributes, threshold_config):
        """
        初始化实时监控系统
        
        参数:
        model: 要监控的模型
        sensitive_attributes: 敏感属性列表
        threshold_config: 阈值配置
        """
        import datetime
        
        self.model = model
        self.sensitive_attributes = sensitive_attributes
        self.threshold_config = threshold_config
        
        # 初始化监控数据存储
        self.monitoring_data = {}
        for attr in sensitive_attributes:
            self.monitoring_data[attr] = {
                'group_stats': {},
                'alert_history': []
            }
        
        # 初始化滑动窗口数据
        self.window_size = 1000  # 监控窗口大小
        self.recent_predictions = []
    
    def process_prediction(self, prediction_data):
        """
        处理单个预测结果并更新监控数据
        
        参数:
        prediction_data: 包含预测结果和敏感属性信息的数据
        
        返回:
        是否触发警报
        """
        # 保存预测结果
        self.recent_predictions.append(prediction_data)
        
        # 维护滑动窗口
        if len(self.recent_predictions) > self.window_size:
            self.recent_predictions.pop(0)
        
        # 更新监控统计
        alerts = []
        
        for attr in self.sensitive_attributes:
            if attr in prediction_data:
                group_value = prediction_data[attr]
                
                # 更新群体统计
                if group_value not in self.monitoring_data[attr]['group_stats']:
                    self.monitoring_data[attr]['group_stats'][group_value] = {
                        'total_count': 0,
                        'positive_count': 0,
                        'prediction_history': []
                    }
                
                stats = self.monitoring_data[attr]['group_stats'][group_value]
                stats['total_count'] += 1
                stats['positive_count'] += 1 if prediction_data['prediction'] == 1 else 0
                stats['prediction_history'].append(prediction_data['prediction'])
                
                # 限制历史记录长度
                if len(stats['prediction_history']) > self.window_size:
                    stats['prediction_history'].pop(0)
        
        # 检查是否需要计算指标并触发警报
        if len(self.recent_predictions) >= self.window_size / 2:  # 当窗口至少填满一半时开始检查
            for attr in self.sensitive_attributes:
                if len(self.monitoring_data[attr]['group_stats']) >= 2:  # 至少需要两个群体
                    alert = self._check_fairness_metrics(attr)
                    if alert:
                        alerts.append(alert)
        
        # 记录警报
        for alert in alerts:
            self.monitoring_data[alert['attribute']]['alert_history'].append({
                'timestamp': prediction_data['timestamp'],
                'alert_type': alert['type'],
                'severity': alert['severity'],
                'details': alert['details']
            })
        
        return alerts
    
    def _check_fairness_metrics(self, sensitive_attr):
        """
        检查特定敏感属性的公平性指标
        
        参数:
        sensitive_attr: 敏感属性
        
        返回:
        如果触发警报，返回警报信息；否则返回None
        """
        import numpy as np
        
        group_stats = self.monitoring_data[sensitive_attr]['group_stats']
        
        # 计算统计平等
        positive_rates = []
        group_sizes = []
        
        for group, stats in group_stats.items():
            if stats['total_count'] > 0:  # 确保有足够的数据
                positive_rate = stats['positive_count'] / stats['total_count']
                positive_rates.append(positive_rate)
                group_sizes.append(stats['total_count'])
        
        if len(positive_rates) >= 2:
            # 计算群体间的最大差异
            max_diff = max(positive_rates) - min(positive_rates)
            
            # 检查是否超过阈值
            if max_diff > self.threshold_config.get('demographic_parity_threshold', 0.1):
                return {
                    'attribute': sensitive_attr,
                    'type': 'demographic_parity_violation',
                    'severity': 'high' if max_diff > 0.2 else 'medium',
                    'details': {
                        'max_difference': max_diff,
                        'threshold': self.threshold_config.get('demographic_parity_threshold', 0.1),
                        'group_count': len(positive_rates)
                    }
                }
        
        # 其他指标检查可以在这里添加
        
        return None
    
    def generate_monitoring_report(self):
        """
        生成监控报告
        
        返回:
        监控报告
        """
        import datetime
        
        report = {
            'timestamp': datetime.datetime.now().isoformat(),
            'window_size': len(self.recent_predictions),
            'attribute_summaries': {},
            'total_alerts': 0
        }
        
        for attr in self.sensitive_attributes:
            # 计算当前统计信息
            attr_summary = {
                'group_count': len(self.monitoring_data[attr]['group_stats']),
                'alert_count': len(self.monitoring_data[attr]['alert_history']),
                'group_statistics': {}
            }
            
            # 收集群体统计
            for group, stats in self.monitoring_data[attr]['group_stats'].items():
                if stats['total_count'] > 0:
                    attr_summary['group_statistics'][group] = {
                        'total_predictions': stats['total_count'],
                        'positive_rate': stats['positive_count'] / stats['total_count'],
                        'alert_count': sum(1 for a in self.monitoring_data[attr]['alert_history'] 
                                         if a['attribute'] == attr)
                    }
            
            report['attribute_summaries'][attr] = attr_summary
            report['total_alerts'] += attr_summary['alert_count']
        
        return report

class AutoMitigationSystem:
    """
    自动化偏见缓解系统
    """
    
    def __init__(self, model, initial_data, sensitive_attributes, target_variable):
        """
        初始化自动化缓解系统
        
        参数:
        model: 要优化的模型
        initial_data: 初始训练数据
        sensitive_attributes: 敏感属性列表
        target_variable: 目标变量
        """
        self.model = model
        self.initial_data = initial_data
        self.sensitive_attributes = sensitive_attributes
        self.target_variable = target_variable
        
        # 初始化缓解策略库
        self.mitigation_strategies = {
            'reweighting': self._reweighting,
            'resampling': self._resampling,
            'postprocessing': self._postprocessing,
            'adversarial_debiasing': self._adversarial_debiasing
        }
        
        # 记录策略历史
        self.strategy_history = []
    
    def auto_mitigate(self, audit_results, performance_threshold=0.8):
        """
        根据审计结果自动选择并应用缓解策略
        
        参数:
        audit_results: 偏见审计结果
        performance_threshold: 性能下降阈值
        
        返回:
        优化后的模型和缓解报告
        """
        import copy
        
        mitigation_report = {
            'original_fairness': {},
            'applied_strategies': [],
            'final_fairness': {},
            'performance_impact': {}
        }
        
        # 记录原始公平性
        for attr, scores in audit_results['model_audit']['fairness_scores']['attribute_scores'].items():
            mitigation_report['original_fairness'][attr] = scores['weighted_score']
        mitigation_report['original_fairness']['overall'] = audit_results['model_audit']['fairness_scores']['overall_score']
        
        # 计算原始性能
        X_test = self.initial_data.drop([self.target_variable] + self.sensitive_attributes, axis=1)
        y_test = self.initial_data[self.target_variable]
        original_performance = self._evaluate_performance(X_test, y_test)
        
        # 选择缓解策略
        strategies_to_apply = self._select_strategies(audit_results)
        
        # 应用策略
        for strategy_name, strategy_config in strategies_to_apply:
            if strategy_name in self.mitigation_strategies:
                try:
                    # 应用策略
                    new_model, strategy_report = self.mitigation_strategies[strategy_name](
                        self.model, 
                        self.initial_data,
                        self.sensitive_attributes,
                        self.target_variable,
                        strategy_config
                    )
                    
                    # 评估应用后的性能
                    new_performance = self._evaluate_performance(X_test, y_test)
                    performance_change = new_performance - original_performance
                    
                    # 检查性能是否下降过多
                    if performance_change >= -performance_threshold:
                        # 性能下降在可接受范围内，采用新模型
                        self.model = new_model
                        mitigation_report['applied_strategies'].append({
                            'name': strategy_name,
                            'config': strategy_config,
                            'performance_impact': performance_change,
                            'details': strategy_report
                        })
                        
                        # 更新性能基准
                        original_performance = new_performance
                    else:
                        # 性能下降过多，跳过此策略
                        mitigation_report['applied_strategies'].append({
                            'name': strategy_name,
                            'status': 'skipped',
                            'reason': '性能下降过多',
                            'performance_impact': performance_change
                        })
                except Exception as e:
                    mitigation_report['applied_strategies'].append({
                        'name': strategy_name,
                        'status': 'failed',
                        'error': str(e)
                    })
        
        # 记录最终公平性
        from ai_fairness_audit import AuditFramework
        framework = AuditFramework()
        final_audit = framework.run_audit(
            data=self.initial_data,
            model=self.model,
            target_variable=self.target_variable,
            audit_type='model_only'
        )
        
        for attr, scores in final_audit['model_audit']['fairness_scores']['attribute_scores'].items():
            mitigation_report['final_fairness'][attr] = scores['weighted_score']
        mitigation_report['final_fairness']['overall'] = final_audit['model_audit']['fairness_scores']['overall_score']
        
        # 记录性能影响
        mitigation_report['performance_impact'] = {
            'original': original_performance,
            'final': self._evaluate_performance(X_test, y_test),
            'change': self._evaluate_performance(X_test, y_test) - original_performance
        }
        
        return self.model, mitigation_report
    
    def _select_strategies(self, audit_results):
        """
        根据审计结果选择合适的缓解策略
        
        参数:
        audit_results: 偏见审计结果
        
        返回:
        要应用的策略列表
        """
        strategies = []
        
        # 分析问题类型
        has_high_severity_issues = audit_results['audit_summary']['high_severity_issues'] > 0
        has_data_issues = 'issues' in audit_results['data_audit'] and audit_results['data_audit']['issues']
        has_model_issues = 'issues' in audit_results['model_audit'] and audit_results['model_audit']['issues']
        
        # 根据问题选择策略
        if has_high_severity_issues:
            # 高严重性问题，考虑更激进的策略
            if has_data_issues:
                strategies.append(('resampling', {'method': 'oversampling', 'minority_ratio': 0.8}))
            
            if has_model_issues:
                strategies.append(('adversarial_debiasing', {'epochs': 50, 'lambda': 0.5}))
        else:
            # 中等严重性问题，考虑温和的策略
            if has_data_issues:
                strategies.append(('reweighting', {'method': 'demographic_parity'}))
            
            if has_model_issues:
                strategies.append(('postprocessing', {'method': 'threshold_calibration'}))
        
        # 总是添加后处理作为备选
        if not any(s[0] == 'postprocessing' for s in strategies):
            strategies.append(('postprocessing', {'method': 'threshold_calibration'}))
        
        return strategies
    
    def _evaluate_performance(self, X, y):
        """
        评估模型性能
        
        参数:
        X: 特征
        y: 标签
        
        返回:
        性能得分
        """
        from sklearn.metrics import accuracy_score, f1_score
        
        y_pred = self.model.predict(X)
        accuracy = accuracy_score(y, y_pred)
        f1 = f1_score(y, y_pred, average='weighted')
        
        # 返回综合性能得分
        return 0.5 * accuracy + 0.5 * f1
    
    def _reweighting(self, model, data, sensitive_attributes, target_variable, config):
        """
        数据重加权缓解策略
        """
        # 实现重加权逻辑
        # ...
        return model, {'method': config['method'], 'weights_applied': True}
    
    def _resampling(self, model, data, sensitive_attributes, target_variable, config):
        """
        数据重采样缓解策略
        """
        # 实现重采样逻辑
        # ...
        return model, {'method': config['method'], 'resampled': True}
    
    def _postprocessing(self, model, data, sensitive_attributes, target_variable, config):
        """
        后处理缓解策略
        """
        # 实现后处理逻辑
        # ...
        return model, {'method': config['method'], 'calibrated': True}
    
    def _adversarial_debiasing(self, model, data, sensitive_attributes, target_variable, config):
        """
        对抗性去偏缓解策略
        """
        # 实现对抗性去偏逻辑
        # ...
        return model, {'epochs': config['epochs'], 'lambda': config['lambda']}

class HumanInTheLoopAudit:
    """
    人机协作审计系统
    """
    
    def __init__(self, audit_system, expert_knowledge_base=None):
        """
        初始化人机协作审计系统
        
        参数:
        audit_system: 基础审计系统
        expert_knowledge_base: 专家知识库
        """
        import copy
        
        self.audit_system = audit_system
        self.expert_knowledge = expert_knowledge_base or {}
        self.expert_feedback_history = []
    
    def run_collaborative_audit(self, data, model=None, target_variable=None, expert_session=None):
        """
        运行人机协作审计
        
        参数:
        data: 要审计的数据
        model: 要审计的模型
        target_variable: 目标变量
        expert_session: 专家会话（可选，用于实时交互）
        
        返回:
        协作审计结果
        """
        import copy
        import datetime
        
        # 运行基础审计
        base_audit_results = self.audit_system.run_audit(
            data=data,
            model=model,
            target_variable=target_variable
        )
        
        # 整合专家知识
        enhanced_results = self._integrate_expert_knowledge(base_audit_results)
        
        # 如果有专家会话，进行实时交互
        if expert_session:
            # 识别需要专家确认的问题
            issues_for_review = self._identify_issues_for_expert_review(enhanced_results)
            
            # 收集专家反馈
            if issues_for_review:
                expert_feedback = expert_session.request_review(issues_for_review)
                enhanced_results = self._apply_expert_feedback(enhanced_results, expert_feedback)
                
                # 记录反馈
                self.expert_feedback_history.append({
                    'timestamp': datetime.datetime.now(),
                    'feedback': expert_feedback,
                    'audit_context': {
                        'data_sample_size': len(data),
                        'model_type': str(type(model)) if model else 'None',
                        'target_variable': target_variable
                    }
                })
        
        return enhanced_results
    
    def _integrate_expert_knowledge(self, audit_results):
        """
        整合专家知识到审计结果中
        """
        import copy
        
        # 复制审计结果
        enhanced_results = copy.deepcopy(audit_results)
        
        # 应用专家规则
        for attr, issues in enhanced_results['model_audit'].get('issues', {}).items():
            if attr in self.expert_knowledge.get('attribute_rules', {}):
                for rule in self.expert_knowledge['attribute_rules'][attr]:
                    # 应用规则到问题评估
                    if self._evaluate_rule(rule, issues):
                        # 更新问题严重性或添加专家注释
                        if 'expert_note' in rule:
                            for issue in issues:
                                issue['expert_note'] = rule['expert_note']
                        
                        if 'severity_adjustment' in rule:
                            for issue in issues:
                                current_severity = issue.get('severity', 'medium')
                                adjustment = rule['severity_adjustment']
                                if adjustment == 'increase' and current_severity == 'medium':
                                    issue['severity'] = 'high'
                                elif adjustment == 'decrease' and current_severity == 'high':
                                    issue['severity'] = 'medium'
        
        # 应用领域特定的公平性定义
        if 'domain_specific_metrics' in self.expert_knowledge:
            for metric_name, metric_def in self.expert_knowledge['domain_specific_metrics'].items():
                # 计算特定领域指标
                enhanced_results['model_audit']['domain_specific_metrics'] = \
                    enhanced_results['model_audit'].get('domain_specific_metrics', {})
                enhanced_results['model_audit']['domain_specific_metrics'][metric_name] = \
                    self._calculate_domain_metric(metric_def, audit_results)
        
        return enhanced_results
    
    def collect_expert_feedback(self, audit_results, feedback_items):
        """
        收集和整合专家反馈
        
        参数:
        audit_results: 审计结果
        feedback_items: 专家反馈项目
        
        返回:
        更新后的审计结果
        """
        # 应用反馈
        updated_results = self._apply_expert_feedback(audit_results, feedback_items)
        
        # 学习专家偏好，更新知识库
        self._learn_from_feedback(feedback_items)
        
        return updated_results
    
    def _identify_issues_for_expert_review(self, audit_results):
        """
        识别需要专家审查的问题
        """
        issues_for_review = []
        
        # 选择边界情况或高影响问题
        for attr, issues in audit_results['model_audit'].get('issues', {}).items():
            for issue in issues:
                # 严重性为high的问题始终需要审查
                if issue.get('severity') == 'high':
                    issues_for_review.append({
                        'attribute': attr,
                        'issue': issue,
                        'review_reason': '高严重性问题'
                    })
                # 边界情况也需要审查
                elif 'score' in issue and 0.4 <= issue['score'] <= 0.6:
                    issues_for_review.append({
                        'attribute': attr,
                        'issue': issue,
                        'review_reason': '边界分数，需要专家判断'
                    })
        
        return issues_for_review
    
    def _apply_expert_feedback(self, audit_results, feedback):
        """
        应用专家反馈到审计结果
        """
        import copy
        
        updated_results = copy.deepcopy(audit_results)
        
        # 处理每个反馈项
        for feedback_item in feedback:
            if feedback_item['type'] == 'issue_update':
                # 更新特定问题
                attr = feedback_item['attribute']
                issue_index = feedback_item['issue_index']
                update_fields = feedback_item['updates']
                
                if attr in updated_results['model_audit']['issues']:
                    issues = updated_results['model_audit']['issues']
                    if 0 <= issue_index < len(issues):
                        for field, value in update_fields.items():
                            issues[issue_index][field] = value
            
            elif feedback_item['type'] == 'metric_adjustment':
                # 调整指标权重或阈值
                metric = feedback_item['metric']
                adjustment = feedback_item['adjustment']
                
                # 这里应该有相应的逻辑来重新计算公平性得分
                # ...
        
        return updated_results
    
    def _learn_from_feedback(self, feedback):
        """
        从专家反馈中学习，更新知识库
        """
        # 实现简单的学习逻辑
        # 例如，统计特定类型反馈的频率，识别模式
        # ...
        pass

# ai_fairness_audit/__init__.py
"""
AI公平性审计框架 - 2025年版
提供全面的AI系统偏见检测、评估和缓解功能
"""

__version__ = "2.5.0"

from .core import AuditFramework
from .metrics import FairnessMetrics
from .reporting import ReportGenerator
from .preprocessing import DataPreprocessor
from .mitigation import BiasMitigation
from .monitoring import RealTimeMonitor

__all__ = [
    "AuditFramework",
    "FairnessMetrics", 
    "ReportGenerator",
    "DataPreprocessor",
    "BiasMitigation",
    "RealTimeMonitor"
]

# ai_fairness_audit/core.py
"""
审计框架核心模块
实现审计流程控制和协调
"""

import copy
import json
import datetime
import numpy as np
import pandas as pd

from .metrics import FairnessMetrics
from .reporting import ReportGenerator
from .preprocessing import DataPreprocessor
from .mitigation import BiasMitigation

class AuditFramework:
    """
    AI公平性审计框架主类
    协调各组件完成端到端的偏见审计流程
    """
    
    def __init__(self, config=None):
        """
        初始化审计框架
        
        参数:
        config: 审计配置字典
        """
        # 默认配置
        self.config = {
            'metrics': {
                'demographic_parity': True,
                'equal_opportunity': True,
                'equalized_odds': True,
                'disparate_impact': True,
                'statistical_parity_difference': True,
                'equalized_odds_difference': True,
                'average_odds_difference': True
            },
            'data_audit': {
                'balance_check': True,
                'representation_analysis': True,
                'sensitive_attribute_check': True
            },
            'model_audit': {
                'prediction_analysis': True,
                'feature_importance_audit': True,
                'adversarial_testing': False
            },
            'severity_thresholds': {
                'high': 0.2,
                'medium': 0.1
            }
        }
        
        # 更新配置
        if config:
            self._update_config(config)
        
        # 初始化组件
        self.metrics = FairnessMetrics()
        self.report_generator = ReportGenerator()
        self.preprocessor = DataPreprocessor()
        self.mitigation = BiasMitigation()
    
    def _update_config(self, new_config):
        """
        更新配置
        
        参数:
        new_config: 新的配置字典
        """
        for key, value in new_config.items():
            if key in self.config and isinstance(self.config[key], dict) and isinstance(value, dict):
                self.config[key].update(value)
            else:
                self.config[key] = value
    
    def run_audit(self, data, model=None, target_variable=None, audit_type='comprehensive'):
        """
        运行偏见审计
        
        参数:
        data: 要审计的数据集(pandas DataFrame)
        model: 要审计的模型(可选)
        target_variable: 目标变量名称(可选)
        audit_type: 审计类型，可选值:
            - 'comprehensive': 全面审计(数据+模型)
            - 'data_only': 仅数据审计
            - 'model_only': 仅模型审计
            
        返回:
        审计结果字典
        """
        # 验证输入
        self._validate_input(data, model, target_variable, audit_type)
        
        # 初始化结果
        results = {
            'audit_summary': {
                'timestamp': datetime.datetime.now().isoformat(),
                'audit_type': audit_type,
                'high_severity_issues': 0,
                'medium_severity_issues': 0,
                'low_severity_issues': 0
            },
            'sensitive_attributes': [],
            'data_audit': {},
            'model_audit': {},
            'recommendations': []
        }
        
        # 预处理数据
        processed_data = self._preprocess_data(data, target_variable)
        
        # 识别敏感属性
        sensitive_attributes = self._identify_sensitive_attributes(processed_data)
        results['sensitive_attributes'] = sensitive_attributes
        
        # 根据审计类型执行相应的审计
        if audit_type in ['comprehensive', 'data_only']:
            data_audit_results = self._run_data_audit(processed_data, sensitive_attributes, target_variable)
            results['data_audit'] = data_audit_results
            
            # 更新问题计数
            results['audit_summary']['high_severity_issues'] += data_audit_results.get('high_severity_issues', 0)
            results['audit_summary']['medium_severity_issues'] += data_audit_results.get('medium_severity_issues', 0)
            results['audit_summary']['low_severity_issues'] += data_audit_results.get('low_severity_issues', 0)
            
            # 添加数据审计的建议
            results['recommendations'].extend(data_audit_results.get('recommendations', []))
        
        if audit_type in ['comprehensive', 'model_only'] and model is not None:
            model_audit_results = self._run_model_audit(
                processed_data, 
                model, 
                sensitive_attributes, 
                target_variable
            )
            results['model_audit'] = model_audit_results
            
            # 更新问题计数
            results['audit_summary']['high_severity_issues'] += model_audit_results.get('high_severity_issues', 0)
            results['audit_summary']['medium_severity_issues'] += model_audit_results.get('medium_severity_issues', 0)
            results['audit_summary']['low_severity_issues'] += model_audit_results.get('low_severity_issues', 0)
            
            # 添加模型审计的建议
            results['recommendations'].extend(model_audit_results.get('recommendations', []))
        
        # 生成总体建议
        if results['recommendations']:
            results['prioritized_actions'] = self._prioritize_actions(results['recommendations'])
        
        return results
    
    def _validate_input(self, data, model, target_variable, audit_type):
        """
        验证输入参数的有效性
        
        参数:
        data: 要验证的数据集
        model: 要验证的模型
        target_variable: 目标变量名称
        audit_type: 审计类型
        """
        # 验证数据类型
        if not isinstance(data, pd.DataFrame):
            raise ValueError("data必须是pandas DataFrame类型")
        
        # 验证审计类型
        valid_audit_types = ['comprehensive', 'data_only', 'model_only']
        if audit_type not in valid_audit_types:
            raise ValueError(f"audit_type必须是以下之一: {', '.join(valid_audit_types)}")
        
        # 如果审计模型，确保提供了模型和目标变量
        if audit_type in ['comprehensive', 'model_only']:
            if model is None:
                raise ValueError("审计模型时必须提供model参数")
            if target_variable is None:
                raise ValueError("审计模型时必须提供target_variable参数")
            if target_variable not in data.columns:
                raise ValueError(f"target_variable '{target_variable}' 不在数据中")
    
    def _preprocess_data(self, data, target_variable):
        """
        预处理数据
        
        参数:
        data: 原始数据
        target_variable: 目标变量名称
        
        返回:
        预处理后的数据
        """
        # 创建数据副本以避免修改原始数据
        processed_data = copy.deepcopy(data)
        
        # 基本清理
        processed_data = self.preprocessor.clean_data(
            processed_data, 
            target_variable=target_variable
        )
        
        return processed_data
    
    def _identify_sensitive_attributes(self, data):
        """
        自动识别数据中的敏感属性
        
        参数:
        data: 数据集
        
        返回:
        敏感属性列表
        """
        # 常见敏感属性名称
        common_sensitive_attrs = [
            'gender', 'sex', 'race', 'ethnicity', 'religion', 'age', 'birth',
            'nationality', 'country_of_origin', 'marital_status', 'sexual_orientation',
            'disability', 'pregnancy', 'parenthood', 'veteran', 'military',
            'genetic', 'blood', 'health', 'disability_status',
            # 中文敏感属性
            '性别', '种族', '民族', '宗教', '年龄', '国籍', '婚姻状况',
            '性取向', '残疾', '怀孕', '育儿', '退伍', '遗传', '健康'
        ]
        
        # 检查哪些敏感属性在数据中
        sensitive_attributes = []
        for col in data.columns:
            # 直接匹配
            if col.lower() in common_sensitive_attrs:
                sensitive_attributes.append(col)
            # 部分匹配
            elif any(sa.lower() in col.lower() for sa in common_sensitive_attrs):
                sensitive_attributes.append(col)
        
        return sensitive_attributes
    
    def _run_data_audit(self, data, sensitive_attributes, target_variable):
        """
        运行数据审计
        
        参数:
        data: 数据集
        sensitive_attributes: 敏感属性列表
        target_variable: 目标变量名称
        
        返回:
        数据审计结果
        """
        results = {
            'issues': {},
            'statistics': {},
            'high_severity_issues': 0,
            'medium_severity_issues': 0,
            'low_severity_issues': 0,
            'recommendations': []
        }
        
        # 数据平衡性检查
        if self.config['data_audit']['balance_check']:
            balance_results = self._check_data_balance(data, sensitive_attributes, target_variable)
            results['statistics']['balance'] = balance_results
            
            # 识别问题
            for attr, stats in balance_results.items():
                if attr not in results['issues']:
                    results['issues'][attr] = []
                
                # 检查群体大小差异
                sizes = list(stats['group_sizes'].values())
                max_size = max(sizes)
                min_size = min(sizes)
                
                if max_size > 10 * min_size:  # 超过10倍的差异
                    severity = 'high'
                    results['high_severity_issues'] += 1
                elif max_size > 5 * min_size:  # 超过5倍的差异
                    severity = 'medium'
                    results['medium_severity_issues'] += 1
                else:
                    severity = 'low'
                    results['low_severity_issues'] += 1
                
                results['issues'][attr].append({
                    'type': 'imbalanced_representation',
                    'description': f'{attr}属性的群体表示不均衡',
                    'severity': severity,
                    'details': {
                        'max_group_size': max_size,
                        'min_group_size': min_size,
                        'ratio': max_size / min_size
                    }
                })
        
        # 代表性分析
        if self.config['data_audit']['representation_analysis']:
            # 这里可以实现更详细的代表性分析
            pass
        
        # 添加数据审计建议
        if results['issues']:
            results['recommendations'] = self._generate_data_recommendations(results['issues'])
        
        return results
    
    def _run_model_audit(self, data, model, sensitive_attributes, target_variable):
        """
        运行模型审计
        
        参数:
        data: 数据集
        model: 要审计的模型
        sensitive_attributes: 敏感属性列表
        target_variable: 目标变量名称
        
        返回:
        模型审计结果
        """
        results = {
            'issues': {},
            'fairness_scores': {
                'overall_score': 0,
                'attribute_scores': {}
            },
            'high_severity_issues': 0,
            'medium_severity_issues': 0,
            'low_severity_issues': 0,
            'recommendations': []
        }
        
        # 准备特征和标签
        features = data.drop([target_variable], axis=1) if target_variable in data.columns else data
        y_true = data[target_variable]
        
        # 获取模型预测
        try:
            y_pred = model.predict(features)
        except Exception as e:
            raise RuntimeError(f"模型预测失败: {str(e)}")
        
        # 如果模型支持概率预测，获取概率
        y_pred_proba = None
        if hasattr(model, 'predict_proba'):
            try:
                y_pred_proba = model.predict_proba(features)[:, 1]  # 二分类问题，取正类概率
            except:
                y_pred_proba = None
        
        # 计算每个敏感属性的公平性指标
        total_score = 0
        attribute_scores = {}
        
        for attr in sensitive_attributes:
            if attr not in data.columns:
                continue
            
            # 计算公平性指标
            metrics_results = self.metrics.calculate_fairness_metrics(
                y_true, 
                y_pred, 
                sensitive_attribute=data[attr],
                y_pred_proba=y_pred_proba,
                metrics_config=self.config['metrics']
            )
            
            attribute_scores[attr] = {
                'weighted_score': metrics_results['weighted_score'],
                'detailed_scores': metrics_results['detailed_scores'],
                'group_stats': metrics_results['group_stats']
            }
            
            total_score += metrics_results['weighted_score']
            
            # 识别问题
            if attr not in results['issues']:
                results['issues'][attr] = []
            
            # 基于指标阈值识别问题
            for metric_name, score in metrics_results['detailed_scores'].items():
                # 对于差异指标，较低的分数表示较大的差异
                if metric_name in ['statistical_parity_difference', 'equalized_odds_difference', 'average_odds_difference']:
                    # 这些指标应该接近0，表示差异小
                    if abs(score) > self.config['severity_thresholds']['high']:
                        severity = 'high'
                        results['high_severity_issues'] += 1
                    elif abs(score) > self.config['severity_thresholds']['medium']:
                        severity = 'medium'
                        results['medium_severity_issues'] += 1
                    else:
                        severity = 'low'
                        results['low_severity_issues'] += 1
                    
                    if severity != 'low':
                        results['issues'][attr].append({
                            'type': f'high_{metric_name}',
                            'description': f'{metric_name}指标表明存在显著差异',
                            'severity': severity,
                            'score': score
                        })
                elif metric_name == 'disparate_impact':
                    # 差异影响应该接近1
                    if abs(score - 1) > self.config['severity_thresholds']['high']:
                        severity = 'high'
                        results['high_severity_issues'] += 1
                    elif abs(score - 1) > self.config['severity_thresholds']['medium']:
                        severity = 'medium'
                        results['medium_severity_issues'] += 1
                    else:
                        severity = 'low'
                        results['low_severity_issues'] += 1
                    
                    if severity != 'low':
                        results['issues'][attr].append({
                            'type': 'high_disparate_impact',
                            'description': '差异影响表明存在潜在歧视',
                            'severity': severity,
                            'score': score
                        })
        
        # 计算总体公平性分数
        if sensitive_attributes:
            results['fairness_scores']['overall_score'] = total_score / len(sensitive_attributes)
        
        results['fairness_scores']['attribute_scores'] = attribute_scores
        
        # 特征重要性审计
        if self.config['model_audit']['feature_importance_audit'] and hasattr(model, 'feature_importances_'):
            importance_results = self._audit_feature_importance(model, features, sensitive_attributes)
            results['feature_importance_audit'] = importance_results
        
        # 添加模型审计建议
        if results['issues']:
            results['recommendations'] = self._generate_model_recommendations(results['issues'])
        
        return results
    
    def _check_data_balance(self, data, sensitive_attributes, target_variable):
        """
        检查数据平衡性
        
        参数:
        data: 数据集
        sensitive_attributes: 敏感属性列表
        target_variable: 目标变量名称
        
        返回:
        平衡性统计结果
        """
        balance_results = {}
        
        for attr in sensitive_attributes:
            if attr not in data.columns:
                continue
            
            # 计算群体大小
            group_sizes = data[attr].value_counts().to_dict()
            
            # 如果有目标变量，计算每个群体的目标变量分布
            target_distribution = {}
            if target_variable and target_variable in data.columns:
                for group in group_sizes.keys():
                    group_data = data[data[attr] == group]
                    target_dist = group_data[target_variable].value_counts(normalize=True).to_dict()
                    target_distribution[group] = target_dist
            
            balance_results[attr] = {
                'group_sizes': group_sizes,
                'target_distribution': target_distribution
            }
        
        return balance_results
    
    def _audit_feature_importance(self, model, features, sensitive_attributes):
        """
        审计特征重要性
        
        参数:
        model: 要审计的模型
        features: 特征数据
        sensitive_attributes: 敏感属性列表
        
        返回:
        特征重要性审计结果
        """
        # 获取特征重要性
        importances = model.feature_importances_
        
        # 创建特征重要性字典
        feature_importance = {}
        for i, feature in enumerate(features.columns):
            feature_importance[feature] = importances[i]
        
        # 检查敏感属性的重要性
        sensitive_importance = {}
        high_importance_sensitive = []
        
        # 获取最大重要性作为参考
        max_importance = max(importances) if importances.size > 0 else 1
        
        for attr in sensitive_attributes:
            if attr in feature_importance:
                importance = feature_importance[attr]
                sensitive_importance[attr] = importance
                
                # 如果敏感属性的重要性过高，标记为问题
                if importance > 0.1 * max_importance:  # 如果重要性超过最大重要性的10%
                    high_importance_sensitive.append(attr)
        
        # 排序特征重要性
        sorted_importance = sorted(feature_importance.items(), key=lambda x: x[1], reverse=True)
        
        return {
            'feature_importance': feature_importance,
            'sorted_importance': sorted_importance,
            'sensitive_attributes_importance': sensitive_importance,
            'high_importance_sensitive': high_importance_sensitive
        }
    
    def _generate_data_recommendations(self, issues):
        """
        生成数据相关的建议
        
        参数:
        issues: 发现的问题
        
        返回:
        建议列表
        """
        recommendations = []
        
        # 统计不同类型的问题
        issue_types = {}
        for attr, attr_issues in issues.items():
            for issue in attr_issues:
                issue_type = issue['type']
                if issue_type not in issue_types:
                    issue_types[issue_type] = []
                issue_types[issue_type].append(attr)
        
        # 为数据不平衡提供建议
        if 'imbalanced_representation' in issue_types:
            attrs = issue_types['imbalanced_representation']
            recommendations.append({
                'type': 'data_resampling',
                'description': f'针对{"、".join(attrs)}属性进行数据重采样，平衡群体表示',
                'priority': 'high' if any(issue['severity'] == 'high' for attr_issues in issues.values() 
                                         for issue in attr_issues if issue['type'] == 'imbalanced_representation')
                            else 'medium',
                'suggestions': [
                    '考虑使用过采样技术(如SMOTE)增加少数群体数据',
                    '对多数群体进行欠采样，保持数据分布',
                    '收集更多的少数群体数据'
                ]
            })
        
        return recommendations
    
    def _generate_model_recommendations(self, issues):
        """
        生成模型相关的建议
        
        参数:
        issues: 发现的问题
        
        返回:
        建议列表
        """
        recommendations = []
        
        # 统计每个敏感属性的问题严重性
        attr_severity = {}
        for attr, attr_issues in issues.items():
            max_severity = max(issue['severity'] for issue in attr_issues)
            attr_severity[attr] = max_severity
        
        # 根据问题严重性生成建议
        high_severity_attrs = [attr for attr, severity in attr_severity.items() if severity == 'high']
        medium_severity_attrs = [attr for attr, severity in attr_severity.items() if severity == 'medium']
        
        if high_severity_attrs:
            recommendations.append({
                'type': 'model_retraining',
                'description': f'针对{"、".join(high_severity_attrs)}属性重新训练模型，使用公平性约束',
                'priority': 'high',
                'suggestions': [
                    '使用带有公平性正则化的模型',
                    '尝试对抗性去偏方法',
                    '考虑使用公平性感知的算法(如公平决策树)'  
                ]
            })
        
        if medium_severity_attrs:
            recommendations.append({
                'type': 'postprocessing_adjustment',
                'description': f'对{"、".join(medium_severity_attrs)}属性应用后处理公平性调整',
                'priority': 'medium',
                'suggestions': [
                    '使用阈值校准方法调整不同群体的决策阈值',
                    '应用重新加权技术',
                    '考虑使用公平性约束的预测校正'
                ]
            })
        
        return recommendations
    
    def _prioritize_actions(self, recommendations):
        """
        对建议进行优先级排序
        
        参数:
        recommendations: 建议列表
        
        返回:
        排序后的建议
        """
        # 按优先级排序
        priority_order = {'high': 0, 'medium': 1, 'low': 2}
        prioritized = sorted(recommendations, 
                           key=lambda x: priority_order.get(x.get('priority', 'medium'), 1))
        
        # 添加执行顺序
        for i, rec in enumerate(prioritized):
            rec['execution_order'] = i + 1
        
        return prioritized
    
    def apply_mitigation(self, audit_results, model, data, target_variable):
        """
        根据审计结果应用缓解措施
        
        参数:
        audit_results: 审计结果
        model: 要优化的模型
        data: 数据集
        target_variable: 目标变量
        
        返回:
        优化后的模型和缓解报告
        """
        # 获取优先级最高的建议
        if 'prioritized_actions' in audit_results:
            prioritized_actions = audit_results['prioritized_actions']
        else:
            prioritized_actions = self._prioritize_actions(audit_results['recommendations'])
        
        # 应用缓解措施
        mitigation_report = {
            'applied_actions': [],
            'before_after': {}
        }
        
        # 记录缓解前的性能
        X = data.drop([target_variable], axis=1) if target_variable in data.columns else data
        y = data[target_variable]
        
        before_performance = self._evaluate_model_performance(model, X, y)
        mitigation_report['before_after']['performance'] = before_performance
        
        # 应用每个建议的缓解措施
        current_model = copy.deepcopy(model)
        
        for action in prioritized_actions[:3]:  # 优先应用前3个高优先级建议
            action_type = action['type']
            
            try:
                if action_type == 'data_resampling':
                    # 应用数据重采样
                    adjusted_data = self.mitigation.resample_data(
                        data, 
                        audit_results['sensitive_attributes'],
                        target_variable
                    )
                    
                    # 使用调整后的数据重新训练模型
                    X_adjusted = adjusted_data.drop([target_variable], axis=1)
                    y_adjusted = adjusted_data[target_variable]
                    
                    # 重新训练
                    current_model.fit(X_adjusted, y_adjusted)
                    
                    mitigation_report['applied_actions'].append({
                        'type': action_type,
                        'status': 'success',
                        'details': '使用重采样数据重新训练模型'
                    })
                
                elif action_type == 'model_retraining':
                    # 应用模型级缓解
                    current_model = self.mitigation.apply_fairness_constraints(
                        current_model,
                        data,
                        audit_results['sensitive_attributes'],
                        target_variable
                    )
                    
                    mitigation_report['applied_actions'].append({
                        'type': action_type,
                        'status': 'success',
                        'details': '应用公平性约束重新训练模型'
                    })
                
                elif action_type == 'postprocessing_adjustment':
                    # 应用后处理缓解
                    current_model = self.mitigation.apply_postprocessing_adjustment(
                        current_model,
                        data,
                        audit_results['sensitive_attributes'],
                        target_variable
                    )
                    
                    mitigation_report['applied_actions'].append({
                        'type': action_type,
                        'status': 'success',
                        'details': '应用后处理公平性调整'
                    })
            except Exception as e:
                mitigation_report['applied_actions'].append({
                    'type': action_type,
                    'status': 'failed',
                    'error': str(e)
                })
        
        # 重新运行审计以评估缓解效果
        mitigation_results = self.run_audit(
            data=data,
            model=current_model,
            target_variable=target_variable,
            audit_type='model_only'
        )
        
        mitigation_report['after_audit'] = mitigation_results
        
        # 计算缓解后的性能
        after_performance = self._evaluate_model_performance(current_model, X, y)
        mitigation_report['before_after']['performance'] = {
            'before': before_performance,
            'after': after_performance
        }
        
        # 计算公平性改进
        if 'fairness_scores' in audit_results.get('model_audit', {}) and 'fairness_scores' in mitigation_results.get('model_audit', {}):
            before_fairness = audit_results['model_audit']['fairness_scores']['overall_score']
            after_fairness = mitigation_results['model_audit']['fairness_scores']['overall_score']
            
            mitigation_report['before_after']['fairness'] = {
                'before': before_fairness,
                'after': after_fairness,
                'improvement': after_fairness - before_fairness
            }
        
        return current_model, mitigation_report
    
    def _evaluate_model_performance(self, model, X, y):
        """
        评估模型性能
        
        参数:
        model: 要评估的模型
        X: 特征
        y: 标签
        
        返回:
        性能指标字典
        """
        from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
        
        y_pred = model.predict(X)
        
        return {
            'accuracy': accuracy_score(y, y_pred),
            'precision': precision_score(y, y_pred, average='weighted'),
            'recall': recall_score(y, y_pred, average='weighted'),
            'f1_score': f1_score(y, y_pred, average='weighted')
        }
    
    def save_audit_results(self, results, file_path):
        """
        保存审计结果到文件
        
        参数:
        results: 审计结果
        file_path: 文件路径
        """
        # 转换numpy类型以便JSON序列化
        def convert_numpy(obj):
            if isinstance(obj, np.integer):
                return int(obj)
            elif isinstance(obj, np.floating):
                return float(obj)
            elif isinstance(obj, np.ndarray):
                return obj.tolist()
            return obj
        
        # 递归转换所有numpy类型
        def recursive_convert(data):
            if isinstance(data, dict):
                return {k: recursive_convert(v) for k, v in data.items()}
            elif isinstance(data, list):
                return [recursive_convert(item) for item in data]
            else:
                return convert_numpy(data)
        
        # 转换结果
        serializable_results = recursive_convert(results)
        
        # 保存到文件
        with open(file_path, 'w', encoding='utf-8') as f:
            json.dump(serializable_results, f, ensure_ascii=False, indent=2)
    
    def load_audit_results(self, file_path):
        """
        从文件加载审计结果
        
        参数:
        file_path: 文件路径
        
        返回:
        审计结果
        """
        with open(file_path, 'r', encoding='utf-8') as f:
            results = json.load(f)
        
        return results

# 金融行业信贷评分模型的特定审计配置
finance_audit_config = {
    'metrics': {
        'demographic_parity': True,
        'equal_opportunity': True,
        'disparate_impact': True,
        'statistical_parity_difference': True
    },
    'data_audit': {
        'balance_check': True,
        'representation_analysis': True,
        'sensitive_attribute_check': True,
        'credit_history_distribution': True,  # 金融特定检查
        'income_distribution': True           # 金融特定检查
    },
    'model_audit': {
        'prediction_analysis': True,
        'feature_importance_audit': True,
        'threshold_analysis': True,           # 金融特定检查
        'credit_score_distribution': True     # 金融特定检查
    },
    'severity_thresholds': {
        'high': 0.15,  # 金融行业更严格的阈值
        'medium': 0.08
    },
    'regulatory_compliance': {
        'equal_credit_opportunity_act': True,
        'community_reinvestment_act': True
    }
}

# 使用行业特定配置
finance_framework = AuditFramework(config=finance_audit_config)

# 运行金融行业特定审计
audit_results = finance_framework.run_audit(
    data=credit_data,
    model=credit_scoring_model,
    target_variable='default',
    audit_type='comprehensive'
)

# 医疗行业诊断模型的特定审计配置
healthcare_audit_config = {
    'metrics': {
        'demographic_parity': True,
        'equal_opportunity': True,
        'equalized_odds': True,
        'treatment_equity': True  # 医疗特定指标
    },
    'data_audit': {
        'balance_check': True,
        'representation_analysis': True,
        'sensitive_attribute_check': True,
        'clinical_feature_distribution': True,  # 医疗特定检查
        'healthcare_access_analysis': True       # 医疗特定检查
    },
    'model_audit': {
        'prediction_analysis': True,
        'feature_importance_audit': True,
        'clinical_relevance_check': True,        # 医疗特定检查
        'false_positive_analysis': True          # 医疗特定检查
    },
    'severity_thresholds': {
        'high': 0.1,   # 医疗行业更严格的阈值
        'medium': 0.05
    },
    'regulatory_compliance': {
        'hipaa_compliance': True,
        'medical_equity_guidelines': True
    }
}

# 使用行业特定配置
healthcare_framework = AuditFramework(config=healthcare_audit_config)

# 运行医疗行业特定审计
audit_results = healthcare_framework.run_audit(
    data=patient_data,
    model=diagnostic_model,
    target_variable='condition_present',
    audit_type='comprehensive'
)