AI 偏见缓解技术：从数据到推理的全链路方案

# 常见公平性指标计算
import numpy as np
from sklearn.metrics import confusion_matrix

def fairness_metrics(y_true, y_pred, sensitive_attr):
    """
    计算常见公平性指标
    y_true: 真实标签 (0/1)
    y_pred: 预测标签 (0/1)
    sensitive_attr: 敏感属性 (0/1)，如性别、种族
    """
    group_0 = sensitive_attr == 0
    group_1 = sensitive_attr == 1

    # Demographic Parity Difference (DPD)
    # 要求不同群体的预测正例率相同
    pred_positive_rate_0 = y_pred[group_0].mean()
    pred_positive_rate_1 = y_pred[group_1].mean()
    dpd = abs(pred_positive_rate_0 - pred_positive_rate_1)

    # Equal Opportunity Difference (EOD)
    # 要求不同群体的真正例率相同
    tp_0 = y_pred[group_0 & (y_true == 1)].mean()
    tp_1 = y_pred[group_1 & (y_true == 1)].mean()
    eod = abs(tp_0 - tp_1)

    # Equalized Odds Difference (EQOD)
    # 要求不同群体的真正例率和假正例率都相同
    fp_0 = y_pred[group_0 & (y_true == 0)].mean()
    fp_1 = y_pred[group_1 & (y_true == 0)].mean()
    eqod = max(abs(tp_0 - tp_1), abs(fp_0 - fp_1))

    # Disparate Impact Ratio (DIR)
    dir_ratio = min(pred_positive_rate_0, pred_positive_rate_1) / \
                max(pred_positive_rate_0, pred_positive_rate_1, 1e-10)

    return {
        "demographic_parity_diff": dpd,
        "equal_opportunity_diff": eod,
        "equalized_odds_diff": eqod,
        "disparate_impact_ratio": dir_ratio,
        "positive_rate_group_0": pred_positive_rate_0,
        "positive_rate_group_1": pred_positive_rate_1,
    }

from imblearn.over_sampling import SMOTE
import pandas as pd

def fairness_aware_oversample(df, sensitive_col, label_col, target_col):
    """
    基于公平性的过采样
    确保每个敏感属性组合的样本比例均衡
    """
    # 计算每个组合的样本数
    group_counts = df.groupby([sensitive_col, label_col]).size()
    target_count = group_counts.max()

    resampled_dfs = []

    for (sensitive_val, label_val), group_df in df.groupby([sensitive_col, label_col]):
        if len(group_df) < target_count:
            # 对少数群体进行 SMOTE 过采样
            smote = SMOTE(sampling_strategy=target_count, random_state=42)
            X = group_df.drop(columns=[label_col, sensitive_col])
            y = group_df[label_col]
            X_resampled, y_resampled = smote.fit_resample(X, y)
            # 重新组装 DataFrame
            resampled_group = pd.concat([
                X_resampled,
                pd.DataFrame({label_col: y_resampled}),
                pd.DataFrame({sensitive_col: [sensitive_val] * len(X_resampled)})
            ], axis=1)
            resampled_dfs.append(resampled_group)
        else:
            resampled_dfs.append(group_df)

    return pd.concat(resampled_dfs, ignore_index=True)

def compute_fairness_weights(df, sensitive_col, label_col):
    """
    计算公平性感知的样本权重
    思路：调整权重使得敏感属性与标签的联合分布均匀
    """
    n = len(df)
    group_sizes = df.groupby([sensitive_col, label_col]).size()

    weights = []
    for _, row in df.iterrows():
        s_val = row[sensitive_col]
        y_val = row[label_col]
        group_size = group_sizes[(s_val, y_val)]
        # 权重 = 总样本数 / (组数 * 组大小)
        n_groups = len(group_sizes)
        weight = n / (n_groups * group_size)
        weights.append(weight)

    return np.array(weights)

# 在模型训练中使用
# sample_weight = compute_fairness_weights(train_df, "gender", "label")
# model.fit(X_train, y_train, sample_weight=sample_weight)

import torch
import torch.nn as nn

class AdversarialDebiasing(nn.Module):
    """
    对抗性去偏模型
    主预测器尝试预测目标
    对抗器尝试从隐藏层预测敏感属性
    """

    def __init__(self, input_dim, hidden_dim, num_classes, num_sensitive):
        super().__init__()

        # 主预测器
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
        )

        self.predictor = nn.Linear(hidden_dim, num_classes)

        # 对抗器（用于学习去偏表示）
        self.adversary = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, num_sensitive),
        )

    def forward(self, x, return_hidden=False):
        hidden = self.encoder(x)
        pred = self.predictor(hidden)
        if return_hidden:
            return pred, hidden
        return pred

    def adversary_pred(self, hidden):
        return self.adversary(hidden)


