64. 联邦学习与隐私计算在生物医学¶

传统方式（集中学习）：
  医院A的数据 ──┐
  医院B的数据 ──┼──→ 集中到一个中心服务器 ──→ 训练模型
  医院C的数据 ──┘
  问题：数据要搬家！隐私风险大！法规不允许！

联邦学习方式（数据不动，模型动）：
  医院A：本地训练 → 上传模型参数更新 ──┐
  医院B：本地训练 → 上传模型参数更新 ──┼──→ 中心聚合参数 → 下发新模型
  医院C：本地训练 → 上传模型参数更新 ──┘
  优势：数据从不离开医院！只有"学到的知识"在流动！

第1轮：
  1. 中心服务器发送初始模型给所有参与方
  2. 每个参与方用自己的数据训练几个epoch
  3. 每个参与方把模型更新（梯度/参数差）发给中心
  4. 中心聚合所有更新（通常用平均：FedAvg算法）
  5. 中心把聚合后的模型发回给所有参与方

第2轮：重复步骤2-5...
  ...
第N轮：模型收敛，大家都得到一个好模型

问题：
  - 单家医院的样本量太小（罕见病可能只有几十例）
  - AI模型需要大数据才能训好
  - 但法规禁止直接共享患者数据

解决：
  联邦学习 = 不共享数据，只共享模型知识
  → 等效于用几万例数据训练，但数据从不集中

特征维度相同，样本不同
白话：大家有相同的体检项目，但检查的人不一样

医院A：1000个患者 × [年龄, 血压, 血糖, 基因表达...]
医院B：2000个患者 × [年龄, 血压, 血糖, 基因表达...]
医院C：1500个患者 × [年龄, 血压, 血糖, 基因表达...]

应用：多中心临床研究（相同指标，不同患者群）

样本相同，特征维度不同
白话：同一批人，在不同地方做了不同检查

基因组中心：500个患者 × [基因变异数据]
医院：      500个患者 × [临床诊断、用药记录]
体检中心：  500个患者 × [体检指标、影像数据]

应用：整合同一批患者在不同机构的多维数据

样本和特征都不完全重叠
白话：不同医院有不同的患者做了不同的检查，但想共同训练一个模型

适用：参与方之间数据差异很大但任务相似
例如：A医院有大量肺癌CT数据，B医院有少量肺癌CT但有大量病理文本
通过迁移学习，B可以利用A学到的影像特征

# ============================================================
# Flower联邦学习示例：多中心基因表达数据联合训练
# pip install flwr torch scikit-learn
# ============================================================

import flwr as fl                   # Flower框架
import torch                        # PyTorch
import torch.nn as nn               # 神经网络模块
from torch.utils.data import DataLoader, TensorDataset
import numpy as np

# ---------- 1. 定义模型 ----------
class GeneExpressionClassifier(nn.Module):
    """基因表达数据分类器（例：预测2型糖尿病风险）"""
    def __init__(self, n_genes=1000, n_classes=2):
        super().__init__()
        self.network = nn.Sequential(
            nn.Linear(n_genes, 256),    # 输入：1000个基因的表达值
            nn.ReLU(),                  # 激活函数
            nn.Dropout(0.3),            # 防止过拟合
            nn.Linear(256, 64),         # 隐藏层
            nn.ReLU(),
            nn.Linear(64, n_classes)    # 输出：2分类（患病/健康）
        )

    def forward(self, x):
        return self.network(x)

# ---------- 2. 定义Flower客户端 ----------
class BiomedicalClient(fl.client.NumPyClient):
    """每个医院运行一个Client"""

    def __init__(self, model, train_loader, test_loader):
        self.model = model
        self.train_loader = train_loader
        self.test_loader = test_loader
        self.criterion = nn.CrossEntropyLoss()
        self.optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

    def get_parameters(self, config):
        """把本地模型参数打包发给服务器"""
        return [val.cpu().numpy() for val in self.model.state_dict().values()]

    def set_parameters(self, parameters):
        """接收服务器下发的全局模型参数"""
        params_dict = zip(self.model.state_dict().keys(), parameters)
        state_dict = {k: torch.tensor(v) for k, v in params_dict}
        self.model.load_state_dict(state_dict, strict=True)

    def fit(self, parameters, config):
        """本地训练（数据不出本机）"""
        self.set_parameters(parameters)  # 加载全局模型

        self.model.train()
        for epoch in range(5):  # 本地训练5个epoch
            for X_batch, y_batch in self.train_loader:
                self.optimizer.zero_grad()
                outputs = self.model(X_batch)
                loss = self.criterion(outputs, y_batch)
                loss.backward()
                self.optimizer.step()

        # 返回更新后的参数和训练样本数
        return self.get_parameters(config={}), len(self.train_loader.dataset), {}

    def evaluate(self, parameters, config):
        """本地评估"""
        self.set_parameters(parameters)
        self.model.eval()

        correct, total, total_loss = 0, 0, 0.0
        with torch.no_grad():
            for X_batch, y_batch in self.test_loader:
                outputs = self.model(X_batch)
                total_loss += self.criterion(outputs, y_batch).item()
                _, predicted = torch.max(outputs, 1)
                total += y_batch.size(0)
                correct += (predicted == y_batch).sum().item()

        accuracy = correct / total
        return total_loss / len(self.test_loader), total, {"accuracy": accuracy}

# ---------- 3. 模拟多中心数据 ----------
def create_hospital_data(n_samples=500, n_genes=1000):
    """模拟一个医院的基因表达数据"""
    X = np.random.randn(n_samples, n_genes).astype(np.float32)  # 基因表达矩阵
    y = (X[:, :10].sum(axis=1) > 0).astype(np.int64)           # 模拟标签

    dataset = TensorDataset(torch.tensor(X), torch.tensor(y))
    train_size = int(0.8 * n_samples)
    train_set, test_set = torch.utils.data.random_split(
        dataset, [train_size, n_samples - train_size]
    )
    return DataLoader(train_set, batch_size=32), DataLoader(test_set, batch_size=32)

