简介：一套运用联邦学习技术实现高校学生成绩预测的完整解决方案，以Python实现并基于Streamlit构建交互式可视化平台。项目面向计算机、信息工程、智能科学等专业在校师生与从业者，可支撑课程实践、综合作业及毕业设计等教学环节，代码规范、架构清晰，适合有一定编程基础的读者二次开发与功能拓展。

压缩包共67个文件，大小约2.26MB，以py源码、pyc编译文件、csv数据与备份为主：py包含核心算法实现，csv为样本数据及训练指标记录，png提供混淆矩阵可视化，README说明整体使用方式。目前已有75人学习查阅。

内容涵盖联邦学习成绩预测核心流程，提供main_local、main_scaffold、main_fedrep等多个入口及FedProx、APFL、MTL等对比方法模型文件，便于复现实验、对比策略效果；作为答辩环节95分的毕业设计作品，这套代码在算法实现、工程组织与可视化呈现上均有较高参考价值。

1. 联邦学习驱动的成绩预测：不共享成绩单也能训练出可用模型

三个学校各自握着几百份学生成绩和课堂行为数据，都想训一个跨校通用的成绩预测模型，但谁都不愿意把数据交给别人——这不仅是隐私问题，更是数据归属和信任成本问题。传统做法把数据汇总到一台服务器集中训练，在真实教务场景里根本走不通。联邦学习把训练搬到数据所在的地方，本地算完只上传模型权重，服务器做聚合后下发更新。这份资源用 Python 手写了 FedAvg 全流程，从模拟多校数据集、特征工程到 PyTorch 训练、Streamlit 可视化平台，覆盖了做毕业设计最关心的完整闭环。适合正在选题目、需要交付可演示系统的本科生，也适合想弄明白联邦学习工程落地细节的开发者。

2. 联邦学习的核心机制：数据不出校门，模型如何变聪明

2.1 为什么成绩数据不适合集中训练

先看一个现实问题：成绩单本身属于敏感个人信息，再加上各学校数据格式不统一、字段含义有出入，直接合并成一个大 CSV 在大多数场景下是不现实的。就算技术上允许，机房存谁的服务器、模型训完归谁所有，这些归属问题也足够让项目在需求阶段就卡住。

联邦学习换了个思路：模型在每所学校本地训练，只把训练产生的模型权重（或者说参数增量）传到聚合服务器。服务器拿到各客户端的权重，按样本量加权合并成新的全局模型，再下发回去。整个过程中原始成绩数据从未离开学校内部网络。代价也很明确——通信开销变大了，训练速度比集中式慢不少，而且各端数据分布如果差异太大，聚合出来的模型可能谁都覆盖不好。

2.2 FedAvg 聚合流程：四步走

FedAvg（联邦平均）是最常用的基线算法，核心流程可以拆成四步：

1. 服务器初始化全局模型参数 w0，分发给参与本轮训练的客户端。
2. 每个客户端用自己的本地数据，在 w0 基础上训练若干轮（比如 3 个 epoch）。
3. 客户端把更新后的权重传回服务器。
4. 服务器按各客户端样本量占比做加权平均，得到新的全局权重。

聚合公式写作：全局权重 = Σ (n_k / N) × w_k，其中 n_k 是第 k 个客户端的样本数，N 是参与客户端总样本数。样本越多的学校，对全局模型的话语权越大。这个设计的直觉很朴素——数据多的一方应该被更信任。

2.3 为什么是 FedAvg 而不是其他联邦方案

联邦学习不只有横向联邦这一种。成绩预测场景里，各学校的学生、特征维度基本一致，只是样本分布不同，属于典型的横向联邦，FedAvg 就是最顺手的起点。纵向联邦（多方持有不同特征、样本重叠）在这里没有应用条件，因为教务处手里那点数据还够不上需要纵向对齐的体量。

另一个常被问到的问题是：为什么不用 XGBoost 而选神经网络？XGBoost 在表格数据上确实很强，但它的模型本质是一堆决策树，做联邦聚合要传树结构，实现成本高、聚合也不平滑。神经网络权重是连续张量，直接做加权平均即可，配合 PyTorch 实现起来非常自然。选型逻辑就一句话：FedAvg + 神经网络，是这个场景下工程成本最低、可解释路径最清晰的组合。

3. 先把数据切成三份：非IID切分与成绩特征工程设计

3.1 用模拟数据复现“三所学校”场景

