AI 偏见与公平性

import pandas as pd
import numpy as np

# 检查数据集中敏感属性的分布
def check_representation(df, sensitive_attr):
    """检查敏感属性在数据集中的分布"""
    dist = df[sensitive_attr].value_counts(normalize=True)
    print(f"--- {sensitive_attr} 分布 ---")
    for group, ratio in dist.items():
        print(f"  {group}: {ratio:.2%}")
    return dist

df = pd.read_csv("hiring_data.csv")
check_representation(df, "gender")

# 检测标注者之间的偏见一致性
from sklearn.metrics import cohen_kappa_score

def check_annotator_bias(labels_a, labels_b):
    """计算两个标注者之间的一致性"""
    kappa = cohen_kappa_score(labels_a, labels_b)
    if kappa < 0.4:
        print(f"警告：标注一致性较低 (kappa={kappa:.3f})")
    elif kappa < 0.6:
        print(f"注意：标注一致性中等 (kappa={kappa:.3f})")
    else:
        print(f"标注一致性良好 (kappa={kappa:.3f})")
    return kappa

# 不同标注者对同一组简历的评分
kappa = check_annotator_bias(ratings_A, ratings_B)

def statistical_parity_difference(y_pred, sensitive, pos_label=1):
    """计算统计 Parity 差异"""
    groups = set(sensitive)
    rates = {}
    for g in groups:
        mask = sensitive == g
        rates[g] = np.mean(y_pred[mask] == pos_label)
    # 最大组与最小组的差异
    diff = max(rates.values()) - min(rates.values())
    print(f"各组阳性预测率: {rates}")
    print(f"统计 Parity 差异: {diff:.4f}")
    return diff, rates

from sklearn.metrics import confusion_matrix

def equal_opportunity_difference(y_true, y_pred, sensitive, pos_label=1):
    """计算机会均等差异（TPR 差异）"""
    groups = set(sensitive)
    tprs = {}
    for g in groups:
        mask = sensitive == g
        tn, fp, fn, tp = confusion_matrix(
            y_true[mask], y_pred[mask], labels=[0, 1]
        ).ravel()
        tprs[g] = tp / (tp + fn) if (tp + fn) > 0 else 0
    diff = max(tprs.values()) - min(tprs.values())
    print(f"各组真正例率: {tprs}")
    print(f"机会均等差异: {diff:.4f}")
    return diff, tprs

def disparate_impact_ratio(y_pred, sensitive, pos_label=1):
    """计算差异影响比率（4/5 法则）"""
    groups = set(sensitive)
    rates = {}
    for g in groups:
        mask = sensitive == g
        rates[g] = np.mean(y_pred[mask] == pos_label)
    # 取最小值除以最大值
    sorted_rates = sorted(rates.values())
    ratio = sorted_rates[0] / sorted_rates[1] if sorted_rates[1] > 0 else 0
    print(f"差异影响比率: {ratio:.4f}")
    if ratio < 0.8:
        print("  警告：违反 4/5 法则（EEOC 标准）")
    return ratio

# 4/5 法则：若比率 < 0.8，可能存在歧视

def theil_index_fairness(y_pred, y_true, sensitive, pos_label=1):
    """计算公平性的 Theil 指数分解"""
    groups = list(set(sensitive))
    # 计算各组的错误率
    error_rates = []
    for g in groups:
        mask = sensitive == g
        err = np.mean(y_pred[mask] != y_true[mask])
        error_rates.append(err)
    
    mean_err = np.mean(error_rates)
    n = len(error_rates)
    if mean_err == 0:
        return 0.0
    # Theil 指数
    theil = np.sum([e * np.log(e / mean_err) if e > 0 else 0 
                    for e in error_rates]) / (n * mean_err)
    print(f"群体间 Theil 指数: {theil:.4f}")
    return theil

# 预处理：重加权
import numpy as np

def compute_reweighting_weights(y, sensitive):
    """计算重加权权重"""
    classes = np.unique(y)
    groups = np.unique(sensitive)
    n = len(y)
    weights = np.ones(n)
    
    for c in classes:
        for g in groups:
            mask = (y == c) & (sensitive == g)
            n_cg = mask.sum()
            n_c = (y == c).sum()
            n_g = (sensitive == g).sum()
            if n_cg > 0 and n_c > 0 and n_g > 0:
                weights[mask] = (n_c * n_g) / (n * n_cg)
    return weights

weights = compute_reweighting_weights(y_train, sensitive_train)
model.fit(X_train, y_train, sample_weight=weights)

# 后处理：阈值调整
from sklearn.metrics import roc_curve

def find_optimal_thresholds(y_true, y_score, sensitive, target_tpr=0.8):
    """为不同群体找到满足目标 TPR 的最优阈值"""
    thresholds = {}
    groups = set(sensitive)
    for g in groups:
        mask = sensitive == g
        fpr, tpr, thr = roc_curve(y_true[mask], y_score[mask])
        # 找到达到目标 TPR 的阈值
        idx = np.searchsorted(tpr, target_tpr)
        if idx < len(thr):
            thresholds[g] = thr[idx]
        else:
            thresholds[g] = thr[-1]
        print(f"  {g}: threshold={thresholds[g]:.4f}")
    return thresholds

# 使用不同阈值进行预测
for g in groups:
    mask = sensitive_test == g
    y_pred[mask] = (y_score[mask] >= thresholds[g]).astype(int)

