深度学习学习笔记：按个体划分数据，避免评估被信息泄漏污染

一个模型获得很高的测试分数，不一定说明它学会了希望学习的规律。它也可能认出了同一个人的重复记录，利用了未来信息，或者通过预处理提前看到了测试集。

在生物数据中，“一行”经常不是一个独立个体。一个人可以贡献多个时间点，一个供体可以贡献很多细胞，一个样本可以拆成许多片段。划分数据前，需要先回答：以后模型面对的新对象是什么？

这里选择的任务是泛化到未见过的个体，所以按照个体 ID 分组。示例全部使用人工生成的数字，不涉及真实受试者，不给出任何疾病预测结论；模型分数只用于检查程序能否运行。

1. 从评价对象倒推划分单位

如果未来要预测新供体，同一供体的记录不应同时出现在训练集和测试集。如果要评估未来时间点，随机混合过去与未来可能不合适。如果要推广到另一家中心，还需要按中心或外部队列进行额外评估。

按个体分组解决的是一种依赖结构，不会自动解决时间泄漏、批次混杂、标签错误和中心差异。一个 ID 列也不能替代研究设计。

下面用 GroupShuffleSplit，它按组生成索引；参数比例对应组的比例，而不是保证相同比例的记录数。组大小不等时，两者会不同。它也不保证标签分层，所以划分后仍要检查每一部分的类别分布。见 scikit-learn 的 GroupShuffleSplit 文档。

2. 安装与运行边界

示例需要 Python、NumPy、scikit-learn 与 PyTorch，使用 CPU 和小张量，不下载数据集。下面的安装命令在自己新建的空练习目录中执行；若当前目录已有 .venv，不要直接覆盖它。如果已有兼容环境，也不需要重新安装。

1
2
3
python -m venv .venv
source .venv/bin/activate
python -m pip install numpy scikit-learn torch

这是 Linux/WSL 的激活写法。PyTorch 的具体安装方式会随平台变化，本文不要求 CUDA；记录你实际安装的版本，不把“相同随机种子”理解为跨设备、跨版本逐位相同。

将下一段保存为 grouped_training.py，执行 python grouped_training.py。代码只在内存中使用模拟数据，不覆盖磁盘上的模型或数据文件。

3. 完整代码：先分组，再拟合预处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import copy
import numpy as np
import sklearn
import torch
from torch import nn
from torch.utils.data import DataLoader, TensorDataset
from sklearn.model_selection import GroupShuffleSplit
from sklearn.preprocessing import StandardScaler

seed = 17
rng = np.random.default_rng(seed)
torch.manual_seed(seed)

# 24 synthetic people; each contributes 3 records with 6 features.
groups = np.repeat(np.arange(24), 3)
y = (groups % 2).astype(np.int64)
X = rng.normal(size=(len(groups), 6)).astype(np.float32)
X[:, 0] += y * 0.8  # An artificial signal, not a biological relationship.

outer = GroupShuffleSplit(n_splits=1, test_size=0.25, random_state=seed)
development, test = next(outer.split(X, y, groups))
inner = GroupShuffleSplit(n_splits=1, test_size=1 / 3, random_state=seed + 1)
train_local, valid_local = next(
    inner.split(X[development], y[development], groups[development])
)
train, valid = development[train_local], development[valid_local]

splits = {"train": train, "valid": valid, "test": test}
group_sets = {name: set(groups[index]) for name, index in splits.items()}
assert not (group_sets["train"] & group_sets["valid"])
assert not (group_sets["train"] & group_sets["test"])
assert not (group_sets["valid"] & group_sets["test"])
for name, index in splits.items():
    labels = np.bincount(y[index], minlength=2).tolist()
    print(name, "groups=", len(group_sets[name]), "rows=", len(index), "labels=", labels)
    if min(labels) == 0:
        raise ValueError("A split has only one class; revise the evaluation design.")
print("PASS: no person occurs in more than one split")

# Only training records estimate the scaler's parameters.
scaler = StandardScaler().fit(X[train])
scaled = {name: scaler.transform(X[index]).astype(np.float32)
          for name, index in splits.items()}
assert np.allclose(scaled["train"].mean(axis=0), 0, atol=1e-5)

train_data = TensorDataset(torch.from_numpy(scaled["train"]), torch.from_numpy(y[train]))
loader = DataLoader(train_data, batch_size=12, shuffle=True,
                    generator=torch.Generator().manual_seed(seed), num_workers=0)
