# 检测偏见
$ python
>>> import numpy as np
>>> import pandas as pd
>>> from sklearn.datasets import fetch_openml
>>> from sklearn.model_selection import train_test_split
>>> from sklearn.linear_model import LogisticRegression
>>> from fairlearn.metrics import demographic_parity_difference
>>> from fairlearn.reductions import GridSearch, DemographicParity

# 加载数据
X, y = fetch_openml(data_id=1590, as_frame=True, return_X_y=True)
y = (y == ‘>50K’).astype(int)

# 分割数据
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 训练模型
model = LogisticRegression()
model.fit(X_train, y_train)

# 检测偏见
y_pred = model.predict(X_test)
sex = X_test[‘sex’]
dp_diff = demographic_parity_difference(y_test, y_pred, sensitive_features=sex)
print(f”Demographic parity difference: {dp_diff:.4f}”)

# 缓解偏见
constraint = DemographicParity()
mitigator = GridSearch(model, constraint=constraint, grid_size=10)
mitigator.fit(X_train, y_train, sensitive_features=X_train[‘sex’])

# 评估缓解后的模型
y_pred_mitigated = mitigator.predict(X_test)
dp_diff_mitigated = demographic_parity_difference(y_test, y_pred_mitigated, sensitive_features=sex)
print(f”Mitigated demographic parity difference: {dp_diff_mitigated:.4f}”)

# 联邦学习示例
$ python
>>> import torch
>>> import syft as sy
>>> from torch import nn, optim

# 初始化虚拟工作者
hook = sy.TorchHook(torch)
worker1 = sy.VirtualWorker(hook, id=”worker1″)
worker2 = sy.VirtualWorker(hook, id=”worker2″)

# 创建虚拟数据
data1 = torch.tensor([[1., 2.], [3., 4.]]).send(worker1)
target1 = torch.tensor([[0.], [1.]]).send(worker1)

data2 = torch.tensor([[5., 6.], [7., 8.]]).send(worker2)
target2 = torch.tensor([[1.], [0.]]).send(worker2)

# 定义模型
model = nn.Linear(2, 1)

# 训练模型
optimizer = optim.SGD(model.parameters(), lr=0.1)

for epoch in range(10):
# 在worker1上训练
model.send(worker1)
optimizer.zero_grad()
pred = model(data1)
loss = ((pred – target1) ** 2).sum()
loss.backward()
optimizer.step()
model.get()

# 在worker2上训练
model.send(worker2)
optimizer.zero_grad()
pred = model(data2)
loss = ((pred – target2) ** 2).sum()
loss.backward()
optimizer.step()
model.get()

print(f”Epoch {epoch+1}, Loss: {loss.get().item():.4f}”)

# 模型解释示例
$ python
>>> import shap
>>> from sklearn.datasets import load_iris
>>> from sklearn.ensemble import RandomForestClassifier

# 加载数据并训练模型
X, y = load_iris(return_X_y=True)
model = RandomForestClassifier()
model.fit(X, y)

# 创建SHAP解释器
explainer = shap.TreeExplainer(model)
shap_values = explainer(X)

# 生成摘要图
shap.summary_plot(shap_values, X, feature_names=[‘sepal length’, ‘sepal width’, ‘petal length’, ‘petal width’])

# 生成依赖图
shap.dependence_plot(0, shap_values, X, feature_names=[‘sepal length’, ‘sepal width’, ‘petal length’, ‘petal width’])