新冠病毒预测项目实战

一.导入包

import torch
import matplotlib.pyplot as plt #画图
import matplotlib
matplotlib.use("TkAgg")
import numpy as np  #矩阵相关
import csv     #csv文件
import pandas  #csv文件
from torch.utils.data import Dataset, DataLoader
import torch.nn as nn
from torch import optim
import time

注意：

若电脑显卡配置高则额外添加以下包：

import matplotlib
matplotlib.use("TkAgg")

且若用到

model = torch.load(model_path).to(device)

改为

model = torch.load(model_path, weights_only=False).to(device) 

二.核心代码

1.处理数据

# 定义数据集类 - 就像创建一个专门处理COVID数据的"数据处理器"
class CovidDataset(Dataset):    # 继承PyTorch的Dataset类
    def __init__(self, file_path, mode):
        # 打开CSV文件
        with open(file_path, "r") as f:
            # 读取CSV文件的所有内容，变成Python列表格式
            ori_data = list(csv.reader(f))
            # 将数据转为numpy数组，去掉第一行标题和第一列ID，只保留数值数据
            # 这就像我们看数据时跳过表头，只看实际的数字部分
            csv_data = np.array(ori_data)[1:, 1:].astype(float)        
            
            # 划分训练集和验证集：每5个数据取4个作为训练，1个作为验证
            # 这就像把一叠卡片每5张分成一组，每组取4张练习，留1张检查
            if mode == "train":
                indices = [i for i in range(len(csv_data)) if i % 5 != 0]
            elif mode == "val":
                indices = [i for i in range(len(csv_data)) if i % 5 == 0]
            elif mode == "test":
                indices = [i for i in range(len(csv_data))]  # 测试集用全部数据
            
            # 提取特征数据（前93列）
            X = torch.tensor(csv_data[indices, :93])
            
            # 如果不是测试集，提取标签数据（最后一列），这是我们要预测的值
            if mode != "test":
                self.Y = torch.tensor(csv_data[indices, -1])
            
            # 对特征数据进行标准化处理：让每个特征都变成均值为0，标准差为1
            # 这就像把所有的数据都调整到统一的尺度上比较
            self.X = (X - X.mean(dim=0, keepdim=True)) / X.std(dim=0, keepdim=True)
            self.mode = mode  # 记住当前是训练、验证还是测试模式
    def __getitem__(self, item):
        # 根据索引获取一个数据样本
        # 如果是测试模式，只返回特征X；否则返回特征X和标签Y
        # 像从一堆卡片中抽出一张来看
        if self.mode == "test":
            return self.X[item].float()
        else:
            return self.X[item].float(), self.Y[item].float()

    def __len__(self):
        # 返回数据集的大小，像数一数有多少张卡片
        return len(self.X)

要继承PyTorch的Dataset类获取所需数据X和Y

init函数作为Python类的初始化方法，会在创建对象时自动调用，它接收两个参数，file_path是数据的路径(CSV文件)，mode是训练模式"train"/“val”/“test”(训练/验证/测试)

直接在左侧复制绝对路径，类似下图

逢五取一(太简单一般不用)的方式将covid_train.csv划分为训练集和验证集，val(验证集)模式下逢五取一，train(训练集)模式下非五就取，用indices列表，存放序号，再用indices取出csv_data里的数据

2.构造模型

myModel继承torch.nn.Module，用于设计神经网络的模型框架，这里设计的是2层全连接网络

# 定义神经网络模型
class myModel(nn.Module):      # 继承PyTorch的Module类
    def __init__(self, inDim):
        super(myModel, self).__init__()  # 调用父类初始化方法
        
        # 第一层全连接层：把93个输入特征变成128个隐藏特征
        # 这就像把93个观察指标转化成128个更抽象的特征
        self.fc1 = nn.Linear(inDim, 128)
        
        # ReLU激活函数：给神经网络增加非线性，让它能学习复杂模式
        self.relu1 = nn.ReLU()
        
        # 第二层全连接层：把128个隐藏特征变成1个输出值
        # 这就像从128个抽象特征中计算出最终的预测结果
        self.fc2 = nn.Linear(128, 1)

    def forward(self, x):    # 数据向前传播
        # 数据通过第一层，就像信息通过第一道处理工序
        x = self.fc1(x)
        # 经过ReLU激活，让负值归零，正值保留
        x = self.relu1(x)
        # 数据通过第二层，得到最终的输出
        x = self.fc2(x)

        # 如果输出的维度大于1，去掉多余的维度（确保输出形状正确）
        # 这就像整理文件，去掉不必要的包装
        if len(x.size()) > 1:
            x = x.squeeze(1)       
        return x

