# ============================================================ # 第一部分:导入依赖 # ============================================================ import torch # PyTorch 核心库 import torch.nn as nn # 神经网络层、损失函数 import torch.optim as optim # 优化器(Adam、SGD 等) from torch.utils.data import DataLoader, TensorDataset # 数据加载工具 from sklearn.datasets import make_classification # 生成模拟二分类数据 from sklearn.model_selection import train_test_split # 划分训练/测试集 from sklearn.preprocessing import StandardScaler # 标准化 from sklearn.metrics import accuracy_score, roc_auc_score # 评估指标 import numpy as np # 数值计算 # ============================================================ # 第二部分:数据准备 # ============================================================ # 2.1 生成模拟二分类数据 # n_samples=5000:5000 个样本 # n_features=20:20 个特征 # n_informative=10:其中 10 个是真正有信息量的 # n_redundant=5:5 个是冗余特征(由信息特征线性组合而成) # weights=[0.9, 0.1]:类别比例 9:1,故意制造类别不平衡 # random_state=42:随机种子,保证每次运行结果一致 X, y = make_classification( n_samples=5000, n_features=20, n_informative=10, n_redundant=5, weights=[0.9, 0.1], random_state=42 ) print(f"数据形状: X={X.shape}, y={y.shape}") print(f"类别分布: {np.bincount(y)}") # 输出如 [4500 500] # 2.2 划分训练集和测试集 # test_size=0.2:测试集占 20%,训练集占 80% # stratify=y:分层采样,保证训练/测试集的类别比例一致 # 不平衡数据必须用 stratify,否则测试集可能几乎没有正样本 X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) print(f"训练集: {X_train.shape}, 测试集: {X_test.shape}") # 2.3 标准化 # fit_transform 在训练集上:计算均值和标准差,并转换 # transform 在测试集上:用训练集的均值和标准差转换,不重新计算 # 为什么要分开?防止数据泄露(测试集统计信息不能参与训练) scaler = StandardScaler() X_train = scaler.fit_transform(X_train) X_test = scaler.transform(X_test) # 2.4 转成 PyTorch 张量 # FloatTensor:PyTorch 默认用 float32,不能用 float64 # reshape(-1, 1):把 y 从 (4000,) 变成 (4000, 1) # 为什么?模型输出是 (batch_size, 1),标签也必须形状一致 X_train_t = torch.FloatTensor(X_train) y_train_t = torch.FloatTensor(y_train).reshape(-1, 1) X_test_t = torch.FloatTensor(X_test) y_test_t = torch.FloatTensor(y_test).reshape(-1, 1) # 2.5 构造 DataLoader # TensorDataset:把 X 和 y 打包,每个样本是 (x_i, y_i) # DataLoader: # batch_size=64:每批 64 个样本 # shuffle=True:每个 epoch 打乱顺序,防止模型学到样本顺序 train_dataset = TensorDataset(X_train_t, y_train_t) train_loader = DataLoader( train_dataset, batch_size=64, shuffle=True ) # ============================================================ # 第三部分:定义模型 # ============================================================ class DNNBinaryClassifier(nn.Module): """ DNN 二分类模型 结构:输入层 → 隐藏层1 → 隐藏层2 → 隐藏层3 → 输出层 输出层不加 Sigmoid,配合 BCEWithLogitsLoss 使用 """ def __init__(self, input_dim): super().__init__() # 调用父类 nn.Module 的初始化,必须写 self.net = nn.Sequential( # 第一层:input_dim → 128 nn.Linear(input_dim, 128), # 全连接层,权重矩阵 (128, input_dim) nn.BatchNorm1d(128), # 批归一化,对每个 batch 的 128 维做归一化 nn.ReLU(), # 激活函数 max(0, x),引入非线性 nn.Dropout(0.3), # 随机丢弃 30% 神经元,防止过拟合 # 第二层:128 → 64 nn.Linear(128, 64), nn.BatchNorm1d(64), nn.ReLU(), nn.Dropout(0.3), # 第三层:64 → 32 nn.Linear(64, 32), nn.ReLU(), # 接近输出层,不再加 BatchNorm 和 Dropout # 输出层:32 → 1 nn.Linear(32, 1) # 输出 1 个值(logit),不加 Sigmoid ) def forward(self, x): # 前向传播:数据依次经过每一层 return self.net(x) # ============================================================ # 第四部分:训练配置 # ============================================================ # 4.1 选择设备 # 有 GPU 用 CUDA,没有就用 CPU device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') print(f"使用设备: {device}") # 4.2 初始化模型,并移到设备 model = DNNBinaryClassifier(input_dim=X_train.shape[1]).to(device) # 4.3 计算 pos_weight,处理类别不平衡 # pos_weight = 负样本数 / 正样本数 # 本例中约等于 9,表示正样本的损失权重是负样本的 9 倍 # 作用:让模型更关注少数类(正样本) pos_weight = torch.tensor( [(y_train == 0).sum() / (y_train == 1).sum()] ).to(device) print(f"pos_weight: {pos_weight.item():.4f}") # 4.4 损失函数 # BCEWithLogitsLoss:二元交叉熵 + Sigmoid,数值稳定 # 输入是 logits(未经过 Sigmoid 的输出),内部会自动做 Sigmoid # pos_weight 用于类别不平衡 criterion = nn.BCEWithLogitsLoss(pos_weight=pos_weight) # 4.5 优化器 # Adam:自适应学习率,收敛快 # lr=1e-3:学习率 0.001 # weight_decay=1e-5:L2 正则化,防止过拟合 optimizer = optim.Adam( model.parameters(), lr=1e-3, weight_decay=1e-5 ) # 4.6 学习率调度器(可选) # 每 10 个 epoch,学习率乘 0.5 scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.5) # ============================================================ # 第五部分:训练循环 # ============================================================ n_epochs = 20 # 训练 20 轮 for epoch in range(n_epochs): # 5.1 设为训练模式 # 启用 Dropout,BatchNorm 用当前 batch 统计量 model.train() total_loss = 0 # 累计一个 epoch 的损失 # 5.2 遍历 DataLoader for X_batch, y_batch in train_loader: # X_batch.shape = (64, 20) # y_batch.shape = (64, 1) # 5.3 把数据移到设备(GPU/CPU) X_batch = X_batch.to(device) y_batch = y_batch.to(device) # 5.4 清空梯度 # PyTorch 默认累加梯度,不清空会导致更新错误 optimizer.zero_grad() # 5.5 前向传播 # 输出 logits,形状 (64, 1) logits = model(X_batch) # 5.6 计算损失 loss = criterion(logits, y_batch) # 5.7 反向传播 # 计算所有参数的梯度 loss.backward() # 5.8 更新参数 # w = w - lr * grad optimizer.step() # 5.9 累计损失 total_loss += loss.item() # 5.10 更新学习率 scheduler.step() # 5.11 打印每个 epoch 的平均损失 avg_loss = total_loss / len(train_loader) current_lr = optimizer.param_groups[0]['lr'] print(f"Epoch {epoch+1}/{n_epochs}, Loss: {avg_loss:.4f}, LR: {current_lr:.6f}") # ============================================================ # 第六部分:评估 # ============================================================ # 6.1 设为评估模式 # 关闭 Dropout,BatchNorm 用训练时学到的全局统计量 model.eval() # 6.2 关闭梯度计算 # 评估不需要反向传播,省内存、加速 with torch.no_grad(): # 6.3 前向传播,得到 logits logits = model(X_test_t.to(device)) # 6.4 手动 Sigmoid,得到概率 y_pred_prob = torch.sigmoid(logits).cpu().numpy().ravel() # 6.5 按 0.5 阈值转成硬分类 y_pred = (y_pred_prob > 0.5).astype(int) # 6.6 计算评估指标 acc = accuracy_score(y_test, y_pred) auc = roc_auc_score(y_test, y_pred_prob) print(f"\n测试集评估:") print(f"准确率: {acc:.4f}") print(f"AUC: {auc:.4f}")逐部分关键解释
第一部分:导入依赖
| 导入 | 作用 |
|---|---|
torch | PyTorch 核心 |
torch.nn | 网络层、损失函数 |
torch.optim | 优化器 |
DataLoader, TensorDataset | 数据加载 |
make_classification | 生成模拟数据 |
train_test_split | 划分数据集 |
StandardScaler | 标准化 |
accuracy_score, roc_auc_score | 评估指标 |
第二部分:数据准备
关键点:
weights=[0.9, 0.1]:故意制造不平衡,模拟真实场景。stratify=y:保证训练/测试集类别比例一致,否则测试集可能没有正样本。fit_transformvstransform:训练集:
fit_transform,计算统计量 + 转换。测试集:
transform,只用训练集的统计量转换,防止数据泄露。
reshape(-1, 1):把 y 从(4000,)变成(4000, 1),和模型输出形状一致。shuffle=True:每个 epoch 打乱数据,防止模型学到顺序规律。
第三部分:模型定义
关键点:
继承
nn.Module:PyTorch 所有模型的基类。super().__init__():必须写,否则参数无法正确注册。nn.Sequential:把层按顺序串起来。每层结构:
Linear:全连接,做线性变换。BatchNorm1d:批归一化,加速训练、提升稳定性。ReLU:激活函数,引入非线性。Dropout(0.3):随机丢弃 30% 神经元,防止过拟合。
输出层不加 Sigmoid:配合
BCEWithLogitsLoss,数值更稳定。
第四部分:训练配置
关键点:
device:自动检测 GPU,没有就用 CPU。pos_weight:处理类别不平衡,给正样本加权。公式:负样本数 / 正样本数。
本例约 9,正样本损失权重是负样本的 9 倍。
BCEWithLogitsLoss:输入是 logits(未 Sigmoid)。
内部自动做 Sigmoid + 交叉熵,数值稳定。
pos_weight用于不平衡。
Adam:自适应学习率,收敛快。
lr=1e-3:学习率。weight_decay=1e-5:L2 正则。
StepLR:每 10 个 epoch 学习率乘 0.5,帮助收敛。
第五部分:训练循环
每个 batch 的 7 步:
| 步骤 | 代码 | 作用 |
|---|---|---|
| 1 | optimizer.zero_grad() | 清空梯度 |
| 2 | logits = model(X_batch) | 前向传播 |
| 3 | loss = criterion(logits, y_batch) | 计算损失 |
| 4 | loss.backward() | 反向传播,算梯度 |
| 5 | optimizer.step() | 更新参数 |
| 6 | total_loss += loss.item() | 累计损失 |
| 7 | scheduler.step() | 更新学习率(每 epoch) |
为什么zero_grad()必须写?
PyTorch 默认累加梯度,不清空会导致更新错误。
为什么model.train()必须写?
启用 Dropout 和 BatchNorm 的训练行为。
第六部分:评估
关键点:
model.eval():关闭 Dropout,BatchNorm 用全局统计量。torch.no_grad():关闭梯度计算,省内存、加速。torch.sigmoid(logits):手动 Sigmoid,把 logits 转成概率。(y_pred_prob > 0.5):按 0.5 阈值硬分类。accuracy_score+roc_auc_score:不平衡时,准确率会虚高。
AUC 更能反映真实效果。
运行结果示例
text
数据形状: X=(5000, 20), y=(5000,) 类别分布: [4502 498] 训练集: (4000, 20), 测试集: (1000, 20) 使用设备: cpu pos_weight: 9.0361 Epoch 1/20, Loss: 0.5234, LR: 0.001000 Epoch 2/20, Loss: 0.3121, LR: 0.001000 ... Epoch 20/20, Loss: 0.1523, LR: 0.000500 测试集评估: 准确率: 0.9230 AUC: 0.9521
核心记忆点
| 要点 | 说明 |
|---|---|
| 标准化 | fit_transform训练集,transform测试集 |
| 张量形状 | y 必须 reshape 成(N, 1) |
| 模型结构 | Linear + BatchNorm + ReLU + Dropout |
| 输出层 | 不加 Sigmoid,配BCEWithLogitsLoss |
| 不平衡 | pos_weight处理 |
| 训练 7 步 | zero_grad → forward → loss → backward → step |
| train/eval | 训练用train(),评估用eval() |
| no_grad | 评估时关闭梯度 |
| 评估指标 | 不平衡看 AUC,不只看准确率 |
常见的 7 个坑
忘了
optimizer.zero_grad()→ 梯度累加,训练发散。忘了
model.train()/model.eval()→ Dropout 和 BatchNorm 行为不对。数据没移到 GPU→ 报错 "Expected all tensors on same device"。
y 形状不对→
BCELoss报错,必须(N, 1)。BCEWithLogitsLoss配 Sigmoid→ 重复计算,数值不稳定。测试集
fit标准化→ 数据泄露,评估虚高。只看准确率→ 不平衡时虚高,要看 AUC、F1、Recall。
一句话总结
PyTorch DNN 二分类流程:
数据:生成 → 划分 → 标准化 → 转张量 → DataLoader。
模型:继承
nn.Module,堆叠 Linear + BN + ReLU + Dropout。训练:
train()→ 7 步循环 →scheduler.step()。评估:
eval()→no_grad()→ Sigmoid → AUC。