5月26日day37打卡

早停策略和模型权重的保存

知识点回顾：

1. 过拟合的判断：测试集和训练集同步打印指标
2. 模型的保存和加载
   1. 仅保存权重
   2. 保存权重和模型
   3. 保存全部信息checkpoint，还包含训练状态
3. 早停策略

作业：对信贷数据集训练后保存权重，加载权重后继续训练50轮，并采取早停策略

# 导入必要库：数据处理、科学计算、深度学习框架、机器学习工具、可视化库
import pandas as pd# 数据处理库，用于读取CSV文件、数据清洗、特征工程等结构化数据操作
import numpy as np# 数值计算库，提供高效的多维数组运算和数学函数（科学计算基础）
import random# Python内置随机数生成库，用于设置随机种子保证实验可复现性
import torch# PyTorch深度学习框架核心库，支持张量运算、自动微分和GPU加速
import torch.nn as nn# PyTorch神经网络模块，包含层（如全连接层）、损失函数、激活函数等组件
from torch.utils.data import Dataset, DataLoader#导入PyTorch数据集加载相关工具,Dataset类用于创建自定义数据集，DataLoader用于批量加载数据
from sklearn.model_selection import train_test_split# 导入scikit-learn工具,train_test_split用于划分训练集和测试集
from sklearn.preprocessing import StandardScaler, LabelEncoder# StandardScaler用于数据标准化（Z-score标准化）,LabelEncoder用于将类别标签转换为数值编码
from sklearn.metrics import classification_report, roc_auc_score# 导入模型评估指标,classification_report生成分类报告（准确率、召回率、F1等）,roc_auc_score计算ROC AUC评分（用于评估分类模型性能)
import matplotlib.pyplot as plt#导入matplotlib的pyplot模块，用于数据可视化# 固定随机种子函数：确保实验可复现性（所有随机数生成器使用相同初始状态）
def set_seed(seed=42):random.seed(seed)# Python内置随机数生成器np.random.seed(seed)# NumPy随机数生成器torch.manual_seed(seed)# PyTorch CPU随机数生成器torch.cuda.manual_seed_all(seed)# PyTorch GPU随机数生成器（多卡支持）torch.backends.cudnn.deterministic = True# 强制CuDNN使用确定性算法set_seed(42)# 应用固定种子# 读取数据
data = pd.read_csv(r'D:\桌面\研究项目\打卡文件\python60-days-challenge-master\data.csv')
target_col = 'Credit Default'# 目标列（预测变量：信用违约）# 缺失值处理：数值特征用中位数填充，非数值特征用'Unknown'填充
data = data.fillna(data.median(numeric_only=True)) # 数值特征中位数填充
data = data.fillna('Unknown')# 剩余缺失值用'Unknown'标记# 特征分类：手动指定分类特征，其余为数值特征（排除目标列）
categorical_features = ['Home Ownership', 'Purpose', 'Term', 'Years in current job']
numerical_features = [col for col in data.columns if col not in categorical_features + [target_col]]# 分类特征编码：使用LabelEncoder将文本分类特征转换为数值（0~n-1）
for col in categorical_features:le = LabelEncoder()data[col] = le.fit_transform(data[col])# 原地转换# 特征与目标分离：合并分类特征和数值特征作为输入X，目标列作为y
X = data[categorical_features + numerical_features]
y = data[target_col]# 数值特征标准化：消除量纲影响（均值0，标准差1）
scaler = StandardScaler()
X[numerical_features] = scaler.fit_transform(X[numerical_features])# 仅对数值特征标准化# 划分训练集/测试集（按目标列分层划分，保持类别分布一致，随机种子固定）
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, stratify=y, random_state=42)# 自定义数据集类：继承PyTorch Dataset，将数据转换为张量格式
class CreditDataset(Dataset):def __init__(self, X, y):# 将DataFrame转换为float32张量（PyTorch常用浮点精度）self.X = torch.tensor(X.values, dtype=torch.float32)self.y = torch.tensor(y.values, dtype=torch.float32)# 目标值为浮点型（二分类标签）def __len__(self):return len(self.X)# 数据集长度def __getitem__(self, idx):return self.X[idx], self.y[idx]# 按索引获取样本# 创建数据加载器：训练集打乱数据，测试集不打乱
train_dataset = CreditDataset(X_train, y_train)
test_dataset = CreditDataset(X_test, y_test)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)# 训练批次64，打乱数据
test_loader = DataLoader(test_dataset, batch_size=256, shuffle=False)# 测试批次256（更大批次加速推理）# 定义神经网络模型：二分类全连接网络
class CreditNet(nn.Module):def __init__(self, input_dim):super(CreditNet, self).__init__()# 网络结构：线性层+批量归一化+激活函数+Dropout（防止过拟合）self.model = nn.Sequential(nn.Linear(input_dim, 64),# 输入层→隐藏层1（64神经元）nn.BatchNorm1d(64),# 批量归一化（加速训练）nn.ReLU(),# 激活函数（引入非线性）nn.Dropout(0.3), # Dropout（随机失活30%神经元）nn.Linear(64, 32), # 隐藏层1→隐藏层2（32神经元）nn.BatchNorm1d(32),# 批量归一化nn.ReLU(),# 激活函数nn.Dropout(0.2), # Dropout（随机失活20%神经元）nn.Linear(32, 1)# 隐藏层2→输出层（1个神经元，二分类输出）)def forward(self, x):return self.model(x).squeeze(1)# 输出压缩为1维（匹配BCEWithLogitsLoss输入要求）# 设备配置：优先使用GPU（CUDA可用时），否则使用CPU
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = CreditNet(X_train.shape[1]).to(device)# 初始化模型并移动到目标设备
criterion = nn.BCEWithLogitsLoss()# 二分类交叉熵损失（内置Sigmoid，数值更稳定）
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)# Adam优化器（自适应学习率）# 训练函数：执行单轮训练，返回平均训练损失
def train(model, loader, criterion, optimizer):model.train()# 开启训练模式（启用Dropout和BN训练状态）total_loss = 0for X_batch, y_batch in loader:X_batch, y_batch = X_batch.to(device), y_batch.to(device)# 数据移动到目标设备optimizer.zero_grad()# 清空梯度缓存outputs = model(X_batch)# 前向传播（计算预测值）loss = criterion(outputs, y_batch)# 计算损失（预测值与真实值差异）loss.backward()# 反向传播（计算梯度）optimizer.step()# 优化器更新参数total_loss += loss.item() * X_batch.size(0)# 累计总损失（乘以批次大小）return total_loss / len(loader.dataset) # 返回平均损失（总损失/数据集大小）# 评估函数：计算测试集AUC和分类报告
def evaluate(model, loader):model.eval() # 开启评估模式（禁用Dropout和BN更新）preds, targets = [], []with torch.no_grad():# 禁用梯度计算（加速推理）for X_batch, y_batch in loader:X_batch = X_batch.to(device)# 数据移动到目标设备outputs = torch.sigmoid(model(X_batch)).cpu().numpy()# 计算Sigmoid概率并转移到CPUpreds.extend(outputs)# 收集预测概率targets.extend(y_batch.numpy()) # 收集真实标签preds = np.array(preds)targets = np.array(targets)preds_label = (preds > 0.5).astype(int)# 概率转换为二分类标签（阈值0.5）auc = roc_auc_score(targets, preds)# 计算AUC（概率与真实标签的排序性能）report = classification_report(targets, preds_label, digits=4)# 分类报告（精确率、召回率等）return auc, report# 训练主循环：执行多轮训练并记录指标
epochs = 20
train_losses = []# 记录每轮训练损失
test_aucs = []# 记录每轮测试AUCfor epoch in range(epochs):train_loss = train(model, train_loader, criterion, optimizer)# 执行训练auc, _ = evaluate(model, test_loader) # 执行评估train_losses.append(train_loss)test_aucs.append(auc)# 打印当前轮次训练指标print(f"Epoch {epoch+1}/{epochs} - Train Loss: {train_loss:.4f} - Test AUC: {auc:.4f}")# 训练过程可视化：绘制损失曲线和AUC曲线
plt.figure(figsize=(12,5))
# 训练损失子图
plt.subplot(1,2,1)
plt.plot(range(1, epochs+1), train_losses, marker='o')
plt.xlabel('Epoch')
plt.ylabel('Train Loss')
plt.title('Training Loss Curve')
plt.grid(True)
# 测试AUC子图
plt.subplot(1,2,2)
plt.plot(range(1, epochs+1), test_aucs, marker='o', color='orange')
plt.xlabel('Epoch')
plt.ylabel('Test AUC')
plt.title('Test AUC Curve')
plt.grid(True)
plt.tight_layout() # 自动调整子图间距
plt.show()# 显示图表# 保存模型权重
torch.save(model.state_dict(), "credit_model.pth")