# 模拟 COMPAS 偏见分析
import pandas as pd
from sklearn.metrics import confusion_matrix

def analyze_compas_bias(df):
    """分析 COMPAS 系统的群体差异"""
    races = df["race"].unique()
    results = {}
    for race in races:
        subset = df[df["race"] == race]
        tn, fp, fn, tp = confusion_matrix(
            subset["actual_recidivism"],
            subset["predicted_risk"],
            labels=[0, 1]
        ).ravel()
        fpr = fp / (fp + tn)  # 假阳性率
        fnr = fn / (fn + tp)  # 假阴性率
        results[race] = {"FPR": fpr, "FNR": fnr}
    return pd.DataFrame(results).T

# ProPublica 发现：黑人 FPR 更高，白人 FNR 更高

# 检测贷款审批中的代理歧视
from sklearn.ensemble import RandomForestClassifier
from sklearn.inspection import permutation_importance

def detect_proxy_discrimination(X, y, sensitive, protected_features):
    """检测敏感属性是否通过代理变量间接影响预测"""
    model = RandomForestClassifier(random_state=42)
    model.fit(X, y)
    
    # 置换重要性分析
    result = permutation_importance(model, X, y, n_repeats=10)
    importances = result.importances_mean
    
    # 检查受保护属性相关的特征
    for feat, imp in zip(X.columns, importances):
        if feat in protected_features:
            print(f"  {feat}: importance={imp:.4f}")
            if imp > 0.1:
                print(f"    警告：{feat} 可能是敏感属性的代理变量")

# 绘制公平性-准确性前沿
import numpy as np
import matplotlib.pyplot as plt

def fairness_accuracy_frontier(X, y, sensitive, alphas):
    """计算不同公平性约束下的准确性前沿"""
    from sklearn.linear_model import LogisticRegression
    
    accuracies = []
    fairness_gaps = []
    
    for alpha in alphas:
        # 简化示例：使用正则化强度模拟公平性约束
        model = LogisticRegression(C=alpha, max_iter=1000)
        model.fit(X, y)
        y_pred = model.predict(X)
        
        acc = np.mean(y_pred == y)
        # 计算公平性差距
        groups = set(sensitive)
        rates = [np.mean(y_pred[sensitive == g]) for g in groups]
        gap = max(rates) - min(rates)
        
        accuracies.append(acc)
        fairness_gaps.append(gap)
    
    return accuracies, fairness_gaps

# 量化公平性约束下的性能损失
def measure_fairness_cost(base_model, debiased_model, 
                          X_test, y_test, sensitive):
    """计算去偏带来的性能代价"""
    y_base = base_model.predict(X_test)
    y_debias = debiased_model.predict(X_test)
    
    acc_base = np.mean(y_base == y_test)
    acc_debias = np.mean(y_debias == y_test)
    
    # 公平性改善
    groups = set(sensitive)
    base_rates = [np.mean(y_base[sensitive == g]) for g in groups]
    debias_rates = [np.mean(y_debias[sensitive == g]) for g in groups]
    
    base_gap = max(base_rates) - min(base_rates)
    debias_gap = max(debias_rates) - min(debias_rates)
    
    print(f"准确率变化: {acc_base:.4f} -> {acc_debias:.4f} (损失: {acc_base-acc_debias:.4f})")
    print(f"公平性差距: {base_gap:.4f} -> {debias_gap:.4f} (改善: {base_gap-debias_gap:.4f})")
    return acc_base - acc_debias, base_gap - debias_gap

from aif360.datasets import BinaryLabelDataset
from aif360.metrics import BinaryLabelDatasetMetric
import pandas as pd

# 1. 加载数据并定义受保护属性
df = pd.read_csv("adult_dataset.csv")
dataset = BinaryLabelDataset(
    df=df,
    label_names=["income"],
    protected_attribute_names=["race", "sex"],
    favorable_classes=[1],
    privileged_classes=[["White"], ["Male"]]
)

# 2. 计算去偏前的公平性指标
metric = BinaryLabelDatasetMetric(
    dataset,
    unprivileged_groups=[{"race": 0, "sex": 0}],
    privileged_groups=[{"race": 1, "sex": 1}]
)

print(f"统计 Parity 差异: {metric.statistical_parity_difference():.4f}")
print(f"差异影响比率: {metric.disparate_impact():.4f}")

from aif360.algorithms.preprocessing import Reweighing
from aif360.metrics import ClassificationMetric
from sklearn.linear_model import LogisticRegression

# 3. 应用 Reweighing 去偏
rw = Reweighing(
    unprivileged_groups=[{"race": 0, "sex": 0}],
    privileged_groups=[{"race": 1, "sex": 1}]
)
dataset_rw = rw.fit_transform(dataset)

# 4. 训练模型
model = LogisticRegression(max_iter=1000)
model.fit(dataset_rw.features, dataset_rw.labels.ravel())
y_pred = model.predict(dataset_rw.features)

# 5. 评估去偏效果
pred_dataset = dataset_rw.copy()
pred_dataset.labels = y_pred.reshape(-1, 1)
pred_metric = ClassificationMetric(
    dataset_rw, pred_dataset,
    unprivileged_groups=[{"race": 0, "sex": 0}],
    privileged_groups=[{"race": 1, "sex": 1}]
)
print(f"去偏后统计 Parity 差异: {pred_metric.statistical_parity_difference():.4f}")
print(f"去偏后准确率: {pred_metric.accuracy():.4f}")