第三十七天打卡

知识点回顾：

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

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

import pandas as pd
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler, StandardScaler, OneHotEncoder, LabelEncoder
import time
import matplotlib.pyplot as plt
from tqdm import tqdm
from imblearn.over_sampling import SMOTE# ------------------- 设备配置（GPU/CPU） -------------------
# 检查是否有可用的GPU：如果有则用GPU加速训练（速度更快），否则用CPU
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(f"使用设备: {device}")  # 打印当前使用的设备（确认是否启用GPU）# ------------------- 加载并清洗数据 -------------------
# 加载信贷预测数据集（假设data.csv在当前目录下）
# 数据包含用户信息（如收入、工作年限）和标签（是否违约：Credit Default）
data = pd.read_csv(r'data.csv')# 丢弃无用的Id列（Id是用户唯一标识，与信贷违约无关）
data = data.drop(['Id'], axis=1)  # axis=1表示按列删除# 区分连续特征（数值型）和离散特征（文本型/类别型）
# 连续特征：比如年龄、收入（可以取任意数值）
# 离散特征：比如职业、教育程度（只能取有限的类别）
continuous_features = data.select_dtypes(include=['float64', 'int64']).columns.tolist()  # 数值列
discrete_features = data.select_dtypes(exclude=['float64', 'int64']).columns.tolist()  # 非数值列# 离散特征用众数（出现次数最多的值）填充缺失值
# 例：如果"职业"列有缺失，用出现最多的职业填充
for feature in discrete_features:if data[feature].isnull().sum() > 0:  # 检查是否有缺失值mode_value = data[feature].mode()[0]  # 计算众数data[feature].fillna(mode_value, inplace=True)  # 填充缺失值# 连续特征用中位数（中间位置的数）填充缺失值
# 例：如果"收入"列有缺失，用所有收入的中间值填充（比平均数更抗异常值）
for feature in continuous_features:if data[feature].isnull().sum() > 0:median_value = data[feature].median()  # 计算中位数data[feature].fillna(median_value, inplace=True)# ------------------- 离散特征编码（转成数值） -------------------
# 有顺序的离散特征（比如"工作年限"有"1年"<"2年"<"10+年"）用标签编码（转成数字）
mappings = {"Years in current job": {"10+ years": 10,  # "10+年"对应数字10（最大）"2 years": 2,     # "2年"对应数字2"3 years": 3,"< 1 year": 0,    # "<1年"对应数字0（最小）"5 years": 5,"1 year": 1,"4 years": 4,"6 years": 6,"7 years": 7,"8 years": 8,"9 years": 9},"Home Ownership": {  # 房屋所有权（有顺序：租房 < 房贷 < 有房贷 < 自有房？）"Home Mortgage": 0,  # 房贷"Rent": 1,           # 租房"Own Home": 2,       # 自有房"Have Mortgage": 3   # 有房贷（可能顺序需要根据业务调整）},"Term": {  # 贷款期限（短期 < 长期）"Short Term": 0,  # 短期"Long Term": 1    # 长期}
}# 使用映射字典将文本转成数字（标签编码）
data["Years in current job"] = data["Years in current job"].map(mappings["Years in current job"])
data["Home Ownership"] = data["Home Ownership"].map(mappings["Home Ownership"])
data["Term"] = data["Term"].map(mappings["Term"])# 无顺序的离散特征（比如"贷款用途"：购车/教育/装修，彼此无大小关系）用独热编码
# 独热编码：将1列转成N列（N是类别数），每列用0/1表示是否属于该类别
data = pd.get_dummies(data, columns=['Purpose'])  # 对"Purpose"列做独热编码# 独热编码后会生成新列（比如Purpose_购车、Purpose_教育），需要将这些列的类型从bool转成int（0/1）
list_final = []  # 存储新生成的列名
data2 = pd.read_csv(r'data.csv')  # 重新读取原始数据（对比列名）
for i in data.columns:if i not in data2.columns:  # 原始数据没有的列，就是新生成的独热列list_final.append(i)
for i in list_final:data[i] = data[i].astype(int)  # 将bool型（True/False）转成int（1/0）# ------------------- 分离特征和标签 -------------------
X = data.drop(['Credit Default'], axis=1)  # 特征数据（所有列，除了标签列）
y = data['Credit Default']  # 标签数据（0=未违约，1=违约）# 划分训练集（80%）和测试集（20%）：训练集用来学习规律，测试集验证模型效果
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)  # random_state固定随机种子，保证结果可复现# 特征标准化（将特征缩放到0-1区间，避免大数值特征"欺负"小数值特征）
scaler = MinMaxScaler()  # 创建MinMaxScaler（最小-最大标准化）
X_train = scaler.fit_transform(X_train)  # 用训练集拟合标准化参数并转换
X_test = scaler.transform(X_test)  # 用训练集的参数转换测试集（保证数据分布一致）# 将数据转成PyTorch张量（神经网络只能处理张量数据），并移动到GPU（如果有）
# FloatTensor：32位浮点数（特征数据）
# LongTensor：64位整数（标签数据，分类任务需要）
X_train = torch.FloatTensor(X_train).to(device)
y_train = torch.LongTensor(y_train.values).to(device)
X_test = torch.FloatTensor(X_test).to(device)
y_test = torch.LongTensor(y_test.values).to(device)# ------------------- 定义神经网络模型 -------------------
class MLP(nn.Module):def __init__(self):super(MLP, self).__init__()  # 调用父类构造函数（必须）# 全连接层1：输入30个特征（根据数据预处理后的列数确定），输出64个神经元self.fc1 = nn.Linear(30, 64)self.relu = nn.ReLU()  # 激活函数（引入非线性，让模型能学习复杂规律）self.dropout = nn.Dropout(0.2)  # Dropout层（随机丢弃30%的神经元，防止过拟合）# 全连接层2：输入64个神经元，输出32个神经元self.fc2 = nn.Linear(64, 32)# 全连接层3：输入32个神经元，输出2个类别（0=未违约，1=违约）self.fc3 = nn.Linear(32, 2)def forward(self, x):# 前向传播：数据从输入层→隐藏层→输出层的计算流程x = self.fc1(x)    # 输入层→隐藏层1：30→64x = self.relu(x)   # 激活函数（过滤负数值）x = self.dropout(x)  # 应用Dropout（防止过拟合）x = self.fc2(x)    # 隐藏层1→隐藏层2：64→32x = self.relu(x)   # 激活函数x = self.fc3(x)    # 隐藏层2→输出层：32→2（输出未归一化的分数）return x# ------------------- 初始化模型、损失函数、优化器 -------------------
model = MLP().to(device)  # 实例化模型并移动到GPU
criterion = nn.CrossEntropyLoss()  # 交叉熵损失函数（适合分类任务）
optimizer = optim.Adam(model.parameters(), lr=0.001)  # Adam优化器（比SGD更智能，自动调整学习率）# ------------------- 模型训练 -------------------
num_epochs = 20000  # 训练轮数（完整遍历训练集的次数）
train_losses = []         # 记录每200轮的训练损失
test_losses = []          # 记录每200轮的测试损失
accuracies = []     # 记录每200轮的测试准确率
epochs = []         # 记录对应的轮数
# ==========新增早停相关参数==========
best_test_loss = float('inf')  # 记录最佳测试集损失
best_epoch = 0                 # 记录最佳epoch
patience =  50              # 早停耐心值（连续多少轮测试集损失未改善时停止训练）
counter = 0                    # 早停计数器
early_stopped = False          # 是否早停标志
# =====================================
start_time = time.time()  # 记录训练开始时间# 创建tqdm进度条（可视化训练进度）
with tqdm(total=num_epochs, desc="训练进度", unit="epoch") as pbar:for epoch in range(num_epochs):# 前向传播：模型根据输入数据计算预测值outputs = model(X_train)  # 模型输出（形状：[训练样本数, 2]，表示每个样本属于2个类别的分数）train_loss = criterion(outputs, y_train)  # 计算损失（预测值与真实标签的差异，越小越好）# 反向传播和参数更新optimizer.zero_grad()  # 清空历史梯度（避免梯度累加）train_loss.backward()        # 反向传播计算梯度（自动求导）optimizer.step()       # 根据梯度更新模型参数（优化器核心操作）# 每200轮记录一次损失和准确率（避免记录太频繁影响速度）if (epoch + 1) % 200 == 0:# 在测试集上评估模型（不更新参数，只看效果）model.eval()  # 切换到评估模式（关闭Dropout，保证结果稳定）with torch.no_grad():  # 禁用梯度计算（节省内存，加速推理）test_outputs = model(X_test)  # 测试集预测值test_loss = criterion(test_outputs, y_test)  # 计算测试集损失model.train()  # 切换回训练模式# 记录损失值和准确率train_losses.append(train_loss.item())  # 训练集损失test_losses.append(test_loss.item())  # 测试集损失epochs.append(epoch + 1)  # 记录轮数# 更新进度条显示的信息（当前损失和准确率）pbar.set_postfix({'Train Loss': f'{train_loss.item():.4f}', 'Test Loss': f'{test_loss.item():.4f}'})# ===== 新增早停逻辑 =====if test_loss.item() < best_test_loss: # 如果当前测试集损失小于最佳损失best_test_loss = test_loss.item() # 更新最佳损失best_epoch = epoch + 1 # 更新最佳epochcounter = 0 # 重置计数器# 保存最佳模型torch.save(model.state_dict(), 'best_model.pth')else:counter += 1if counter >= patience:print(f"早停触发！在第{epoch+1}轮，测试集损失已有{patience}轮未改善。")print(f"最佳测试集损失出现在第{best_epoch}轮，损失值为{best_test_loss:.4f}")early_stopped = Truebreak  # 终止训练循环# ======================# 每1000轮更新一次进度条（避免进度条刷新太频繁）if (epoch + 1) % 1000 == 0:pbar.update(1000)  # 进度条前进1000步# 确保进度条最终显示100%（防止最后一轮未更新）if pbar.n < num_epochs:pbar.update(num_epochs - pbar.n)# 计算总训练时间并打印
time_all = time.time() - start_time
print(f'Training time: {time_all:.2f} seconds')# ===== 新增：加载最佳模型用于最终评估 =====
if early_stopped:print(f"加载第{best_epoch}轮的最佳模型进行最终评估...")model.load_state_dict(torch.load('best_model.pth'))
# ================================# ------------------- 可视化训练结果 -------------------
# 创建双y轴图表（损失和准确率在同一张图显示）
# 可视化损失曲线
plt.figure(figsize=(10, 6))
plt.plot(epochs, train_losses, label='Train Loss')
plt.plot(epochs, test_losses, label='Test Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('Training and Test Loss over Epochs')
plt.legend()
plt.grid(True)
plt.show()# ------------------- 最终测试集评估 -------------------
model.eval()  # 切换到评估模式
with torch.no_grad():  # 禁用梯度计算outputs = model(X_test)  # 测试集预测值_, predicted = torch.max(outputs, 1)  # 取预测类别（0或1）correct = (predicted == y_test).sum().item()  # 正确预测的样本数accuracy = correct / y_test.size(0)  # 计算准确率print(f'测试集准确率: {accuracy * 100:.2f}%')  # 打印准确率（百分比形式）
#测试集准确率: 77.20%