# 定义早停类：用于在模型性能不再提升时提前终止训练，防止过拟合
class EarlyStopping:def __init__(self, patience=5, delta=1e-4):"""参数：patience (int): 允许性能不提升的最大轮数（耐心值）delta (float): 性能提升的最小阈值（当前分数需超过最佳分数+delta才算有效提升）"""self.patience = patience # 耐心值（连续多少轮无提升则停止）self.delta = delta# 最小提升阈值（防止微小波动误判）self.best_score = None# 记录当前最佳分数（初始为None）self.counter = 0# 无提升轮数计数器self.early_stop = False# 早停触发标志（默认未触发）def __call__(self, score):"""根据当前分数更新早停状态"""# 首次调用或当前分数显著优于最佳分数时更新最佳值if self.best_score is None or score > self.best_score + self.delta:self.best_score = score# 更新最佳分数self.counter = 0# 重置无提升计数器else:self.counter += 1# 无提升轮数+1# 当无提升轮数超过耐心值时触发早停if self.counter >= self.patience:self.early_stop = True# 设置早停标志为True# 加载已保存的模型权重并继续训练（基于之前训练的模型参数）
model.load_state_dict(torch.load("credit_model.pth")) # 从本地文件加载模型参数
epochs_continue = 50 # 继续训练的总轮数
early_stopping = EarlyStopping(patience=5, delta=1e-4)  # 初始化早停监控器（耐心5轮，最小提升0.0001）
train_losses2 = []# 记录继续训练过程中的训练损失
test_aucs2 = []# 记录继续训练过程中的测试AUC# 继续训练主循环
for epoch in range(epochs_continue):# 执行单轮训练并计算损失train_loss = train(model, train_loader, criterion, optimizer)# 评估测试集并获取AUC分数auc, _ = evaluate(model, test_loader)# 记录指标train_losses2.append(train_loss)test_aucs2.append(auc)# 打印当前轮次训练信息（带[Continue]标识区分阶段）print(f"[Continue] Epoch {epoch+1}/{epochs_continue} - Train Loss: {train_loss:.4f} - Test AUC: {auc:.4f}")# 更新早停状态（传入当前测试AUC作为评估指标）early_stopping(auc)# 检查是否触发早停if early_stopping.early_stop:print("Early stopping triggered!")# 打印早停提示break # 终止训练循环# 可视化继续训练的损失和AUC曲线（与初始训练阶段对比）
plt.figure(figsize=(12,5))
# 训练损失子图
plt.subplot(1,2,1)
plt.plot(range(1, len(train_losses2)+1), train_losses2, marker='o')# 绘制损失变化曲线
plt.xlabel('Epoch')# x轴标签：轮次
plt.ylabel('Train Loss')# y轴标签：训练损失
plt.title('Continue Training Loss Curve') # 图表标题：继续训练损失曲线
plt.grid(True)# 显示网格线# 测试AUC子图
plt.subplot(1,2,2)
plt.plot(range(1, len(test_aucs2)+1), test_aucs2, marker='o', color='orange')# 绘制AUC变化曲线
plt.xlabel('Epoch') # x轴标签：轮次
plt.ylabel('Test AUC')# y轴标签：测试AUC
plt.title('Continue Test AUC Curve')# 图表标题：继续训练AUC曲线
plt.grid(True)# 显示网格线
plt.tight_layout()# 自动调整子图间距防止重叠
plt.show()# 显示图表# 最终模型评估（训练结束后对测试集进行完整评估）
auc, report = evaluate(model, test_loader)# 获取最终AUC和分类报告
print(f"\nFial Test AUC: {auc:.4f}")# 打印最终测试AUC分数
print("Classification Report:\n", report)# 打印详细分类报告（精确率、召回率等指标）

 

@浙大疏锦行