def train_debiased_model(model, dataloader, sensitive_attr, epochs=10,
                         lambda_adv=1.0):
    """
    训练对抗性去偏模型
    lambda_adv: 对抗损失的权重（越大去偏越强）
    """
    optimizer_pred = torch.optim.Adam(
        list(model.encoder.parameters()) + list(model.predictor.parameters()),
        lr=0.001
    )
    optimizer_adv = torch.optim.Adam(model.adversary.parameters(), lr=0.001)

    criterion_task = nn.CrossEntropyLoss()
    criterion_adv = nn.CrossEntropyLoss()

    for epoch in range(epochs):
        for batch_x, batch_y, batch_s in dataloader:
            # Step 1: 更新对抗器
            optimizer_adv.zero_grad()
            _, hidden = model(batch_x, return_hidden=True)
            adv_pred = model.adversary_pred(hidden.detach())
            loss_adv = criterion_adv(adv_pred, batch_s)
            loss_adv.backward()
            optimizer_adv.step()

            # Step 2: 更新主预测器（最小化任务损失，最大化对抗损失）
            optimizer_pred.zero_grad()
            pred, hidden = model(batch_x, return_hidden=True)
            loss_task = criterion_task(pred, batch_y)
            adv_pred = model.adversary_pred(hidden)
            # 梯度反转：最大化对抗器预测敏感属性的能力
            # 即让隐藏表示对敏感属性"不可预测"
            loss_total = loss_task - lambda_adv * criterion_adv(adv_pred, batch_s)
            loss_total.backward()
            optimizer_pred.step()

    return model

# 使用 Fairlearn 进行公平性约束训练
from fairlearn.reductions import ExponentiatedGradient, DemographicParity
from sklearn.linear_model import LogisticRegression

def fair_train_with_constraint(X_train, y_train, sensitive_features,
                               constraint="demographic_parity", eps=0.01):
    """
    使用 Fairlearn 的减少方法进行公平性约束训练
    """
    # 定义基础模型
    base_model = LogisticRegression(solver='liblinear', fit_intercept=True)

    # 定义公平性约束
    if constraint == "demographic_parity":
        fairness_constraint = DemographicParity()
    elif constraint == "equalized_odds":
        from fairlearn.reductions import EqualizedOdds
        fairness_constraint = EqualizedOdds()
    else:
        raise ValueError(f"不支持的约束类型: {constraint}")

    # 创建约束优化器
    mitigator = ExponentiatedGradient(
        base_model,
        constraints=fairness_constraint,
        eps=eps  # 容忍度
    )

    # 训练
    mitigator.fit(X_train, y_train, sensitive_features=sensitive_features)

    return mitigator

# 使用示例
# mitigator = fair_train_with_constraint(
#     X_train, y_train, sensitive_features,
#     constraint="demographic_parity", eps=0.01
# )
# y_pred_fair = mitigator.predict(X_test)

def calibrate_thresholds(y_proba, sensitive_attr, target_metric="demographic_parity"):
    """
    为不同敏感群体设置不同阈值以实现公平性
    """
    unique_groups = np.unique(sensitive_attr)
    thresholds = {}

    if target_metric == "demographic_parity":
        # 目标：使各群体的正例预测率相同
        target_rate = y_proba.mean()  # 全局正例率作为目标
        for group in unique_groups:
            group_proba = y_proba[sensitive_attr == group]
            # 找到使该群体预测正例率达到目标的阈值
            thresholds[group] = np.percentile(group_proba,
                                              (1 - target_rate) * 100)

    elif target_metric == "equalized_odds":
        # 需要迭代优化以满足 TPR 和 FPR 相等
        pass  # 实现略

    return thresholds


def fair_predict(y_proba, sensitive_attr, thresholds):
    """
    使用分组阈值进行预测
    """
    predictions = np.zeros_like(y_proba, dtype=int)

    for group, threshold in thresholds.items():
        group_mask = sensitive_attr == group
        predictions[group_mask] = (y_proba[group_mask] >= threshold).astype(int)

    return predictions

def ensemble_fairness(models, X, sensitive_attr, strategy="voting"):
    """
    集成多个模型的预测结果以减少偏见
    """
    predictions = np.column_stack([m.predict(X) for m in models])

    if strategy == "voting":
        # 简单投票
        return predictions.mean(axis=1) >= 0.5

    elif strategy == "fair_voting":
        # 公平性加权投票：为每个群体单独调整权重
        final_pred = np.zeros(len(X))
        for group in np.unique(sensitive_attr):
            group_mask = sensitive_attr == group
            group_pred = predictions[group_mask].mean(axis=1)
            final_pred[group_mask] = group_pred >= 0.5
        return final_pred

    elif strategy == "stacking":
        # 使用元学习器学习如何组合（需额外训练）
        pass

# 使用 AIF360 进行全面公平性评估
from aif360.metrics import BinaryLabelDatasetMetric, ClassificationMetric
from aif360.datasets import BinaryLabelDataset

def comprehensive_fairness_eval(dataset, predictions, sensitive_attr_idx=0):
    """
    全面的公平性评估
    """
    # 创建评估数据集
    pred_dataset = dataset.copy()
    pred_dataset.labels = predictions

    # 二分类指标
    metric = ClassificationMetric(
        dataset, pred_dataset,
        unprivileged_groups=[{sensitive_attr_idx: 0}],
        privileged_groups=[{sensitive_attr_idx: 1}]
    )

    return {
        # 差异指标
        "statistical_parity_diff": metric.statistical_parity_difference(),
        "disparate_impact": metric.disparate_impact(),
        "equal_opportunity_diff": metric.equal_opportunity_difference(),
        "average_odds_diff": metric.average_odds_difference(),

        # 绝对值指标
        "false_positive_rate_diff": metric.difference(
            metric.false_positive_rate
        ),
        "false_negative_rate_diff": metric.difference(
            metric.false_negative_rate
        ),
        "accuracy_diff": metric.difference(metric.accuracy),

        # 整体性能
        "overall_accuracy": metric.accuracy(),
        "balanced_accuracy": metric.accuracy(
            privileged=False
        ) * 0.5 + metric.accuracy(privileged=True) * 0.5,
    }