新冠病毒预测的简单调包实战。

一、引进包

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、数据类

"""
    定义Dataset   
    基本都是 init getitem len 三个
"""
class CovidDataset(Dataset):
    def __init__(self, file_path, mode):
        with open(file_path, "r") as f:    # 文件指针f
            ori_data = list(csv.reader(f))
            csv_data = np.array(ori_data)[1:, 1:].astype(float)      # 切片掉第一行和第一列
            # 训练集和验证集逢五取1(8:2)，不推荐
            if mode == "train":
                indices = [i for i in range(len(csv_data)) if i % 5 != 0]	# 这里的len方法是魔术方法，下面有写

            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))]

            X = torch.tensor(csv_data[indices, :93])    # 切片是左闭右开
            if mode != "test":
                self.Y = torch.tensor(csv_data[indices, -1])  # 取最后一列

            # 无量纲化
            self.X = (X - X.mean(dim=0, keepdim=True)) / X.std(dim=0, keepdim=True)
            self.mode = mode	# 加self方便传递

    def __getitem__(self, item):
        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)

CovidDataset继承Dataset类，用于获取所需数据X和Y

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

在文件同目录下，有covid.test.csv和covid_train.csv两个数据文件，观察发现，数据的第一行是数据名称，第一列是数据id，这是训练所不需要的，故切片删掉，获得csv_data，这是一个数据类型为float的numpy数组

然后根据不同的模式，选择不同的数据处理方法。我们需要将covid_train.csv划分为训练集和验证集，这里划分的比例为8:2，也就是“val”模式下逢五取一，“train”模式下非五就取；而covid.test.csv全部取出就好。这样我们会获得indices列表，存放序号。为什么逢五取一不推荐？因为太简单了，可以有更好的随机划分的方法

获得序号后，使用indices取出csv_data里的数据，去除id后covid_train.csv有94个维度，covid.test.csv有93个维度，多出来的那个维度就是标签Y，所以X统一取序号为indices列表内容的前93个维度，而在"train"/"val"模式下，要额外取出最后一列作为标签Y

注意到数据不同纬度的值大小和范围差异较大，于是进行无量纲化处理。这里选择标准化（Z-score），减去均值除以标准差。原数据不服从正态分布也能用标准化吗？可以，反正确实有用

关于getitem和len函数，getitem 实现容器协议，使对象支持索引操作（obj[key]），len 实现长度协议，使对象支持 len() 函数

2、模型类

"""
    定义模型Model
    这里只是做好模型框架，不涉及训练
"""
class myModel(nn.Module):
    def __init__(self, inDim): 		 # inDim输入维度
        super(myModel, self).__init__()		# 调用父类构造函数
        self.fc1 = nn.Linear(inDim, 128)
        self.relu1 = nn.ReLU()
        self.fc2 = nn.Linear(128, 1)

    def forward(self, x):
        x = self.fc1(x)
        x = self.relu1(x)
        x = self.fc2(x)

        if len(x.size()) > 1:       # 这里是2个维度
            x = x.squeeze(1)        # 如果维度大于1，就去掉第二个维度(1表示第二个维度)
        return x

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

在初始化方法init中，需要inDim参数，其实就是数据X的维度(inDim = 93)。要先调用父类构造函数，因为要用到一些父类中写好的变量。第一层网络 self.fc1，是输入层到隐藏层的全连接层（inDim → 128）激活函数选择ReLU 激活函数；第二层网络self.fc2是隐藏层到输出层的全连接层（128 → 1）

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

# 假设输入一个批次的数据，batch_size=3
input_tensor = torch.tensor([[1, 2, 3], [4, 5, 6], [7, 8, 9]])  # 形状: (3, 3)

# 经过模型后
output_before = model(input_tensor)  # 输出可能是: tensor([[0.5], [1.2], [0.8]])
print(output_before.shape)  # torch.Size([3, 1])  ← 有两个维度

# 执行维度处理
if len(output_before.size()) > 1:  # 检查维度数: 2 > 1
    output_after = output_before.squeeze(1)  # 去除第2个维度(索引为1)
print(output_after.shape)  # torch.Size([3])  ← 只有一个维度

# 结果对比
# 处理前: tensor([[0.5], [1.2], [0.8]]) - 形状 (3, 1)
# 处理后: tensor([0.5, 1.2, 0.8])     - 形状 (3,)

3、超参数和训练流程

3.1 超参数

我们会用到这样的训练函数

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

