"""
演示
手机价格分类案例
注意：
这里是预测手机价格区间，0，1，2，3，是一个单标签多分类任务，我们用全连接神经网络实现
神经网络训练的工作流
1.准备数据集
拆分数据集为训练集和测试集。这里简化了数据拆分，实际工程中会拆分为训练集、验证集(模拟考试)、测试集(最终考试)
numpy数值 -> 张量 -> 数据集对象 -> 数据加载器
2.搭建神经网络
使用全连接神经网络
1.继承nn.Module
2.定义__init__方法，构造网络层
3.定义forward方法，前向传播
3.模型训练-在训练集上训练
model.train()
前向传播
计算损失
梯度清零
反向传播
更新参数
4.模型测试-在测试集上测试
model.eval()
"""
# 导入相关模块
import os
import torch
from torch.utils.data import TensorDataset, DataLoader # 数据集对象 数据加载器
import torch.nn as nn # 神经网络模块，实现各种网络层
import torch.optim as optim # 优化器模块，实现各种优化器，比如SGD,Adam/AdamW
from sklearn.model_selection import train_test_split # 划分数据集,pip install scikit-learn
import matplotlib.pyplot as plt # pip install matplotlib
import numpy as np # pip install matplotlib
import pandas as pd # pip install pandas
import time
from torchsummary import summary # 统计模型参数量
# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei'] # 微软雅黑
plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题
# 设置设备，GPU > MPS > CPU
device = torch.device(
"cuda" if torch.cuda.is_available()
else "mps" if torch.backends.mps.is_available()
else "cpu"
)
print(f"当前设备：{device}")
# 设置随机种子
torch.manual_seed(28)
# 1.准备数据集
def create_dataset():
# 1.加载CSV
df = pd.read_csv(r"data/手机价格预测.csv")
# 2.构造输入特征x,输出标签y
# 前20列是输入特征，最后一列是标签y
x = df.iloc[:, :-1]
y = df.iloc[:, -1]
# 转换x的元素类型为浮点数float32
x = x.astype(np.float32)
# print(f"x:{x}, y:{y}")
# print(f"x.shape:{x.shape}, y.shape:{y.shape}")
# 3.拆分数据集为训练集和测试集
# 参数1：输入特征x；参数2：标签y；参数3：训练集拆分比例；参数4：随机状态，固定随机性
# 参数5：stratify,按哪个标签比例进行层次拆分，比如y中类别比例为1：1：1：1，拆分之后的数据也要符合这个比例
x_train, x_test, y_train, y_test = train_test_split(x,y,train_size=0.8,random_state=5,stratify=y)
# print(f"x_train.shape:{x_train.shape}, x_test.shape:{x_test.shape}, y_train.shape:{y_train.shape}, y_test.shape:{y_test.shape}")
# 4.numpy数值 -> 张量 torch.tensor()
x_train = torch.tensor(x_train.values,dtype=torch.float32)
x_test = torch.tensor(x_test.values,dtype=torch.float32)
y_train = torch.tensor(y_train.values,dtype=torch.long)
y_test = torch.tensor(y_test.values,dtype=torch.long)
print(f"x_train.shape:{x_train.shape}, x_test.shape:{x_test.shape}, y_train.shape:{y_train.shape}, y_test.shape:{y_test.shape}")
# 5.张量 -> 数据集对象, 封装输入特征x和输出标签y
train_dataset = TensorDataset(x_train,y_train) # 数据集对象用于定义如何构造一个样本
test_dataset = TensorDataset(x_test,y_test)
# 6.返回
return train_dataset, test_dataset
# 2.搭建神经网络
class MyPhonePriceClassifier(nn.Module):
# 1.定义__init__方法，构造网络层
def __init__(self, input_dim=20, output_dim=4):
# 1.初始化父类成员
super().__init__()
# 2.定义第一个线性层,自动进行参数初始化,kaiming初始化
self.linear1 = nn.Linear(input_dim,256)
# 3.定义第二个线性层
self.linear2 = nn.Linear(256,128)
# 4.定义第三个线性层，输出层
self.linear3 = nn.Linear(128,output_dim)
# 2.定义forward方法，前向传播
def forward(self,x):
# x:(batch_size,20)
# 1.数据输入linear1，relu
x = torch.relu(self.linear1(x)) # (batch_size,256)
# 2.数据输入linear2,relu
x = torch.relu(self.linear2(x)) # (batch_size,128)
# 3.数据输入linear3
x = self.linear3(x) # (batch_size,output_dim)
# 4.返回结果
return x
# 3.模型训练-在训练集上训练
# train_dataset:训练集数据集对象
# input_dim:数据的特征维度，20
# output_dim: 输出维度，4，也就是有4个类别
def train(train_dataset, input_dim=20, output_dim=4, epochs=10):
# 1.创建数据加载器对象
# 参数1：数据集对象；参数2：批次大小；参数3：是否打乱数据，训练时打乱，测试时不打乱；参数4: 是否删除最后一个不完整的批次数据
train_dataloader = DataLoader(train_dataset,batch_size=32,shuffle=True,drop_last=False)
# 2.创建神经网络模型对象
model = MyPhonePriceClassifier(
input_dim=input_dim,
output_dim=output_dim
).to(device) # 迁移模型到设备
# 3.定义损失函数和优化器
loss_fn = nn.CrossEntropyLoss()
optimizer = optim.AdamW(model.parameters(),lr=1e-3,weight_decay=1e-3)
# 定义变量，记录 每个轮次的 训练损失 和 准确率, 用于可视化
train_losses = []
train_accs = []
# 4.模型训练
# 遍历轮数epochs,举个栗子：值日生扫地，要扫10个轮次
for epoch in range(epochs):
# 1.设置训练模式
model.train()
# 2.初始化 当前轮次的 训练的总损失，总样本数，预测正确的样本数
train_loss = 0.0
total_samples = 0
total_correct = 0
# 获取当前时刻
start_time = time.time()
# 3.遍历数据加载器，分批次训练，举个栗子：值日生扫地，在扫第N个轮次的时候，要逐个扫过每一排，就是每个batch
for x, y in train_dataloader:
# x:(batch_size,input_dim)
# y:(batch_size,)
# 0.迁移数据到device
x = x.to(device)
y = y.to(device)
# 1.前向传播
y_pred = model(x) # (batch_size,4)
# print(f"y_pred.shape:{y_pred.shape}")
# 2.计算损失
loss = loss_fn(y_pred, y)
# print(f"loss:{loss}, shape:{loss.shape}") # 0D
# 3.梯度清零
optimizer.zero_grad()
# 4.反向传播
loss.backward()
# 5.更新参数
optimizer.step()
# 6.更新训练指标，训练损失，训练样本数，预测正确的样本数
train_loss += loss.item() * x.shape[0]
total_samples += x.shape[0]
# 更新预测正确的数量
total_correct += (y_pred.argmax(dim=-1) == y).sum().item()
# 4.计算当前轮次的训练平均损失和训练准确率
avg_loss = train_loss / total_samples
acc = total_correct / total_samples
# 获取训练损失和准确率
train_losses.append(avg_loss)
train_accs.append(acc)
# 5.打印训练损失和训练准确率
print(f"Epoch: {epoch+1}/{epochs} | "
f"训练损失：{avg_loss:.4f} | "
f"训练准确率：{acc:.4f} | "
f"训练时间：{(time.time() - start_time):.2f}s")
...
# 5.保存模型文件
# 新建文件夹model
os.makedirs("./model", exist_ok=True)
# 保存模型参数文件，因为在不同设备上迁移更加方便，如果把模型结构也保存了，那么模型在不同设备上迁移困难
print(f"模型参数：{model.state_dict()}")
torch.save(model.state_dict(), r"./model/phone_price_classifier.pth") # python模型文件格式有.pth/.pt,两个都一样
# 返回训练损失和训练准确率
return train_losses, train_accs
# 3.1 可视化训练过程
def plot_train_process(train_losses, train_accs):
plt.figure(figsize=(12, 6))
# 绘制训练损失曲线
plt.subplot(1, 2, 1)
plt.plot(train_losses, label="train_loss")
plt.xlabel("epoch")
plt.ylabel("loss")
plt.legend()
plt.grid(True)
# 绘制训练准确率曲线
plt.subplot(1, 2, 2)
plt.plot(train_accs, label="train_acc")
plt.xlabel("epoch")
plt.ylabel("acc")
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# 4.模型测试-在测试集上测试
# 参数1：测试集
@torch.no_grad() # 关闭梯度计算，节省显卡内存20%
def evaluate(test_dataset, input_dim=20, output_dim=4):
# 1.创建数据加载器对象
# 参数1：数据集对象；参数2：批次大小；参数3：是否打乱数据，训练时打乱，测试时不打乱；参数4: 是否删除最后一个不完整的批次数据
test_dataloader = DataLoader(test_dataset, batch_size=32, shuffle=False, drop_last=False)
# 2.创建神经网络模型对象
model = MyPhonePriceClassifier(
input_dim=input_dim,
output_dim=output_dim
).to(device)
# 迁移模型到设备
# 3.加载模型参数，将模型参数文件的模型参数加载到示例化的模型对象中
model.load_state_dict(torch.load(r"model/phone_price_classifier.pth"))
# 定义损失函数
loss_fn = nn.CrossEntropyLoss()
# 4.模型测试，设置为评估模式model.eval()，告诉模型，现在是训练/评估，不要开启dropout等特殊网络层
model.eval()
# 定义变量，记录测试 的总损失，总样本数，预测正确的样本数
test_loss = 0.0
total_samples = 0
total_correct = 0
# 遍历批次，也就是遍历数据加载器
for x, y in test_dataloader:
# x:(batch_size,input_dim)
# y:(batch_size,)
# 0.迁移数据到device
x = x.to(device)
y = y.to(device)
# 1.前向传播
y_pred = model(x) # (batch_size,4)
# 2.计算损失
loss = loss_fn(y_pred, y)
# 梯度清零
# 反向传播
# 更新参数
# 3.更新指标，损失，样本数，预测正确的样本数
batch_size = x.shape[0]
# 更新总损失
test_loss += loss.item() * batch_size # 当前批次的总损失=平均损失*当前批次的批次大小
# 更新样本数
total_samples += batch_size
# 更新预测正确的样本数
total_correct += (y_pred.argmax(dim=-1) == y).sum().item()
...
# 计算测试损失和测试准确率
avg_test_loss = test_loss / total_samples
avg_test_acc = total_correct / total_samples
# 返回结果
return avg_test_loss, avg_test_acc
# 测试
if __name__ == "__main__":
# 1.准备数据集
train_dataset, test_dataset=create_dataset()
print(f"训练集数量：{len(train_dataset)}")
print(f"测试集数量：{len(test_dataset)}")
print(f"train_dataset:{train_dataset}, test_dataset:{test_dataset}")
# # 2.创建模型对象
# model = MyPhonePriceClassifier() # 默认模型对象在CPU上
# model = model.to(device) # 将模型对象移动到device上，比如GPU
# print(model)
# summary(model, input_size=(20,))
# 3.模型训练
train_losses, train_accs=train(train_dataset, input_dim=20, output_dim=4, epochs=10)
# 可视化训练过程
plot_train_process(train_losses, train_accs)
# 4.模型测试
avg_test_loss, avg_test_acc = evaluate(test_dataset, input_dim=20, output_dim=4)
print(f"测试损失：{avg_test_loss:.4f} | 测试准确率：{avg_test_acc:.4f}")