真实环境里拿不到权限去部署，演示项目就得先造数据。这里要面对的核心问题是：三所学校的数据不能完全一样，否则联邦学习就失去了意义——各客户端数据分布完全一致时，聚合和集中训练几乎没有差别。

常见做法是模拟非IID（非独立同分布）切分：让不同学校的学生成绩基准水平、波动幅度有明显差异。比如学校 A 学生整体成绩偏高，学校 B 偏低，学校 C 成绩离散度大。这个差异直接用正态偏移控制即可。

import numpy as np import pandas as pd from sklearn.model_selection import train_test_split def generate_school_data(school_id, n_samples=400, seed=42): rng = np.random.default_rng(seed + school_id) n_features = 12 # 各学校基础水平不同：模拟非IID数据分布 base_shift = [0.10, 0.0, -0.08][school_id - 1] feature_mean = np.full(n_features, 0.5) + base_shift X = rng.normal(feature_mean, 0.2, (n_samples, n_features)) X = np.clip(X, 0.0, 1.0) # 标签由特征加权组合生成，加噪声避免线性可分过于理想 weights = rng.uniform(0.6, 1.0, n_features) score = np.dot(X, weights) / n_features score += rng.normal(0, 0.12, n_samples) label = (score > 0.5).astype(int) df = pd.DataFrame(X, columns=[f"feature_{i}" for i in range(n_features)]) df["label"] = label return df schools = [generate_school_data(i) for i in [1, 2, 3]]

这份代码里 school_id 直接参与随机种子生成和基础水平偏移，生成三份分布略有差异的数据集。特征数固定为 12，方便后续建模。base_shift 控制学校间的平均成绩差，数值越大该校整体表现越好。clip 操作把特征限制在 [0, 1] 区间，与后面归一化目标对齐。注意 label 的生成用了特征加权求和再和 0.5 作比较，相当于构造了一个隐式的评判阈值。

3.2 特征体系设计：不是只塞成绩单

模拟数据的特征名是占位的，但真实场景里这 12 个特征应该有具体业务含义。常见的成绩预测特征体系可以这样设计：

真实项目里特征工程的重点是把这些行为日志整理成统计特征，比如“过去两周作业提交及时率”比“本学期总提交数”更能反映近期状态变化。本项目模拟数据里直接用 0~1 的连续值表示归一化后的特征，逻辑一致，只是跳过了数据清洗步骤。

3.3 归一化的边界问题

成绩预测的特征量纲差异很大：百分制成绩是 0~100，出勤率是 0~1，线上活跃天数是整数。训练前必须做归一化。这里有个容易被忽略的坑：客户端各自做归一化，还是服务器统一做？

如果三个学校各自用本校数据的 min/max 做归一化，那么特征值为 0.5 在三所学校代表完全不同水平的成绩。正确做法是：服务器保存一份归一化参数（全局 min/max），下发给各客户端，推理阶段也使用同一套参数。这份资源的数据生成阶段已经把特征压到了 [0, 1]，就是为了让这个边界问题不干扰演示主流程。

4. 手写FedAvg：客户端训练、服务器聚合与训练循环

4.1 模型层：一个轻量全连接网络

成绩预测输入是 12 维结构化特征，标签是二分类（及格/不及格），用浅层全连接网络就足够了。模型结构不用复杂，三层、中间层 64 维、Dropout 0.2，是这个规模下比较稳妥的配置。层数叠多了在这个数据量上只会过拟合，不会带来任何精度收益。