forward函数定义了前向传播的过程。在按批次输入参数x时，x的形状为(batch_size, inDim)，而在输出时，形状为(batch_size, 1)，这不是我们想要的，我们想要的形状是(batch_size)，要去掉多余的维度，因为我们做的是回归预测，预测值通常是一维的连续数值，所以要按行合并。

3.训练流程与超参数

(1)训练流程

def train_val(model, train_loader, val_loader, lr, optimizer, device, epochs, save_path):
    model = model.to(device)    # 把模型放到指定的设备（GPU或CPU）上
    
    # 记录训练和验证的损失，用于画图观察学习过程
    plt_train_loss = []  # 训练损失记录
    plt_val_loss = []    # 验证损失记录
    min_val_loss = 999999999999999999.9  # 初始化一个很大的最小损失值

    # 开始训练，每一轮叫一个epoch
    for epoch in range(epochs):          
        model.train()  # 告诉模型现在是训练模式，可以更新参数
        start_time = time.time()  # 记录本轮开始时间
        train_loss = 0.0  # 初始化本轮训练损失
        
        # 遍历训练数据，每次取一小批（batch）数据
        for x, y in train_loader:
            x, y = x.to(device), y.to(device)  # 把数据移到指定设备
            y_pred = model(x)  # 让模型做预测
            bat_loss = loss(y_pred, y, model)  # 计算这批数据的损失
            bat_loss.backward()  # 反向传播，计算梯度
            optimizer.step()  # 更新模型参数
            optimizer.zero_grad()  # 清空梯度，为下一批数据做准备，不清空梯度会累计
            train_loss += bat_loss.cpu().item()  # 累计损失

        # 计算平均训练损失，并记录下来
        plt_train_loss.append(train_loss/train_loader.__len__())#等同len(train_loader)

        # 切换到验证模式
        model.eval()  # 告诉模型现在是评估模式，不更新参数
        val_loss = 0.0
        with torch.no_grad():  # 验证时不计算梯度，节省内存
            for val_x, val_y in val_loader:
                val_x, val_y = val_x.to(device), val_y.to(device)
                val_pred_y = model(val_x)  # 用训练好的模型做预测
                val_bat_loss = loss(val_pred_y, val_y, model)  # 计算验证损失
                val_loss += val_bat_loss.cpu().item()
        plt_val_loss.append(val_loss / val_loader.__len__())

        # 如果这轮的验证损失是到目前为止最小的，保存模型
        # 像考试时发现这次成绩最好，把这次的学习方法记下来
        if val_loss < min_val_loss:
            min_val_loss = val_loss
            torch.save(model, save_path)

        # 打印本轮训练信息：第几轮、耗时、训练损失、验证损失
        print("[%03d/%03d] %2.2f sec(s)  train_loss: %.6f val_loss:%.6f" % \
              (epoch, epochs, time.time()-start_time, plt_train_loss[-1], plt_val_loss[-1]))

    # 训练结束后，画出损失变化图，看看学习过程
    plt.plot(plt_train_loss)
    plt.plot(plt_val_loss)
    plt.title("loss")
    plt.legend(["train", "val"])
    plt.show()

在每一轮epoch中，会首先将模型切换至训练模式。然后写一个for循环，用x和y从事先做好的可迭代对象train_loader中按batch_size接收数据，然后将x放入已经实例化的模型model完成前向传播，然后给函数调用loss传参计算损失，然后调用bat_loss.backward()完成反向传播，然后用优化器进行参数更新和梯度清零，最后累加到本批次的总损失上。for循环结束后，记录该损失的算术平均值。

而在验证模式中，流程基本一致，只是验证过程不涉及训练模型，所以在前向传播中不能计算梯度，前向传播的过程记录了梯度的计算式，但是没有loss所以更新不了，反向传播后有了loss才能进行更新，所以这里是让前向不去计算式子，不更新张量网