model：myModel实例化的神经网络模型

train_loader：训练数据加载器

val_loader：验证数据加载器

lr： 学习率

optimize：优化器

device：训练设备（CPU/GPU）

epochs：训练轮次

save_path：模型保存路径

所以设置超参数如下：

def mseLoss(pred, target, model):
    loss = nn.MSELoss(reduction='mean')    # loss定义为平方差损失
    regularization_loss = 0                    # 正则项
    for param in model.parameters():
        regularization_loss += torch.sum(param ** 2)           # 计算所有参数平方和
    return loss(pred, target) + 0.00075 * regularization_loss        # 0.00075（正则化强度）

# 数据加载参数
train_file = r"E:\复试\李哥考研\第二节 回归实战\代码\regression\covid\covid.train.csv"
test_file = r"E:\复试\李哥考研\第二节 回归实战\代码\regression\covid\covid.test.csv"

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)   # shuffle=True打乱 
val_loader = DataLoader(val_set, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_set, batch_size=1, shuffle=False)     # 测试集不打乱

# 训练参数
loss = mseLoss	# 函数对象赋值
epochs = 20   # 运行轮次
lr = 0.001     # 学习率
device = "cuda" if torch.cuda.is_available() else "cpu"     # 判断GPU可用
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)   # momentum 动量 用于找全局最优解

· DataLoader说明：

基本概念
继承自：torch.utils.data.DataLoader 是 PyTorch 提供的数据加载器类
作用：将数据集封装成可迭代对象，支持批量加载、随机打乱等功能

主要参数
dataset：传入的数据集对象（这里就是 CovidDataset类）
batch_size：每个批次的样本数量
shuffle：是否随机打乱数据顺序

实现功能：

通过DataLoader，将数据集按batch_size整理成可以被迭代的对象。具体做法是先 shuffle = True 是将整个数据集的样本顺序随机打乱，然后按顺序取出 batch_size 个连续的样本作为批次，这样每个批次的x的形状为tensor(16,93)，y变成tensor(16,)。例如：

原始数据索引 [0,1,2,3,4,5,6,7,8,9]
打乱后可能变成 [3,7,1,9,0,5,2,8,4,6]
分批: [3,7,1,9], [0,5,2,8], [4,6]  # 每批4个

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

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

· mseLoss说明：

mseLoss函数会计算损失，计算公式为：

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

注意 当前批次 与 整个模型

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

参数：pred：模型预测值，target：真实标签值，model：神经网络模型

返回值类型为张量tensor

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

· optimizer说明：

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

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

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

3.2 训练流程

def train_val(model, train_loader, val_loader, lr, optimizer, device, epochs, save_path):
    model = model.to(device)    #防止意外 即插即用
    plt_train_loss = []  # 总训练loss
    plt_val_loss = []    # 总验证loss
    min_val_loss = 999999999999999999.9
    for epoch in range(epochs):         # 发枪指令， 冲锋的号角， 模型训练的开始
        
        model.train()   # 模型训练模式
        start_time = time.time()
        train_loss = 0.0  # 浮点形式
        for x, y in train_loader:
            x, y = x.to(device), y.to(device)   # 把数据送到GPU
            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()   # .cpu() 将张量从GPU转移到CPU item() 提取张量中的标量值
        plt_train_loss.append(train_loss/train_loader.__len__())

        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()   # 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()

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

train_val实现了完整的训练和验证流程，主要进行了模型训练和验证，记录了模型训练过程和验证过程的损失变化，并保存历史最佳模型。

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

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

每完成一轮验证，就判断一下本轮获取的loss值是不是历史最低值(即历史最优模型)，如果是就保存下来。验证集上的loss就是越低越好，训练集不是这样

4、预测结果

def evaluate(model_path, test_loader, rel_path, device):
    model = torch.load(model_path, weights_only = False).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())
    with open(rel_path, "w", newline="") as f:  # f写指针
        csv_writer = csv.writer(f)
        csv_writer.writerow(["id", "tested_positive"])
        for i, pred in enumerate(rel):     # 同时得到 第几个 和第几个的结果 enumrate
            csv_writer.writerow([str(i), str(pred)])
        print("结果保存到了"+rel_path)

evaluate(save_path, test_loader, rel_path, device)

evaluate函数会使用历史最优模型来进行预测，然后保存到rel_path中，这样就得到了预测结果

三、什么是魔术方法？

推荐看看这位佬，写得很有意思