联邦学习与隐私保护机器学习：FedAvg、非 IID、DP-SGD 与安全聚合

1. 服务器下发全局模型 w_t 给选中的 K 个客户端
2. 每个客户端用本地数据做 E 轮 SGD，得到 w_t^k
3. 客户端上传更新（或模型）
4. 服务器按样本量加权平均：w_{t+1} = Σ (n_k / n) · w_t^k

import copy
import torch
import torch.nn as nn
from torch.utils.data import DataLoader

def local_train(model, dataloader, epochs, lr):
    """客户端本地训练，返回更新后的 state_dict"""
    local_model = copy.deepcopy(model)
    opt = torch.optim.SGD(local_model.parameters(), lr=lr, momentum=0.9)
    local_model.train()
    for _ in range(epochs):
        for x, y in dataloader:
            opt.zero_grad()
            loss = nn.functional.cross_entropy(local_model(x), y)
            loss.backward()
            opt.step()
    return local_model.state_dict()

def fed_avg(global_state, client_states, client_sizes):
    """按样本量加权平均所有客户端的模型参数"""
    total = sum(client_sizes)
    new_state = copy.deepcopy(global_state)
    for key in global_state.keys():
        new_state[key] = sum(
            cs[key] * (n / total)
            for cs, n in zip(client_states, client_sizes)
        )
    return new_state

# 主循环
for round_t in range(num_rounds):
    selected = sample_clients(clients, fraction=0.1)     # 每轮采样部分客户端
    states, sizes = [], []
    for c in selected:
        sd = local_train(global_model, c.loader, epochs=3, lr=0.01)
        states.append(sd)
        sizes.append(len(c.dataset))
    new_sd = fed_avg(global_model.state_dict(), states, sizes)
    global_model.load_state_dict(new_sd)

IID  ：各方数据分布一致 → 平均即全局最优
非IID：各方分布不同 → 各自跑偏 → 平均后震荡、收敛慢、精度掉

# FedProx：本地损失加入近端项，mu 控制约束强度
def fedprox_local_loss(model, global_params, x, y, mu=0.01):
    ce = nn.functional.cross_entropy(model(x), y)
    prox = sum(
        ((p - gp) ** 2).sum()
        for p, gp in zip(model.parameters(), global_params)
    )
    return ce + (mu / 2) * prox

(ε, δ)-DP：对任意相邻数据集 D、D'，Pr[M(D) ∈ S] ≤ e^ε · Pr[M(D') ∈ S] + δ
ε 越小隐私越强（噪声越大），δ 是失败概率（通常 1e-5）

1. 对每个样本单独算梯度
2. 裁剪到 L2 范数 ≤ C
3. 求和后加 N(0, σ²C²) 噪声，再用加噪梯度更新

import torch

def dp_sgd_step(model, batch, optimizer, clip_norm=1.0, noise_multiplier=1.1):
    """简化版 DP-SGD 单步"""
    x, y = batch
    optimizer.zero_grad()

    # 逐样本梯度 + 裁剪（实际用 torch.func / opacus 实现）
    per_sample_grads = compute_per_sample_grads(model, x, y)
    clipped = []
    for g in per_sample_grads:
        norm = g.norm(2)
        factor = min(1.0, clip_norm / (norm + 1e-6))
        clipped.append(g * factor)

    # 求和 + 加噪
    summed = torch.stack(clipped).sum(dim=0)
    noise = torch.randn_like(summed) * (noise_multiplier * clip_norm)
    summed = summed + noise

    # 把加噪梯度写回参数并更新
    assign_grads(model, summed)
    optimizer.step()

from opacus import PrivacyEngine

model = MyModel()
optimizer = torch.optim.SGD(model.parameters(), lr=0.05)

privacy_engine = PrivacyEngine()
model, optimizer, loader = privacy_engine.make_private(
    module=model,
    optimizer=optimizer,
    data_loader=loader,
    noise_multiplier=1.1,
    max_grad_norm=1.0,
)

for epoch in range(epochs):
    for x, y in loader:
        optimizer.zero_grad()
        loss = nn.functional.cross_entropy(model(x), y)
        loss.backward()
        optimizer.step()
    eps = privacy_engine.get_epsilon(delta=1e-5)   # 实时查看已消耗的 ε
    print(f"epoch {epoch}, ε = {eps:.2f}")

客户端1 上传：g1 + r12
客户端2 上传：g2 - r12
服务器求和  ：(g1 + g2) + (r12 - r12) = g1 + g2   ← 掩码抵消，只见和

1. 隐私求交（PSI）：双方在不暴露各自 ID 全集的前提下找到共同用户
2. 拆分建模：各持一部分特征的模型，中间层加密交互
3. 梯度加密传输：用同态加密或秘密共享保护中间结果

# 个性化 FL：全局模型 + 本地微调
global_model = receive_global_model()
local_model = copy.deepcopy(global_model)
for epoch in range(finetune_epochs):
    train_one_epoch(local_model, local_loader)   # 只用本地数据微调

1. 增大 batch：噪声摊薄到更多样本，效用↑
2. 预训练 + 微调：先用公开数据预训练，DP 微调步数少 → ε 小
3. 更小的 max_grad_norm + 更多步：有时比大裁剪更好
4. 参数高效微调（LoRA）：只对少量参数加 DP，噪声影响小
5. 分组隐私：对不同敏感度的字段用不同 ε

# Flower：极简的联邦客户端/服务器
import flwr as fl

class FlowerClient(fl.client.NumPyClient):
    def get_parameters(self, config):
        return [v.cpu().numpy() for v in model.state_dict().values()]

    def fit(self, parameters, config):
        set_parameters(model, parameters)
        train(model, train_loader, epochs=1)
        return self.get_parameters(config), len(train_loader.dataset), {}

    def evaluate(self, parameters, config):
        set_parameters(model, parameters)
        loss, acc = test(model, test_loader)
        return float(loss), len(test_loader.dataset), {"accuracy": float(acc)}

fl.client.start_numpy_client(server_address="127.0.0.1:8080", client=FlowerClient())

1. 数据最小化：只传模型更新，不传原始数据
2. 明确告知：用户/机构知情并同意参与
3. 隐私预算记录：每次训练消耗的 ε 要可审计
4. 退出机制：参与方可随时退出，已贡献数据的影响可清除
5. 跨境合规：数据不出境，联邦天然契合
6. 安全评估：上线前做攻击测试（MIA、梯度反演）

架构层：客户端本地训练 + 服务器聚合（FedAvg）
优化层：FedProx / SCAFFOLD 对抗非 IID 漂移
隐私层：DP-SGD（加噪）+ 安全聚合（掩码）
模式层：横向 / 纵向 / 迁移 / 个性化
工程层：Flower / FATE + 通信压缩 + 合规审计