(2)超参数

val_loader: 验证数据的数据加载器，用于批量提供验证数据。
train_loader: 训练数据的数据加载器，用于批量提供训练数据。
model: 需要训练的模型，通常是nn.Module的子类。
lr: 学习率（learning rate），用于优化器更新参数的步长。
optimizer: 优化器，如SGD、Adam等，用于更新模型参数。
device: 设备（如'cuda'或'cpu'），模型和数据将被移动到的设备。
epochs: 训练的总轮数。
save_path: 模型保存的路径。

train_file = r"C:\Users\Administrator\Desktop\李哥深度学习项目班\第三节，回归实战代码\regression\covid\covid.train.csv"
test_file = r"C:\Users\Administrator\Desktop\李哥深度学习项目班\第三节，回归实战代码\regression\covid\covid.test.csv"

# for x, y in train_set:
#     pred_y = model(x)
#     print(pred_y)

batch_size = 16

train_set = CovidDataset(train_file, "train")
val_set = CovidDataset(train_file, "val")
test_set = CovidDataset(test_file, "test")

train_loader = DataLoader(train_set, batch_size=batch_size, shuffle=True)
val_loader = DataLoader(val_set, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_set, batch_size=1, shuffle=False)


def mseLoss(pred, target, model):
    loss = nn.MSELoss(reduction='mean')    #平方差损失
    ''' Calculate loss '''
    regularization_loss = 0                    # 正则项
    for param in model.parameters():
        # TODO: you may implement L1/L2 regularization here
        # 使用L2正则项
        # regularization_loss += torch.sum(abs(param))
        regularization_loss += torch.sum(param ** 2)                  # 计算所有参数平方
    return loss(pred, target) + 0.00075 * regularization_loss             # 返回损失。

loss = mseLoss
epochs = 20   #运行轮次
lr = 0.001     #学习率
device = "cuda" if torch.cuda.is_available() else "cpu"
print(device)

data_dim = 93
model = myModel(data_dim).to(device)
save_path = "model_save/best_model.pth"
rel_path = "pred.csv"
optimizer = optim.SGD(params=model.parameters(), lr=lr, momentum=0.9)


train_val(model, train_loader, val_loader, lr, optimizer, device, epochs, save_path)

注意最后一个批次的数据量可能不是 batch_size

测试集不需要分批次，也不需要打乱，到时候就按顺序一个一个预测得到结果就好

• mseLoss

使用mseLoss函数进行优化，mseLoss函数会计算损失，计算公式为：

当前批次损失 = MSE(当前批次预测值, 当前批次真实值) + 0.00075 × Σ(模型所有参数²)

在本案例中，损失是唯一衡量训练好坏的参考

注：loss = mseLoss 是将函数对象赋值给变量 loss，那里不是调用函数，所以不传递参数

• optimizer

优化器类型： SGD：随机梯度下降优化器（Stochastic Gradient Descent），随机的意思就是每次更新只使用一个批次的数据计算梯度，而非全部数据（普通的梯度下降使用全部数据）

PyTorch 不会提供默认优化器，必须显式定义

0.9 是基于经验的选择，适用于大多数情况，通常在 0.5-0.99 之间

(3)预测结果

def evaluate(model_path, test_loader, rel_path, device):
    # 加载之前保存的最好模型
    model = torch.load(model_path).to(device)

    rel = []  # 保存预测结果
    model.eval()  # 设置为评估模式
    with torch.no_grad():  # 不计算梯度
        for x in test_loader:
            x = x.to(device)
            pred = model(x)  # 做预测
            rel.append(pred.cpu().item())  # 保存结果
    
    # 把预测结果保存到CSV文件中
    with open(rel_path, "w", newline="") as f:
        csv_writer = csv.writer(f)
        csv_writer.writerow(["id", "tested_positive"])  # 写表头
        for i, pred in enumerate(rel):     # 逐行写入预测结果
            csv_writer.writerow([str(i), str(pred)])
        print("结果保存到了"+rel_path)

#提交

evaluate(save_path, test_loader, rel_path, device)

使用保存过的最优模型来预测，并保存到rel_path中，并按照题目要求提交，完成。