# ---------- 4. 启动联邦学习 ----------
if __name__ == "__main__":
    # 作为客户端运行（每个医院运行这段代码）
    model = GeneExpressionClassifier()
    train_loader, test_loader = create_hospital_data()
    client = BiomedicalClient(model, train_loader, test_loader)

    # 连接到联邦学习服务器
    fl.client.start_numpy_client(
        server_address="127.0.0.1:8080",  # 服务器地址
        client=client
    )

# 服务器端代码（聚合中心运行）
import flwr as fl

# 使用FedAvg策略聚合
strategy = fl.server.strategy.FedAvg(
    min_fit_clients=3,        # 至少3个客户端参与训练
    min_evaluate_clients=3,   # 至少3个客户端参与评估
    min_available_clients=3,  # 至少3个客户端连接后才开始
)

# 启动服务器
fl.server.start_server(
    server_address="0.0.0.0:8080",
    config=fl.server.ServerConfig(num_rounds=10),  # 10轮联邦训练
    strategy=strategy
)

差分隐私 = 在传出去的数据/参数上故意加点"噪音"

类比：
  你的真实工资是 10000元
  加了差分隐私噪音后告诉别人："我工资在8000-12000之间"
  → 别人得到了有用的统计信息（大概万元左右）
  → 但无法精确知道你的真实工资

在联邦学习中：
  真实梯度更新：[0.5, -0.3, 0.8, ...]
  加噪音后传出：[0.52, -0.28, 0.79, ...]  ← 加了高斯噪音
  → 中心还是能聚合出有意义的全局更新
  → 但攻击者无法从噪音中反推出原始训练数据

ε（隐私预算）：
  - ε越小 → 噪音越大 → 隐私保护越强 → 但模型精度下降
  - ε越大 → 噪音越小 → 隐私保护越弱 → 模型精度保持

  通常取值：
  - ε=1 → 强隐私保护
  - ε=5-10 → 中等保护（实际常用）
  - ε=100 → 几乎没有保护

  白话：ε是隐私和精度之间的"旋钮"，你需要根据场景调节

# Flower内置差分隐私支持
from flwr.server.strategy import (
    DifferentialPrivacyClientSideFixedClipping,
    FedAvg
)

# 在FedAvg基础上添加客户端差分隐私
base_strategy = FedAvg(min_fit_clients=3)
dp_strategy = DifferentialPrivacyClientSideFixedClipping(
    strategy=base_strategy,
    noise_multiplier=1.0,   # 噪音倍数
    clipping_norm=1.0,      # 梯度裁剪范数
    num_sampled_clients=3   # 每轮采样客户端数
)

同态加密 = 数据加密后还能计算

普通加密：
  明文 → 加密 → 密文（不能计算）→ 解密 → 明文 → 计算

同态加密：
  明文 → 加密 → 密文（可以直接计算！）→ 解密 → 计算结果

类比：
  想象一个魔法手套箱：
  - 你把两个数字放进密封箱子里（加密）
  - 别人戴着手套箱的手套做加法运算（密文计算）
  - 你打开箱子取出结果（解密）
  - 别人全程看不到数字是什么，但帮你算出了结果！

传统联邦学习：
  客户端发送明文梯度 → 服务器聚合 → 服务器能看到每个客户端的梯度

同态加密增强：
  客户端加密梯度 → 发送密文 → 服务器在密文上聚合 → 客户端解密结果
  → 服务器全程看不到真实梯度！更安全！

场景：10家医院联合训练糖尿病预测模型
  - 每家医院有几百到几千例患者数据
  - 单独训练：数据太少，模型泛化差
  - 集中训练：法规禁止数据出院
  - 联邦学习：等效于用几万例训练，数据不出院

实际效果（文献报道）：
  - 单中心模型 AUC: 0.75-0.82
  - 联邦模型 AUC: 0.85-0.91（接近集中训练的0.92）

场景：多个基因组中心联合做GWAS（全基因组关联分析）
  - 基因组数据是最敏感的个人信息（可唯一标识）
  - 稀有变异需要大样本量才能检测到显著关联
  - 联邦GWAS：各中心本地计算统计量，只共享汇总统计结果

工具：
  - PLINK + 联邦学习框架
  - sPLINK（安全PLINK，已有开源实现）

场景：制药公司之间合作筛选药物靶点
  - 每家公司都有自己的化合物库和筛选数据（商业机密）
  - 联合训练分子性质预测模型
  - 纵向联邦：A公司有化合物结构，B公司有临床效果

实际项目：
  - MELLODDY项目：10家制药巨头用联邦学习联合训练药物发现模型
  - 参与者：Amgen, AstraZeneca, Novartis等

场景：多家医院联合训练肿瘤检测AI
  - 影像数据量大，传输成本高
  - 不同医院的设备型号不同（域偏移）
  - 联邦学习 + 域适应技术 = 解决设备差异

案例：
  - FeTS（联邦肿瘤分割）：6大洲30+机构联合训练脑肿瘤分割模型
  - Intel + UPenn的联邦学习医学影像项目

你的需求是什么？
├── 学术研究/快速原型 → Flower（简单、文档好、框架无关）
├── 极致隐私要求 → PySyft（专注差分隐私和安全计算）
├── 企业级部署 → FATE（工业级、支持多种FL类型）
└── PyTorch生态 → Flower 或 PySyft（都与PyTorch集成好）