手机价格分类数据集已经上传,用户可以自行下载进行训练。
构建数据集
数据共有 2000 条, 其中 1600 条数据作为训练集, 400 条数据用作测试集。 我们使用 sklearn 的数据集划分工作来完成 。并使用 PyTorch 的 TensorDataset 来将数据集构建为 Dataset 对象,方便构造数据集加载对象。
构建分类网络模型
构建全连接神经网络来进行手机价格分类,该网络主要由四个线性层来构建,使用relu激活函数。
网络共有 4个全连接层, 具体信息如下:
1.第一层: 输入为维度为 20, 输出维度为: 128
2.第二层: 输入为维度为 128, 输出维度为: 256
3.第三层: 输入为维度为 256, 输出维度为: 64
4.第四层: 输入为维度为 64, 输出维度为: 4
模型训练
网络编写完成之后,我们需要编写训练函数。所谓的训练函数,指的是输入数据读取、送入网络、计算损失、更新参数 的流程,该流程较为固定。我们使用的是多分类交叉生损失函数、使用 SGD 优化方法。最终,将训练好的模型持久化 到磁盘中。
编写评估函数
使用训练好的模型,对未知的样本的进行预测的过程。我们这里使用前面单独划分出来的验证集来进行评估。
第一次训练:
import pandas as pd
import torch
from sklearn.model_selection import train_test_split
from torch.utils.data import TensorDataset, DataLoader
from torch import nn
from torchsummary import summary
from torch import optim
import matplotlib.pyplot as plt
# 数据获取与预处理
data = pd.read_csv('./深度学习/data/手机价格预测.csv') # 从CSV文件加载数据
# print(data.head()) # 打印前几行数据(暂时注释掉)
# 分离特征和目标
x = data.iloc[:, :-1] # 获取除最后一列之外的所有特征列
y = data.iloc[:, -1] # 获取最后一列作为目标变量(标签)
# 划分训练集和测试集,测试集比例为20%
x_train, x_test, y_train, y_test = train_test_split(x, y, test_size=0.2, random_state=22)
# 将数据转换为TensorDataset,供PyTorch使用
train_dataset = TensorDataset(torch.tensor(x_train.values, dtype=torch.float32), # 训练集特征转为Tensor
torch.tensor(y_train.values, dtype=torch.int64)) # 训练集标签转为Tensor
test_dataset = TensorDataset(torch.tensor(x_test.values, dtype=torch.float32), # 测试集特征转为Tensor
torch.tensor(y_test.values, dtype=torch.int64)) # 测试集标签转为Tensor
# print(x.shape)
# print(y.shape)
# 定义神经网络模型
class PhoneModel(nn.Module): # 继承自PyTorch的nn.Module类
def __init__(self):
super(PhoneModel, self).__init__()
# 定义三层全连接层与一个输出层
self.layer1 = nn.Linear(in_features=20, out_features=128) # 第一层:输入特征为20,输出为128
self.layer2 = nn.Linear(in_features=128, out_features=256) # 第二层:输入128,输出256
self.layer3 = nn.Linear(in_features=256, out_features=64) # 第三层:输入256,输出64
self.out = nn.Linear(in_features=64, out_features=4) # 输出层:输入64,输出4(假设有4个分类)
self.dropout = nn.Dropout(p=0.4) # Dropout层,用于防止过拟合,p=0.4 表示40%的神经元随机失活
def forward(self, x):
# 前向传播过程定义
x = self.layer1(x) # 输入经过第一层全连接层
x = self.dropout(x) # 进行Dropout操作(随机失活)
x = torch.relu(x) # ReLU激活函数
x = self.dropout(torch.relu(self.layer2(x))) # 第二层:全连接 + Dropout + ReLU激活
x = torch.relu(self.dropout(self.layer3(x))) # 第三层:全连接 + Dropout + ReLU激活
out = self.out(x) # 输出层
return out # 返回输出
# 实例化模型
model = PhoneModel()
summary(model, input_size=(20,), batch_size=8) # 打印模型结构和参数数量,input_size为每个样本输入的形状
# 模型训练部分
optimizer = optim.SGD(model.parameters(), lr=0.00001) # 使用随机梯度下降法(SGD)优化器,学习率设置为0.00001
error = nn.CrossEntropyLoss() # 交叉熵损失函数,适用于多分类问题
epochs = 100 # 定义训练轮数
train_loss_per_epoch = [] # 用于记录每轮的训练损失
# 开始训练
for epoch in range(epochs):
dataloader = DataLoader(train_dataset, batch_size=8, shuffle=True) # 创建DataLoader,按批量(batch)喂入数据
loss_sum = 0 # 用于计算损失总和
num_batches = 0 # 记录每个epoch的批次数量
# 遍历每一个小批次
for x_batch, y_batch in dataloader:
y_pred = model(x_batch) # 模型对当前批次数据进行预测
loss = error(y_pred, y_batch) # 计算当前批次的损失
loss_sum += loss.item() # 累加损失
num_batches += 1 # 更新批次数量
optimizer.zero_grad() # 梯度清零,避免梯度累加
loss.backward() # 反向传播计算梯度
optimizer.step() # 更新模型参数
avg_loss = loss_sum / num_batches # 计算每轮的平均损失
train_loss_per_epoch.append(avg_loss) # 将当前轮的损失记录到列表中
# print(f"Epoch {epoch+1}/{epochs}, Loss: {avg_loss}") # 打印每轮的损失值(暂时注释掉)
# 画出每轮的训练损失曲线
plt.plot(range(1, epochs + 1), train_loss_per_epoch, label='Train Loss') # 绘制损失曲线
plt.xlabel('Epochs') # X轴为训练轮数
plt.ylabel('Loss') # Y轴为损失值
plt.title('Training Loss Over Epochs') # 图表标题
plt.legend() # 添加图例
plt.show() # 显示图像
# 模型保存
torch.save(model.state_dict(), './深度学习/model/phone_model.pth') # 保存模型的参数到文件 'model.pth'
# 模型评估
model = PhoneModel()
model.load_state_dict(torch.load('./深度学习/model/phone_model.pth')) # 加载已经保存的模型参数
test_dataloader = DataLoader(test_dataset, batch_size=8, shuffle=False) # 创建测试集的DataLoader
correct_predictions = 0 # 记录正确预测的数量
for x_batch, y_batch in test_dataloader:
y_pred = model(x_batch) # 模型对测试集进行预测
out = torch.argmax(y_pred, dim=1) # 找到预测值中概率最高的类别
correct_predictions += (y_batch == out).sum().item() # 统计正确预测的数量
# 计算并打印准确率
accuracy = correct_predictions / len(test_dataset) # 计算模型的准确率
print(f"Test Accuracy: {accuracy * 100:.2f}%") # 打印准确率
第一次训练的模型的测试集输出:
----------------------------------------------------------------
Layer (type) Output Shape Param #
================================================================
Linear-1 [8, 128] 2,688
Dropout-2 [8, 128] 0
Linear-3 [8, 256] 33,024
Dropout-4 [8, 256] 0
Linear-5 [8, 64] 16,448
Dropout-6 [8, 64] 0
Linear-7 [8, 4] 260
================================================================
Total params: 52,420
Trainable params: 52,420
Non-trainable params: 0
----------------------------------------------------------------
Input size (MB): 0.00
Forward/backward pass size (MB): 0.05
Params size (MB): 0.20
Estimated Total Size (MB): 0.26
----------------------------------------------------------------
<Figure size 640x480 with 1 Axes>
Test Accuracy: 40.50%
epochs = 100 # 定义训练轮数
第二次训练
我们数据进行标准化,使用adam优化器(可以自适应学习率和修正梯度),我们减少了神经网络层级,增加了训练次数,
from sklearn.preprocessing import StandardScaler
import pandas as pd
import torch
from sklearn.model_selection import train_test_split
from torch.utils.data import TensorDataset, DataLoader
from torch import nn
from torchsummary import summary
from torch import optim
import matplotlib.pyplot as plt
# Step 1: 数据获取与预处理
# 从CSV文件加载数据
data = pd.read_csv('../data/手机价格预测.csv')
# 分离特征和目标变量
# X 代表特征 (输入数据),去掉最后一列
# y 代表目标 (标签),最后一列
X = data.iloc[:, :-1] # 获取除最后一列之外的所有特征列
y = data.iloc[:, -1] # 获取最后一列作为目标变量(分类标签)
# 标准化特征数据
# 初始化标准化对象,将特征数据进行标准化(零均值,单位方差)
scaler = StandardScaler()
X = scaler.fit_transform(X) # 标准化特征,使每列特征的均值为0,标准差为1
# 划分训练集和测试集
# 使用 train_test_split 将数据分为80%的训练集和20%的测试集,random_state=22 保持随机性的可重复性
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=22)
# 将数据转换为 TensorDataset,供 PyTorch 使用
# 将训练集和测试集特征和标签转为 PyTorch 的张量格式
train_dataset = TensorDataset(torch.tensor(X_train, dtype=torch.float32), torch.tensor(y_train.values, dtype=torch.int64))
test_dataset = TensorDataset(torch.tensor(X_test, dtype=torch.float32), torch.tensor(y_test.values, dtype=torch.int64))
# Step 2: 定义神经网络模型
# 定义一个多层全连接神经网络,继承自 nn.Module
class PhonePriceModel(nn.Module):
def __init__(self):
super(PhonePriceModel, self).__init__()
# 定义网络结构
# 第一层:输入特征20,输出128个神经元
self.layer1 = nn.Linear(in_features=20, out_features=128)
# 第二层:输入128个神经元,输出64个神经元
self.layer2 = nn.Linear(in_features=128, out_features=64)
# 输出层:输入64个神经元,输出4个类别 (假设是 4 分类问题)
self.out = nn.Linear(in_features=64, out_features=4)
# Dropout 层,防止过拟合,p=0.4 表示40%的神经元随机失活
self.dropout = nn.Dropout(p=0.4)
# 前向传播函数,定义数据如何流经网络层
def forward(self, x):
# 第一层:全连接层 -> ReLU 激活 -> Dropout
x = self.dropout(torch.relu(self.layer1(x)))
# 第二层:全连接层 -> ReLU 激活 -> Dropout
x = self.dropout(torch.relu(self.layer2(x)))
# 输出层:直接输出,不需要激活函数,交叉熵损失会自动处理
out = self.out(x)
return out
# 实例化模型
model = PhonePriceModel()
# 打印模型结构和参数数量
summary(model, input_size=(20,), batch_size=16)
# Step 3: 模型训练部分
# 使用 Adam 优化器,设置学习率为 0.0001,betas 参数用于控制一阶和二阶矩估计的加权平均
optimizer = torch.optim.Adam(model.parameters(), lr=0.0001, betas=(0.9, 0.99))
# 使用交叉熵损失函数,适用于多分类问题
error = nn.CrossEntropyLoss()
# 设定训练轮数为1000轮
epochs = 1000
# 用于记录每轮的训练损失值
train_loss_per_epoch = []
# 开始训练循环
for epoch in range(epochs):
# 使用 DataLoader 加载训练数据,设置批量大小为 8,数据会在每个 epoch 随机打乱
dataloader = DataLoader(train_dataset, batch_size=8, shuffle=True)
# 初始化每轮的损失总和
loss_sum = 0
# 记录批次数量
num_batches = 0
# 遍历每个小批次
for x_batch, y_batch in dataloader:
# 模型预测当前批次的标签
y_pred = model(x_batch)
# 计算当前批次的损失
loss = error(y_pred, y_batch)
# 累加损失
loss_sum += loss.item()
# 更新批次数量
num_batches += 1
# 梯度清零,防止梯度累加
optimizer.zero_grad()
# 反向传播计算梯度
loss.backward()
# 优化器更新模型参数
optimizer.step()
# 计算每轮的平均损失
avg_loss = loss_sum / num_batches
# 将当前轮的损失记录到列表中
train_loss_per_epoch.append(avg_loss)
# Step 4: 画出训练损失曲线
# 绘制每轮的训练损失变化
plt.plot(range(1, epochs + 1), train_loss_per_epoch, label='Train Loss')
plt.xlabel('Epochs') # 设置 x 轴标签为训练轮数
plt.ylabel('Loss') # 设置 y 轴标签为损失值
plt.title('Training Loss Over Epochs') # 设置图表标题
plt.legend() # 显示图例
plt.show() # 显示图像
# Step 5: 模型保存
# 保存模型的参数到文件
torch.save(model.state_dict(), '../model/phone_model2.pth')
# Step 6: 模型评估
# 加载模型并进行测试集评估
# 首先重新实例化模型并加载保存的权重
model = PhonePriceModel()
model.load_state_dict(torch.load('../model/phone_model2.pth'))
# 创建测试集的 DataLoader,批量大小为8,测试时不需要打乱数据
test_dataloader = DataLoader(test_dataset, batch_size=8, shuffle=False)
# 初始化正确预测的数量
correct_predictions = 0
# 遍历测试集,进行预测
for x_batch, y_batch in test_dataloader:
# 模型预测
y_pred = model(x_batch)
# 获取预测类别,取每个样本的最大概率对应的类别
out = torch.argmax(y_pred, dim=1)
# 计算正确预测的数量
correct_predictions += (y_batch == out).sum().item()
# 计算并打印准确率
accuracy = correct_predictions / len(test_dataset) # 准确率为正确预测的样本数量除以总样本数
print(f"Test Accuracy: {accuracy * 100:.2f}%") # 打印准确率
第二次的输出
----------------------------------------------------------------
Layer (type) Output Shape Param #
================================================================
Linear-1 [16, 128] 2,688
Dropout-2 [16, 128] 0
Linear-3 [16, 64] 8,256
Dropout-4 [16, 64] 0
Linear-5 [16, 4] 260
================================================================
Total params: 11,204
Trainable params: 11,204
Non-trainable params: 0
----------------------------------------------------------------
Input size (MB): 0.00
Forward/backward pass size (MB): 0.05
Params size (MB): 0.04
Estimated Total Size (MB): 0.09
----------------------------------------------------------------
<Figure size 640x480 with 1 Axes>
Test Accuracy: 89.50%
AdamW, 加入L2正则化防止过拟合,加入nn.BatchNorm1d # 批量归一化层,防止过拟合,训练批次增大改成16,增加平滑度。继续训练1000次
from sklearn.preprocessing import StandardScaler
import pandas as pd
import torch
from sklearn.model_selection import train_test_split
from torch.utils.data import TensorDataset, DataLoader
from torch import nn
from torchsummary import summary
from torch import optim
import matplotlib.pyplot as plt
import numpy as np
# Step 1: 数据获取与预处理
# 从CSV文件加载数据,并分离特征和目标变量
data = pd.read_csv('../data/手机价格预测.csv') # 从文件中读取数据
X = data.iloc[:, :-1] # 获取特征列,去除最后一列
y = data.iloc[:, -1] # 获取目标列,即最后一列(分类标签)
# 特征标准化
scaler = StandardScaler() # 初始化标准化对象
X = scaler.fit_transform(X) # 对特征进行标准化(均值为0,方差为1)
# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=22)
# 将训练集和测试集数据转换为Tensor格式,以供PyTorch使用
train_dataset = TensorDataset(torch.tensor(X_train, dtype=torch.float32), torch.tensor(y_train.values, dtype=torch.int64))
test_dataset = TensorDataset(torch.tensor(X_test, dtype=torch.float32), torch.tensor(y_test.values, dtype=torch.int64))
# Step 2: 定义神经网络模型
# 定义一个多层全连接神经网络,继承自nn.Module
class PhonePriceModel(nn.Module):
def __init__(self):
super(PhonePriceModel, self).__init__()
# 定义网络结构,包括全连接层、批量归一化层和Dropout层
self.layer1 = nn.Linear(in_features=20, out_features=256) # 输入层,20维特征,256个神经元
self.bn1 = nn.BatchNorm1d(256) # 批量归一化层,减少训练过程中的梯度消失问题
self.layer2 = nn.Linear(in_features=256, out_features=128) # 隐藏层,256 -> 128
self.bn2 = nn.BatchNorm1d(128) # 批量归一化层
self.layer3 = nn.Linear(in_features=128, out_features=64) # 隐藏层,128 -> 64
self.bn3 = nn.BatchNorm1d(64) # 批量归一化层
self.out = nn.Linear(in_features=64, out_features=4) # 输出层,假设有4个分类
self.dropout = nn.Dropout(p=0.4) # Dropout层,p=0.4表示40%的神经元在训练过程中随机失活,防止过拟合
def forward(self, x):
# 定义前向传播过程
x = self.dropout(torch.relu(self.bn1(self.layer1(x)))) # 第一层 + 批量归一化 + ReLU激活函数 + Dropout
x = self.dropout(torch.relu(self.bn2(self.layer2(x)))) # 第二层 + 批量归一化 + ReLU激活函数 + Dropout
x = self.dropout(torch.relu(self.bn3(self.layer3(x)))) # 第三层 + 批量归一化 + ReLU激活函数 + Dropout
out = self.out(x) # 输出层
return out # 返回模型的预测结果
# 实例化模型
model = PhonePriceModel()
summary(model, input_size=(20,), batch_size=16) # 打印模型结构和参数数量
# Step 3: 设置优化器和损失函数
# 使用AdamW优化器,加入weight_decay防止过拟合(L2正则化)
optimizer = torch.optim.AdamW(model.parameters(), lr=0.0001, betas=(0.9, 0.99), weight_decay=0.01)
# 使用交叉熵损失函数,适用于多分类任务
error = nn.CrossEntropyLoss()
# 设定早停策略,防止过拟合(如果验证损失10个epoch不下降,就停止训练)
class EarlyStopping:
def __init__(self, patience=10, delta=0):
self.patience = patience # 最大等待的epoch数量
self.delta = delta # 损失值变化的最小差异
self.best_loss = np.inf # 最佳验证损失初始值为无穷大
self.counter = 0 # 记录等待的epoch数量
self.early_stop = False # 是否停止训练
def __call__(self, val_loss):
if val_loss < self.best_loss - self.delta: # 如果验证损失有显著下降
self.best_loss = val_loss # 更新最佳验证损失
self.counter = 0 # 重置等待计数
else:
self.counter += 1 # 否则等待计数+1
if self.counter >= self.patience: # 超过最大等待epoch,停止训练
self.early_stop = True
# Step 4: 模型训练部分
epochs = 1000 # 定义训练轮数
train_loss_per_epoch = [] # 记录每个epoch的训练损失
val_loss_per_epoch = [] # 记录每个epoch的验证损失
early_stopping = EarlyStopping(patience=10, delta=0.01) # 定义早停策略
for epoch in range(epochs):
# 训练阶段
model.train() # 启用Dropout和BatchNorm
train_dataloader = DataLoader(train_dataset, batch_size=16, shuffle=True) # 创建训练数据的DataLoader
loss_sum = 0 # 累积损失
num_batches = 0 # 批次数量
for x_batch, y_batch in train_dataloader:
y_pred = model(x_batch) # 前向传播,得到模型预测
loss = error(y_pred, y_batch) # 计算损失
loss_sum += loss.item() # 累积损失
num_batches += 1
optimizer.zero_grad() # 梯度清零
loss.backward() # 反向传播
optimizer.step() # 更新模型参数
avg_loss = loss_sum / num_batches # 计算每个epoch的平均损失
train_loss_per_epoch.append(avg_loss)
# 验证阶段
model.eval() # 禁用Dropout和BatchNorm
val_loss_sum = 0 # 累积验证损失
num_val_batches = 0
with torch.no_grad(): # 禁用梯度计算
for x_batch, y_batch in DataLoader(test_dataset, batch_size=16, shuffle=False):
y_pred = model(x_batch) # 前向传播
val_loss = error(y_pred, y_batch) # 计算验证集损失
val_loss_sum += val_loss.item() # 累积损失
num_val_batches += 1
val_avg_loss = val_loss_sum / num_val_batches # 计算平均验证损失
val_loss_per_epoch.append(val_avg_loss)
print(f"Epoch {epoch+1}/{epochs}, Train Loss: {avg_loss:.4f}, Val Loss: {val_avg_loss:.4f}")
# # 早停检查
# early_stopping(val_avg_loss)
# if early_stopping.early_stop:
# print(f"Early stopping at epoch {epoch+1}")
# break # 如果触发早停,结束训练
# Step 5: 画出损失曲线
plt.plot(range(1, len(train_loss_per_epoch) + 1), train_loss_per_epoch, label='Train Loss') # 绘制训练损失
plt.plot(range(1, len(val_loss_per_epoch) + 1), val_loss_per_epoch, label='Validation Loss') # 绘制验证损失
plt.xlabel('Epochs') # x轴标签为epoch数量
plt.ylabel('Loss') # y轴标签为损失值
plt.title('Train and Validation Loss Over Epochs') # 图标题
plt.legend() # 显示图例
plt.show() # 展示图像
# Step 6: 保存模型
torch.save(model.state_dict(), '../model/phone_model_with_early_stopping.pth') # 保存模型的参数
# Step 7: 模型评估
# 加载保存的模型参数
model.load_state_dict(torch.load('../model/phone_model_with_early_stopping.pth'))
model.eval() # 评估模式
test_dataloader = DataLoader(test_dataset, batch_size=16, shuffle=False)
# 创建 DataLoader 对象,用于从测试集加载数据
# test_dataset 是转换为 TensorDataset 格式的测试集
# batch_size=16 表示每次取 16 个样本
# shuffle=False 表示测试数据不需要打乱顺序
correct_predictions = 0
# 初始化正确预测计数器,初始值为 0
with torch.no_grad():
# 禁用梯度计算,因为在测试阶段我们不需要反向传播和梯度计算,可以节省内存和计算时间
for x_batch, y_batch in test_dataloader:
# 遍历测试集的所有批次,每次从 DataLoader 中获取一个批次
# x_batch 是输入特征,y_batch 是对应的真实标签
y_pred = model(x_batch)
# 使用模型对当前批次的输入特征进行预测,得到预测结果 y_pred
out = torch.argmax(y_pred, dim=1)
# 找到每个样本的预测类别,使用 torch.argmax 找到每个样本在 dim=1 维度上(类别维度)最大概率的索引,即模型预测的类别
correct_predictions += (y_batch == out).sum().item()
# 将预测正确的样本数加到正确预测计数器中
# (y_batch == out) 会返回一个布尔张量,表示预测是否正确
# .sum() 计算这个布尔张量中 True(即预测正确)的数量
# .item() 将结果从张量转换为 Python 的标量并累加到 correct_predictions
accuracy = correct_predictions / len(test_dataset)
# 计算模型在测试集上的准确率,正确预测的数量除以测试集的总样本数
print(f"Test Accuracy: {accuracy * 100:.2f}%")
# 打印模型在测试集上的准确率,保留两位小数,并以百分比形式显示
第三次训练输出:
----------------------------------------------------------------
Layer (type) Output Shape Param #
================================================================
Linear-1 [16, 256] 5,376
BatchNorm1d-2 [16, 256] 512
Dropout-3 [16, 256] 0
Linear-4 [16, 128] 32,896
BatchNorm1d-5 [16, 128] 256
Dropout-6 [16, 128] 0
Linear-7 [16, 64] 8,256
BatchNorm1d-8 [16, 64] 128
Dropout-9 [16, 64] 0
Linear-10 [16, 4] 260
================================================================
Total params: 47,684
Trainable params: 47,684
Non-trainable params: 0
----------------------------------------------------------------
Input size (MB): 0.00
Forward/backward pass size (MB): 0.16
Params size (MB): 0.18
Estimated Total Size (MB): 0.35
----------------------------------------------------------------
Epoch 1/1000, Train Loss: 1.4290, Val Loss: 1.3379
Epoch 2/1000, Train Loss: 1.3701, Val Loss: 1.2727
Epoch 3/1000, Train Loss: 1.3121, Val Loss: 1.2192
Epoch 4/1000, Train Loss: 1.2428, Val Loss: 1.1506
.....
Epoch 991/1000, Train Loss: 0.2797, Val Loss: 0.3519
Epoch 992/1000, Train Loss: 0.3372, Val Loss: 0.3389
Epoch 993/1000, Train Loss: 0.2797, Val Loss: 0.3424
Epoch 994/1000, Train Loss: 0.3425, Val Loss: 0.3157
Epoch 995/1000, Train Loss: 0.3567, Val Loss: 0.3393
Epoch 996/1000, Train Loss: 0.3751, Val Loss: 0.3931
Epoch 997/1000, Train Loss: 0.2894, Val Loss: 0.3579
Epoch 998/1000, Train Loss: 0.3667, Val Loss: 0.3269
Epoch 999/1000, Train Loss: 0.3047, Val Loss: 0.3157
Epoch 1000/1000, Train Loss: 0.2907, Val Loss: 0.3280
<Figure size 640x480 with 1 Axes>
Test Accuracy: 87.75%
第四次训练:
本次在第三次的基础上,因为第三次1000次,图片可以看出,梯度后面没啥变化了,所以新增一个设定早停策略,防止过拟合(如果验证损失20个epoch不下降,就停止训练)
from sklearn.preprocessing import StandardScaler
import pandas as pd
import torch
from sklearn.model_selection import train_test_split
from torch.utils.data import TensorDataset, DataLoader
from torch import nn
from torchsummary import summary
from torch import optim
import matplotlib.pyplot as plt
import numpy as np
# Step 1: 数据获取与预处理
# 从CSV文件加载数据,并分离特征和目标变量
data = pd.read_csv('../data/手机价格预测.csv') # 从文件中读取数据
X = data.iloc[:, :-1] # 获取特征列,去除最后一列
y = data.iloc[:, -1] # 获取目标列,即最后一列(分类标签)
# 特征标准化
scaler = StandardScaler() # 初始化标准化对象
X = scaler.fit_transform(X) # 对特征进行标准化(均值为0,方差为1)
# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=22)
# 将训练集和测试集数据转换为Tensor格式,以供PyTorch使用
train_dataset = TensorDataset(torch.tensor(X_train, dtype=torch.float32), torch.tensor(y_train.values, dtype=torch.int64))
test_dataset = TensorDataset(torch.tensor(X_test, dtype=torch.float32), torch.tensor(y_test.values, dtype=torch.int64))
# Step 2: 定义神经网络模型
# 定义一个多层全连接神经网络,继承自nn.Module
class PhonePriceModel(nn.Module):
def __init__(self):
super(PhonePriceModel, self).__init__()
# 定义网络结构,包括全连接层、批量归一化层和Dropout层
self.layer1 = nn.Linear(in_features=20, out_features=256) # 输入层,20维特征,256个神经元
self.bn1 = nn.BatchNorm1d(256) # 批量归一化层,减少训练过程中的梯度消失问题
self.layer2 = nn.Linear(in_features=256, out_features=128) # 隐藏层,256 -> 128
self.bn2 = nn.BatchNorm1d(128) # 批量归一化层
self.layer3 = nn.Linear(in_features=128, out_features=64) # 隐藏层,128 -> 64
self.bn3 = nn.BatchNorm1d(64) # 批量归一化层
self.out = nn.Linear(in_features=64, out_features=4) # 输出层,假设有4个分类
self.dropout = nn.Dropout(p=0.4) # Dropout层,p=0.4表示40%的神经元在训练过程中随机失活,防止过拟合
def forward(self, x):
# 定义前向传播过程
x = self.dropout(torch.relu(self.bn1(self.layer1(x)))) # 第一层 + 批量归一化 + ReLU激活函数 + Dropout
x = self.dropout(torch.relu(self.bn2(self.layer2(x)))) # 第二层 + 批量归一化 + ReLU激活函数 + Dropout
x = self.dropout(torch.relu(self.bn3(self.layer3(x)))) # 第三层 + 批量归一化 + ReLU激活函数 + Dropout
out = self.out(x) # 输出层
return out # 返回模型的预测结果
# 实例化模型
model = PhonePriceModel()
summary(model, input_size=(20,), batch_size=16) # 打印模型结构和参数数量
# Step 3: 设置优化器和损失函数
# 使用AdamW优化器,加入weight_decay防止过拟合(L2正则化)
optimizer = torch.optim.AdamW(model.parameters(), lr=0.0001, betas=(0.9, 0.99), weight_decay=0.01)
# 使用交叉熵损失函数,适用于多分类任务
error = nn.CrossEntropyLoss()
# 设定早停策略,防止过拟合(如果验证损失10个epoch不下降,就停止训练)
class EarlyStopping:
def __init__(self, patience=10, delta=0):
self.patience = patience # 最大等待的epoch数量
self.delta = delta # 损失值变化的最小差异
self.best_loss = np.inf # 最佳验证损失初始值为无穷大
self.counter = 0 # 记录等待的epoch数量
self.early_stop = False # 是否停止训练
def __call__(self, val_loss):
if val_loss < self.best_loss - self.delta: # 如果验证损失有显著下降
self.best_loss = val_loss # 更新最佳验证损失
self.counter = 0 # 重置等待计数
else:
self.counter += 1 # 否则等待计数+1
if self.counter >= self.patience: # 超过最大等待epoch,停止训练
self.early_stop = True
# Step 4: 模型训练部分
epochs = 1000 # 定义训练轮数
train_loss_per_epoch = [] # 记录每个epoch的训练损失
val_loss_per_epoch = [] # 记录每个epoch的验证损失
early_stopping = EarlyStopping(patience=20, delta=0.01) # 定义早停策略
for epoch in range(epochs):
# 训练阶段
model.train() # 启用Dropout和BatchNorm
train_dataloader = DataLoader(train_dataset, batch_size=16, shuffle=True) # 创建训练数据的DataLoader
loss_sum = 0 # 累积损失
num_batches = 0 # 批次数量
for x_batch, y_batch in train_dataloader:
y_pred = model(x_batch) # 前向传播,得到模型预测
loss = error(y_pred, y_batch) # 计算损失
loss_sum += loss.item() # 累积损失
num_batches += 1
optimizer.zero_grad() # 梯度清零
loss.backward() # 反向传播
optimizer.step() # 更新模型参数
avg_loss = loss_sum / num_batches # 计算每个epoch的平均损失
train_loss_per_epoch.append(avg_loss)
# 验证阶段
model.eval() # 禁用Dropout和BatchNorm
val_loss_sum = 0 # 累积验证损失
num_val_batches = 0
with torch.no_grad(): # 禁用梯度计算
for x_batch, y_batch in DataLoader(test_dataset, batch_size=16, shuffle=False):
y_pred = model(x_batch) # 前向传播
val_loss = error(y_pred, y_batch) # 计算验证集损失
val_loss_sum += val_loss.item() # 累积损失
num_val_batches += 1
val_avg_loss = val_loss_sum / num_val_batches # 计算平均验证损失
val_loss_per_epoch.append(val_avg_loss)
print(f"Epoch {epoch+1}/{epochs}, Train Loss: {avg_loss:.4f}, Val Loss: {val_avg_loss:.4f}")
# 早停检查
early_stopping(val_avg_loss)
if early_stopping.early_stop:
print(f"Early stopping at epoch {epoch+1}")
break # 如果触发早停,结束训练
# Step 5: 画出损失曲线
plt.plot(range(1, len(train_loss_per_epoch) + 1), train_loss_per_epoch, label='Train Loss') # 绘制训练损失
plt.plot(range(1, len(val_loss_per_epoch) + 1), val_loss_per_epoch, label='Validation Loss') # 绘制验证损失
plt.xlabel('Epochs') # x轴标签为epoch数量
plt.ylabel('Loss') # y轴标签为损失值
plt.title('Train and Validation Loss Over Epochs') # 图标题
plt.legend() # 显示图例
plt.show() # 展示图像
# Step 6: 保存模型
torch.save(model.state_dict(), '../model/phone_model_with_early_stopping2.pth') # 保存模型的参数
# Step 7: 模型评估
# 加载保存的模型参数
model.load_state_dict(torch.load('../model/phone_model_with_early_stopping2.pth'))
model.eval() # 评估模式
test_dataloader = DataLoader(test_dataset, batch_size=16, shuffle=False)
# 创建 DataLoader 对象,用于从测试集加载数据
# test_dataset 是转换为 TensorDataset 格式的测试集
# batch_size=16 表示每次取 16 个样本
# shuffle=False 表示测试数据不需要打乱顺序
correct_predictions = 0
# 初始化正确预测计数器,初始值为 0
with torch.no_grad():
# 禁用梯度计算,因为在测试阶段我们不需要反向传播和梯度计算,可以节省内存和计算时间
for x_batch, y_batch in test_dataloader:
# 遍历测试集的所有批次,每次从 DataLoader 中获取一个批次
# x_batch 是输入特征,y_batch 是对应的真实标签
y_pred = model(x_batch)
# 使用模型对当前批次的输入特征进行预测,得到预测结果 y_pred
out = torch.argmax(y_pred, dim=1)
# 找到每个样本的预测类别,使用 torch.argmax 找到每个样本在 dim=1 维度上(类别维度)最大概率的索引,即模型预测的类别
correct_predictions += (y_batch == out).sum().item()
# 将预测正确的样本数加到正确预测计数器中
# (y_batch == out) 会返回一个布尔张量,表示预测是否正确
# .sum() 计算这个布尔张量中 True(即预测正确)的数量
# .item() 将结果从张量转换为 Python 的标量并累加到 correct_predictions
accuracy = correct_predictions / len(test_dataset)
# 计算模型在测试集上的准确率,正确预测的数量除以测试集的总样本数
print(f"Test Accuracy: {accuracy * 100:.2f}%")
# 打印模型在测试集上的准确率,保留两位小数,并以百分比形式显示
第四次的输出
----------------------------------------------------------------
Layer (type) Output Shape Param #
================================================================
Linear-1 [16, 256] 5,376
BatchNorm1d-2 [16, 256] 512
Dropout-3 [16, 256] 0
Linear-4 [16, 128] 32,896
BatchNorm1d-5 [16, 128] 256
Dropout-6 [16, 128] 0
Linear-7 [16, 64] 8,256
BatchNorm1d-8 [16, 64] 128
Dropout-9 [16, 64] 0
Linear-10 [16, 4] 260
================================================================
Total params: 47,684
Trainable params: 47,684
Non-trainable params: 0
----------------------------------------------------------------
Input size (MB): 0.00
Forward/backward pass size (MB): 0.16
Params size (MB): 0.18
Estimated Total Size (MB): 0.35
----------------------------------------------------------------
Epoch 1/1000, Train Loss: 1.4297, Val Loss: 1.3407
Epoch 2/1000, Train Loss: 1.3686, Val Loss: 1.2921
Epoch 3/1000, Train Loss: 1.3004, Val Loss: 1.2270
Epoch 4/1000, Train Loss: 1.2460, Val Loss: 1.1759
Epoch 5/1000, Train Loss: 1.1892, Val Loss: 1.1108
。。。。。。
Epoch 85/1000, Train Loss: 0.5395, Val Loss: 0.3130
Epoch 86/1000, Train Loss: 0.4959, Val Loss: 0.3044
Epoch 87/1000, Train Loss: 0.4969, Val Loss: 0.3064
Epoch 88/1000, Train Loss: 0.5360, Val Loss: 0.3087
Epoch 89/1000, Train Loss: 0.5321, Val Loss: 0.3122
Early stopping at epoch 89
<Figure size 640x480 with 1 Axes>
Test Accuracy: 91.75%
后续如果还想提升的话,可以修改自己的超参数慢慢的炼丹。我们只需要符合需求即可。