model = nn.Sequential(nn.Linear(6, 16), nn.ReLU(), nn.Dropout(0.2), nn.Linear(16, 2))
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
criterion = nn.CrossEntropyLoss()

def evaluate(name):
    model.eval()
    features = torch.from_numpy(scaled[name])
    labels = torch.from_numpy(y[splits[name]])
    with torch.no_grad():
        logits = model(features)
        loss = criterion(logits, labels).item()
        accuracy = (logits.argmax(dim=1) == labels).float().mean().item()
    return loss, accuracy

best_loss, best_state, best_epoch = float("inf"), None, None
for epoch in range(1, 41):
    model.train()
    for features, labels in loader:
        optimizer.zero_grad()
        loss = criterion(model(features), labels)
        loss.backward()
        optimizer.step()
    valid_loss, _ = evaluate("valid")
    if valid_loss < best_loss:
        best_loss = valid_loss
        best_state = copy.deepcopy(model.state_dict())
        best_epoch = epoch

assert best_state is not None
model.load_state_dict(best_state)
test_loss, test_accuracy = evaluate("test")  # Final test evaluation: once.
print("selected_epoch=", best_epoch)
print(f"test_loss={test_loss:.4f} test_accuracy={test_accuracy:.4f}")
print("versions:", "numpy", np.__version__, "sklearn", sklearn.__version__, "torch", torch.__version__)

此处标准化只是演示连续特征的预处理边界，不是建议把所有 RNA-seq 计数直接交给这个网络。真实数据的表示方式、批次处理及任务标签需要单独确定。

4. 应该检查什么输出

这段数据设计固定了每个个体三行，所以三部分的规模应是：训练 12 个个体、36 行；验证 6 个个体、18 行；测试 6 个个体、18 行。还应打印 PASS: no person occurs in more than one split。

类别计数、选中轮次、测试损失与准确率由实际运行产生。本文不预设一个漂亮分数，也不把人工数据的表现解释为真实任务中的性能。改变平台或依赖版本后，训练结果可能变化；分组不重叠和只在训练集拟合标准化的原则不应改变。

state_dict() 中的张量可能继续随训练变化，因此这里对最佳状态做深拷贝，而不是仅保存一个引用。验证集负责选轮次，最终测试集没有进入这条选择过程。

评估时同时使用 model.eval() 和不计算梯度的上下文：前者调整 Dropout 等模块行为，后者避免构建训练用的梯度图。训练模式与评估模式的基本用法见 PyTorch 的参数优化教程。

5. 泄漏不只发生在划分这一行

先对全部样本计算均值、筛选特征，再拆训练集，会让测试数据影响处理规则。正确顺序是先划分，在训练数据上拟合，再把同一个变换应用于验证与测试；交叉验证中的每一折也必须重新拟合。参见 scikit-learn 对数据泄漏的说明。

同样危险的操作还有：用结局发生后的变量预测结局；将同一患者的记录通过不同 ID 引入不同集合；反复查看测试集后挑学习率；在全部数据上做有监督特征筛选；把数据增强产生的近重复样本分到两边。

这里只计算记录级准确率。一个个体有多条记录时，不应把这些记录当成相互独立的个体来估计不确定性。是否先聚合个体预测、如何按个体重采样，以及需要什么指标，取决于应用目标和研究设计。

准确率也不是唯一尺度。类别不平衡时，需要查看混淆矩阵、各类召回等；若使用概率，则还应考虑概率校准。阈值的选择需要发生在开发阶段，不能观察最终测试标签后再调整。

6. 练习：审计流程，不追求更高的数字

第一项练习是将三个集合的个体 ID 打印出来，手工核对交集为空。第二项是改成每人不同数量的记录，观察“组比例”和“行比例”如何分离。

第三项是检查一份自己的模型笔记：哪些步骤调用了 fit？这些步骤看到的是哪个集合？如果更改预处理、模型结构或训练轮次，是否又查看了测试集？把答案写成表格，比只记录最后一个分数更能说明实验边界。

不要为了让测试分数上升，反复随机划分直到遇到好看的结果。若需要评估划分敏感性，应预先确定重复划分或分组交叉验证方案，并按方案完整报告，而不是筛选最好的一次。

官方参考

继续阅读与公开资源

PyTorch 学习教程

官方教程，包含可运行例子与 Notebook 入口

打开公开资源 ↗