公平性与偏差缓解：从度量到去偏实战

P(Ŷ = 1 | A = 0) = P(Ŷ = 1 | A = 1)

P(Ŷ = 1 | Y = 1, A = 0) = P(Ŷ = 1 | Y = 1, A = 1)

P(Ŷ = 1 | Y = y, A = 0) = P(Ŷ = 1 | Y = y, A = 1),  y ∈ {0, 1}

基率不同时：
  预测均等  ⊥  均等化几率  ⊥  校准
  必须根据业务选择其中一个作为主目标

import numpy as np

def disparate_impact(y_pred, group):
    rates = {g: y_pred[group == g].mean() for g in np.unique(group)}
    return min(rates.values()) / max(rates.values()), rates

y_pred = np.array([1, 0, 1, 1, 0, 1, 0, 0])
group = np.array(["A", "A", "A", "A", "B", "B", "B", "B"])
di, rates = disparate_impact(y_pred, group)
print("各组通过率:", {k: round(v, 3) for k, v in rates.items()})
print("差异影响率:", round(di, 3), "(<0.8 需审查)")

import numpy as np
import pandas as pd
from sklearn.metrics import confusion_matrix

def group_metrics(y_true, y_pred, group):
    rows = []
    for g in np.unique(group):
        m = group == g
        tn, fp, fn, tp = confusion_matrix(y_true[m], y_pred[m]).ravel()
        rows.append({
            "group": g, "n": m.sum(),
            "TPR": tp / (tp + fn) if (tp + fn) else 0,
            "FPR": fp / (fp + tn) if (fp + tn) else 0,
            "precision": tp / (tp + fp) if (tp + fp) else 0,
        })
    return pd.DataFrame(rows)

y_true = np.array([1, 1, 0, 0, 1, 0, 1, 0])
y_pred = np.array([1, 0, 0, 0, 1, 1, 1, 0])
group = np.array(["A", "A", "A", "A", "B", "B", "B", "B"])
print(group_metrics(y_true, y_pred, group))

from fairlearn.metrics import MetricFrame, selection_rate, true_positive_rate
from sklearn.metrics import accuracy_score

mf = MetricFrame(
    metrics={"accuracy": accuracy_score,
             "selection_rate": selection_rate,
             "tpr": true_positive_rate},
    y_true=y_true, y_pred=y_pred, sensitive_features=group,
)
print("总体:", mf.overall)
print("分组:\n", mf.by_group)
print("组间差:", mf.difference())

import numpy as np

def compute_weights(y, group):
    """使各群体在标签上的权重和相等"""
    w = np.ones(len(y))
    for g in np.unique(group):
        for label in np.unique(y):
            mask = (group == g) & (y == label)
            p_group = (group == g).mean()
            p_label = (y == label).mean()
            p_joint = mask.mean()
            if p_joint > 0:
                w[mask] = (p_group * p_label) / p_joint
    return w

y = np.array([1, 1, 0, 0, 1, 0, 1, 0])
group = np.array(["A", "A", "A", "A", "B", "B", "B", "B"])
print("样本权重:", compute_weights(y, group).round(3))

L = L_task + λ · L_fairness

import numpy as np
from sklearn.linear_model import LogisticRegression

def fairness_penalty(scores, group):
    """人口均等惩罚：各组平均预测分数之差的平方"""
    means = [scores[group == g].mean() for g in np.unique(group)]
    return np.var(means)

# 示意：λ 越大，公平性越强、精度越低
for lam in [0.0, 0.1, 1.0]:
    print(f"lambda={lam}: 惩罚示例 {round(fairness_penalty(np.array([0.9,0.8,0.3,0.2]), np.array(['A','A','B','B'])), 4)}")

主模型 → 预测 Ŷ  → 任务损失
       ↘ 中间表示 Z → 对手网络 → 敏感属性 A
                     ↺ 主模型学习让对手失效（梯度反转）

from fairlearn.reductions import ExponentiatedGradient, DemographicParity
from sklearn.linear_model import LogisticRegression

mitigator = ExponentiatedGradient(
    estimator=LogisticRegression(max_iter=1000),
    constraints=DemographicParity(),
)
# mitigator.fit(X, y, sensitive_features=group)

import numpy as np

def group_thresholds(scores, group, target_rate):
    """为每个群体找阈值，使正例率达到目标"""
    th = {}
    for g in np.unique(group):
        s = np.sort(scores[group == g])
        idx = int((1 - target_rate) * len(s))
        th[g] = s[min(idx, len(s) - 1)]
    return th

scores = np.array([0.9, 0.7, 0.4, 0.3, 0.8, 0.5, 0.2, 0.1])
group = np.array(["A", "A", "A", "A", "B", "B", "B", "B"])
print("分组阈值:", {k: round(v, 3) for k, v in group_thresholds(scores, group, 0.5).items()})

from fairlearn.postprocessing import ThresholdOptimizer
from sklearn.linear_model import LogisticRegression

base = LogisticRegression(max_iter=1000)
# base.fit(X_train, y_train)
post = ThresholdOptimizer(estimator=base, constraints="equalized_odds",
                          predict_method="predict_proba")
# post.fit(X_train, y_train, sensitive_features=group_train)

from sklearn.feature_selection import mutual_info_classif
import numpy as np

X = np.random.default_rng(0).normal(size=(200, 4))
sensitive = (X[:, 0] + np.random.default_rng(1).normal(0, 0.1, 200) > 0).astype(int)
mi = mutual_info_classif(X, sensitive, random_state=0)
print("各特征与敏感属性的互信息:", mi.round(3))   # 特征 0 最高，是强代理

定义公平目标 → 分组评估基线 → 选择缓解手段 → 上线分组监控
     ↑                                              │
     └────────── 定期复审 + 申诉反馈 ←──────────────┘