import torch import torch.nn as nn class ScorePredictor(nn.Module): def __init__(self, input_dim=12, hidden_dim=64): super().__init__() self.net = nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Dropout(0.2), nn.Linear(hidden_dim, hidden_dim // 2), nn.ReLU(), nn.Linear(hidden_dim // 2, 1), ) def forward(self, x): return self.net(x)

网络结构是明显的“宽进窄出”，前两层保持一定宽度让模型学到特征组合，最后一层压缩到单个 logit。配合后面的 BCEWithLogitsLoss 使用，不需要在最后一层额外套 sigmoid——损失函数内部已经做了数值稳定处理。

4.2 客户端训练类：本地训练只关心两件事

客户端要做的事情就两件：在本地数据上训练若干轮，然后把模型权重和本地样本量上报。权重用于聚合，样本量用于计算聚合比例，别的数据一概不上传——这是联邦学习的边界所在。

class Client: def __init__(self, client_id, X, y, model, lr=0.01): self.client_id = client_id self.X = torch.FloatTensor(X) self.y = torch.FloatTensor(y).view(-1, 1) self.model = model self.lr = lr self.criterion = nn.BCEWithLogitsLoss() def local_train(self, epochs=3): optimizer = optim.SGD(self.model.parameters(), lr=self.lr) self.model.train() for _ in range(epochs): optimizer.zero_grad() logits = self.model(self.X) loss = self.criterion(logits, self.y) loss.backward() optimizer.step() return { "weight": [p.detach().clone() for p in self.model.parameters()], "n_samples": len(self.X), }

local_train 返回的是权重深拷贝，不是模型引用，这一步很重要——如果不 clone，后面聚合时客户端继续训练会把同一份内存里的数值改掉，聚合结果就成了玄学。还有一点容易被忽略：这里没有启用 mini-batch，直接把整个本地数据集一次性喂进去。样本量在 400 左右时没问题，如果真实数据量上万，需要加 batch 循环。

4.3 服务器聚合：加权平均不是简单求均值

服务器聚合逻辑在代码上反而比客户端简单，但聚合权重必须与样本量严格挂钩。

class FedAvgServer: def __init__(self, model): self.global_model = model self.global_weights = [p.detach().clone() for p in model.parameters()] def aggregate(self, local_updates): total_n = sum(u["n_samples"] for u in local_updates) if total_n == 0: return new_weights = [torch.zeros_like(w) for w in self.global_weights] for update in local_updates: weight = update["weight"] contribution = update["n_samples"] / total_n for i, w in enumerate(weight): new_weights[i] += contribution * w for i, p in enumerate(self.global_model.parameters()): p.data.copy_(new_weights[i])

aggregate 的第一步是统计 total_n，它决定了每个客户端的权重系数。举例来说，某客户端有 400 条样本，另一个只有 100 条，那么前者的权重是 0.8，后者 0.2。如果做简单平均，两个客户端的影响力就一样了，小样本客户端的噪声会被放大。代码最后把聚合结果复制回全局模型参数，注意用的是 copy_ 而不是赋值——copy_ 是原地更新，不会破坏模型内部结构。

4.4 完整训练循环：轮次、客户端抽样与评估

真实联邦场景里，每一轮不一定所有客户端都参与。考虑到网络状况、设备可用性，通常会按比例随机抽样一批客户端参与本轮训练。这部分实现直接决定训练收敛速度。

def run_fedavg(clients, server, rounds=30, client_frac=0.8): history = [] n_selected = max(1, int(len(clients) * client_frac)) for rnd in range(rounds): selected_idx = np.random.choice( len(clients), n_selected, replace=False ) local_updates = [] for idx in selected_idx: update = clients[idx].local_train(epochs=3) local_updates.append(update) server.aggregate(local_updates) # 每轮结束后用全部客户端数据做一次评估 acc = evaluate_global(server.global_model, clients) history.append(acc) print(f"Round {rnd+1:02d} | accuracy: {acc:.4f}") return history

这里的 client_frac 是每轮参与训练的客户端比例，设 0.8 表示 3 个客户端中有 2~3 个参与。做毕业设计演示时这个值别调太低，客户端数量本身少，抽多了等于没有联邦意义，抽少了训练不稳定。epochs=3 是 FedAvg 原论文的经典配置，本地训练轮数不宜过大，否则本地过拟合反而损害全局收敛。

5. 联邦学习成绩预测常见问题：非IID、灾难性遗忘与恶意客户端

5.1 非IID数据导致全局模型偏置

现象：训练完成后全局模型在三个客户端上的准确率差异超过 15 个百分点，对数据基础较好的学校效果好，对基础薄弱的学校效果差。

原因：FedAvg 的加权平均假设各客户端数据分布相似。模拟数据里各学校成绩基准不同，全局模型会偏向样本量大的客户端，小样本端被“带偏”。

解决：最直接的调整是把客户端抽样改成“每轮强制包含所有低质数据端”，保证小样本客户端的训练信号每轮都参与。另一个有效方案是改用 FedProx，在客户端本地损失函数里加一个近端项，约束本地更新不要偏离全局模型太远。这个改动在代码上只需要在 loss.backward 之前加一行：loss += (mu / 2) * sum(torch.norm(p - global_p)**2 for p, global_p in zip(model.parameters(), global_weights))，mu 一般取 0.01~0.1。

5.2 本地模型灾难性遗忘

现象：训练到第 10 轮以后，全局模型在部分客户端上的表现突然掉了一截，恢复不回来。

原因：联邦学习里的灾难性遗忘有两种表现。一种是客户端本地连续多轮训练，把前一轮学到的全局知识覆盖了；另一种是全局聚合阶段，新一伦次模型对某些客户端的特征分布产生了“失忆”。本地 epoch 设得越大，这种现象越明显。

解决：把本地 epoch 从默认的 3 下调到 1 试试，这是零成本先说结论的方案。如果调低 epoch 后收敛速度太慢，再考虑在客户端损失函数里加正则约束（EWC 或 L2 惩罚）。实际排查顺序建议是先调 epoch，确认是遗忘问题再动损失函数，不要一上来就加正则。

5.3 恶意客户端投毒与后门攻击

现象：训练过程 Loss 曲线正常，但某个特定输入（例如某作弊模式的特征组合）被持续预测为及格，其他输入预测完全正常。

原因：某个客户端在上传权重时做了手脚，比如在本地数据里插入包含特定特征组合的伪造样本，训出一个“后门”。聚合时恶意权重被混进了全局模型。这是联邦学习比集中式训练更容易遭受的攻击面，因为服务器看不到客户端数据。

解决：在服务器侧加一道“权重审查”逻辑——每轮聚合前计算各客户端上传权重与全局权重的欧氏距离，距离超过平均值 2 个标准差的直接剔除。这个启发式过滤不能防住所有攻击，但能挡住最粗劣的投毒手段。

5.4 归一化参数不一致

现象：训练收敛很好，但推理阶段预测结果明显偏移，同一名学生手工特征计算出的结果和平台展示的结果对不上。

原因：训练时各客户端按自己的统计量做了归一化，推理时加载的特征处于另一套尺度。这个问题在真实部署中比想象中普遍——各学校的数据管道由不同小组维护。

解决：把归一化参数作为全局状态的一部分，随模型一并下发。Flower 框架里可以直接用参数状态机制保存 min/max 向量，手写项目则需要在 FedAvgServer 里维护一份全局归一化字典，每轮训练和推理都用它。

5.5 超参数随机性导致结果不可复现

现象：相同代码跑两次，训练曲线差不少，有时最终准确率相差 3~5 个点。

原因：没固定随机种子。NumPy、PyTorch、Python 内置 random 三个模块各自维护独立的随机状态，只固定一个等于没固定。

解决：训练入口固定三行代码，同时把每轮客户端抽样索引列表记录到日志文件，复现时用同一份抽样序列。

np.random.seed(42) random.seed(42) torch.manual_seed(42)

6. Streamlit可视化平台：从 model.pt 到可交互预测

6.1 加载全局模型与推理链路

训练结束后把全局模型保存为 model.pt，Streamlit 端加载它做推理。这里的关键点是加载时要用与训练完全相同的预处理管道——不是“差不多”，是必须严格一致。

st.title("联邦学习学生成绩预测平台") with st.sidebar: school_id = st.selectbox("选择模拟学校", [1, 2, 3]) feature_values = [ st.slider(f"特征 {i}", 0.0, 1.0, 0.5) for i in range(12) ] if st.button("开始预测"): model = ScorePredictor() model.load_state_dict(torch.load("model.pt")) model.eval() with torch.no_grad(): x = torch.FloatTensor([feature_values]) prob = torch.sigmoid(model(x)).item() st.metric("预测及格概率", f"{prob:.2%}")

这个页面结构里，滑块模拟的是归一化后的特征输入。真实部署时这些滑块应该是表单字段，比如出勤率、作业提交及时率，由后端做归一化转换。st.metric 适合展示单个核心结果，如果要做更细的分析，可以加一行st.bar_chart展示各学校预测结果对比。

6.2 训练过程可视化与模型血缘

Streamlit 的价值不只是把模型包装成一个按钮，它还可以承载训练过程的完整回放。建议在页面上加一个训练历史视图：读取训练时保存的 history.json，画出每轮准确率曲线，把三个客户端各自的数据量、参与轮次用表格展示出来。这样答辩的时候，评审老师看到的不是一个黑匣子，而是从数据切分、联邦训练到最终预测的完整链路。

6.3 我的固定习惯：每轮实验强制走一遍验证流程

做这个项目我踩过最大的坑是：训练脚本改了数据生成逻辑，忘了重新训练模型，结果 Streamlit 页面加载的还是旧权重。从那以后我每次上线前都强制走一遍完整流程——固定随机种子、清空缓存、跑 3 次训练取中位数准确率，核对模型文件修改时间，再启动页面。这套流程看着繁琐，但一个命令行就能写成一个脚本，训练 → 评估 → 比对 → 启动页面一步到位。希望帮到你。

本文还有配套的精品资源，点击获取