这是一个关于病毒感染人数的预测项目的,main函数的主体的讲解。
问题 1:PyTorch 模型相较于其他模型(工具)存在的意义?

先明确:PyTorch 不是 “模型”,是用来搭建 / 训练深度学习模型的工具包(和 TensorFlow/Keras 是同类),它的核心意义就是 “让普通人能轻松写深度学习模型”,对比其他工具(比如纯 Python 手写、TensorFlow),优势和意义体现在 3 点:

对比维度纯 Python 手写模型TensorFlow(老版本)PyTorch
通俗比喻用石头 / 木头自己造一辆车用工厂流水线造车(流程固定,改起来麻烦)用乐高积木造车(灵活,想怎么拼就怎么拼)
核心意义仅适合学习原理,写 100 行代码才能搭简单模型适合工程落地(比如做 APP),但新手入门难新手友好 + 灵活:几行代码就能搭模型,改参数 / 改结构超简单,是现在科研 / 入门的首选
具体到你的项目:

你的新冠预测模型如果用纯 Python 手写,需要自己写 “全连接层的矩阵乘法、梯度下降公式、反向传播”,至少要写 500 行;而用 PyTorch,只需要:

python

运行

class myNet(nn.Module):
    def __init__(self,inDim):
        super(myNet,self).__init__()
        self.fc1 = nn.Linear(inDim, 64)  # 一行代码搞定全连接层
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(64,1)
    def forward(self, x):
        return self.fc2(self.relu(self.fc1(x)))

👉 PyTorch 的核心意义:把深度学习中最复杂的 “矩阵计算、梯度求解、反向传播” 都封装好了,你只需要关注 “模型结构(用几层、几个神经元)”,不用管底层数学计算。

问题 2:逐个拆解导入语句(区别 + 作用 + 你的项目用法)

你觉得这些导入语句 “一摸一样”,是因为没分清 “包→子包→函数 / 类” 的层级,我逐个拆,结合你的代码讲:

导入语句层级关系核心区别具体作用你的项目里怎么用?
import matplotlib导入 matplotlib 主包只导入 “大框架”,不能直接用画图功能提供 matplotlib 的基础配置(比如设置字体、颜色)你代码里没直接用它,是plt的 “依赖”
import matplotlib.pyplot as plt导入 matplotlib 的子包pyplot,并简写为plt这是真正用来画图的核心子包画折线图、柱状图(比如训练 / 验证 loss 曲线)plt.plot(plt_train_loss)画训练 loss;plt.show()显示图
import torch导入 PyTorch 主包提供 PyTorch 的核心基础功能(张量、设备、保存模型)1. 定义张量(torch.tensor());2. 选设备(torch.cuda.is_available());3. 保存模型(torch.save()你的covidDataset里用torch.tensor()转数据;device选 GPU/CPU;train_val里用torch.save()存模型
import torch.nn as nn导入 PyTorch 的神经网络子包nn,简写为nn专门用来搭建神经网络的模块1. 定义层(nn.Linear全连接层、nn.ReLU激活函数);2. 定义损失函数(nn.MSELoss你的myNet里用nn.Linear搭网络;mseLoss里用nn.MSELoss算损失
from torch import optim从 PyTorch 导入优化器子包optim专门用来优化模型的模块(更新模型参数)提供优化器(optim.SGDoptim.Adam你的主流程里用optim.SGD(model.parameters(), lr=0.001)定义优化器,训练时用optimizer.step()更新模型
from torch.utils.data import Dataset,DataLoader从 PyTorch 的工具子包data中,导入两个类专门用来处理数据的工具:1. Dataset:自定义数据集的 “模板”;2. DataLoader:分批加载数据1. covidDataset继承Dataset,自定义数据处理逻辑;2. DataLoader把数据集分成批次(batch_size=256),方便模型分批学你的covidDataset类继承Dataset;主流程里用DataLoader(trainset, batch_size=256)打包训练集
from sklearn.decomposition import PCA从 sklearn 的降维子包导入 PCA 类主成分分析,用来减少特征数量(你代码里注释掉了,没用到)把 93 个特征压缩成少数几个 “综合特征”你代码里get_feature_importance_with_pca函数里用(注释了)
from sklearn.preprocessing import StandardScaler从 sklearn 的预处理子包导入标准化类数据标准化(和你代码里的归一化类似,你没直接用)把数据缩放到 “均值 0,标准差 1”你代码里是手动写的归一化,没用这个类
from sklearn.feature_selection import SelectKBest从 sklearn 的特征选择子包导入选特征的类专门用来筛选有用特征的工具选 k 个和标签最相关的特征你的get_feature_importance里用SelectKBest(chi2, k=4)选特征
from sklearn.feature_selection import chi2从 sklearn 导入卡方检验函数特征选择的 “评价标准”(判断特征是否有用)配合SelectKBest,用卡方检验打分选特征你的get_feature_importanceSelectKBest(chi2, k=4)chi2就是它
核心总结(导入语句的规律):
  • import A:导入整个包 A,用的时候要写A.xxx(比如torch.tensor());
  • import A.B as C:导入 A 包的子包 B,简写为 C,用的时候写C.xxx(比如nn.Linear());
  • from A import B:从 A 包导入 B(函数 / 类),用的时候直接写B(比如optim.SGD());
  • from A.B import C,D:从 A 包的子包 B 导入 C 和 D,直接用C/D(比如Dataset/DataLoader)。
def get_feature_importance(feature_data, label_data, k =4,column = None):
    """
    此处省略 feature_data, label_data 的生成代码。
    如果是 CSV 文件,可通过 read_csv() 函数获得特征和标签。
    这个函数的目的是, 找到所有的特征种, 比较有用的k个特征, 并打印这些列的名字。
    """
    model = SelectKBest(chi2, k=k)      #定义一个选择k个最佳特征的函数
    feature_data = np.array(feature_data, dtype=np.float64)
    # label_data = np.array(label_data, dtype=np.float64)
    X_new = model.fit_transform(feature_data, label_data)   #用这个函数选择k个最佳特征
    #feature_data是特征数据,label_data是标签数据,该函数可以选择出k个特征
    print('x_new', X_new)
    scores = model.scores_                # scores即每一列与结果的相关性
    # 按重要性排序,选出最重要的 k 个
    indices = np.argsort(scores)[::-1]        #[::-1]表示反转一个列表或者矩阵。
    # argsort这个函数, 可以矩阵排序后的下标。 比如 indices[0]表示的是,scores中最小值的下标。

    if column:                            # 如果需要打印选中的列
        k_best_features = [column[i+1] for i in indices[0:k].tolist()]         # 选中这些列 打印
        print('k best features are: ',k_best_features)
    return X_new, indices[0:k]                  # 返回选中列的特征和他们的下标。

通俗解释:
# 这个函数是 “挑有用的特征”—— 比如数据里有 93 列(州名、日期、气温、人口、当日感染数…),不是所有列都对 “预测第三天感染数” 有用,比如 “州名的拼音长度” 可能没用,这个函数会算出每一列和 “第三天感染数” 的相关性,挑出前 k 个最有用的列。
# 核心作用:
# SelectKBest(chi2, k=k):用卡方检验(不用懂原理)选 k 个最相关的特征; scores:每列的 “有用程度分数”,分数越高越有用;
#返回值:选中的 k 列数据 + 这些列在原始数据中的位置(比如第 3 列、第 8 列)。

class covidDataset(Dataset):
    def __init__(self, path, mode="train", feature_dim=5, all_feature=False):
        with open(path,'r') as f:
            csv_data = list(csv.reader(f))
            column = csv_data[0]
            x = np.array(csv_data)[1:,1:-1]
            y = np.array(csv_data)[1:,-1]
            if all_feature:
                col_indices = np.array([i for i in range(0,93)])                  # 若全选,则选中所有列。
            else:
                _, col_indices = get_feature_importance(x, y, feature_dim, column)      # 选重要的dim列。
                # X_new = get_feature_importance_with_pca(x, feature_dim, column)  # 选重要的特征
            col_indices = col_indices.tolist()             # col_indices 从array 转为列表。
            csv_data = np.array(csv_data[1:])[:,1:].astype(float)       #取csvdata从第二行开始, 第二列开始的数据,并转为float
            if mode == 'train':                                # 训练数据逢5选4, 记录他们的所在行
                indices = [i for i in range(len(csv_data)) if i % 5 != 0]
                self.y = torch.tensor(csv_data[indices,-1])      # 训练标签是csvdata的最后一列。 要转化为tensor型
            elif mode == 'val':               # 验证数据逢5选1, 记录他们的所在列
                indices = [i for i in range(len(csv_data)) if i % 5 == 0]
                # data = torch.tensor(csv_data[indices,col_indices])
                self.y = torch.tensor(csv_data[indices,-1])        # 验证标签是csvdata的最后一列。 要转化为tensor型
            else:
                indices = [i for i in range(len(csv_data))]     # 测试机只有数据
                # data = torch.tensor(csv_data[indices,col_indices])
            data = torch.tensor(csv_data[indices, :])           # 根据选中行取 X , 即模型的输入特征
            self.data = data[:, col_indices]                   #  col_indices 表示了重要的K列, 根据重要性, 选中k列。
            self.mode = mode                                   # 表示当前数据集的模式

            self.data = (self.data - self.data.mean(dim=0,keepdim=True)) / self.data.std(dim=0,keepdim=True)  # 对数据进行列归一化
            assert feature_dim == self.data.shape[1]                   # 判断数据的列数是否为规定的dim列, 要不然就报错。

            print('Finished reading the {} set of COVID19 Dataset ({} samples found, each dim = {})'
                  .format(mode, len(self.data), feature_dim))             # 打印读了多少数据

    def __getitem__(self, item):               # getitem 需要完成读下标为item的数据
        if self.mode == 'test':                  # 测试集没标签。   注意data要转为模型需要的float32型
            return self.data[item].float()
        else :                                  # 否则要返回带标签数据
            return self.data[item].float(), self.y[item].float()
    def __len__(self):
        return len(self.data)                 # 返回数据长度。

# 通俗解释:
# 这个类是 “数据管家”—— 把 CSV 文件里的原始数据,处理成模型能直接用的格式,核心做 4 件事:
# 读数据:把 CSV 里的文字转成数字;
# 选特征:只留有用的 k 列;
# 分数据集:
# 训练集:80% 数据(逢 5 选 4),用来教模型;
# 验证集:20% 数据(逢 5 选 1),用来测模型学得好不好;
# 测试集:无标签,用来最终预测;
# 归一化:模型怕 “数据尺度差太大”(比如感染数 1000、气温 20),归一化后数据尺度一致,模型学得更快。
# 核心作用:
# __init__:初始化时处理好所有数据;
# __getitem__:模型要第几个样本,就返回第几个(带标签 / 不带标签);
# __len__:告诉模型 “一共有多少样本”。


class myNet(nn.Module):
    def __init__(self,inDim):
        super(myNet,self).__init__()
        self.fc1 = nn.Linear(inDim, 64)              # 全连接
        self.relu = nn.ReLU()                        # 激活函数
        self.fc2 = nn.Linear(64,1)                     # 全连接

    def forward(self, x):                     #forward, 即模型前向过程
        x = self.fc1(x)
        x = self.relu(x)

        x = self.fc2(x)
        if len(x.size()) > 1:
            return x.squeeze(1)
        else:
            return x
# 通俗解释:
# 这是模型的 “大脑”,非常简单的两层全连接网络:
# 输入:k 个特征(比如 k=6,就是 6 个数字:当日感染数、气温、人口…);
# 第一层:把 6 个输入变成 64 个中间结果(神经元计算);
# ReLU:激活函数 —— 相当于 “筛选有用的中间结果”(不用懂原理,新手默认用);
# 第二层:把 64 个中间结果变成 1 个输出(就是预测的 “第三天感染数”)。
# 核心作用:
# nn.Linear(inDim, 64):全连接层,做简单的线性计算;
# forward:定义模型的计算流程(输入→第一层→ReLU→第二层→输出);
# 最终输出 1 个数字:预测的第三天感染数(回归任务的核心)。




def train_val(model, trainloader, valloader,optimizer, loss, epoch, device, save_):

    # trainloader = DataLoader(trainset,batch_size=batch,shuffle=True)
    # valloader = DataLoader(valset,batch_size=batch,shuffle=True)
    model = model.to(device)                # 模型和数据 ,要在一个设备上。
    plt_train_loss = []
    plt_val_loss = []
    val_rel = []
    min_val_loss = 100000                 # 记录训练验证loss 以及验证loss和结果

    for i in range(epoch):                 # 训练epoch 轮
        start_time = time.time()             # 记录开始时间
        model.train()                         # 模型设置为训练状态
        train_loss = 0.0
        val_loss = 0.0

        for data in trainloader:                     # 从训练集取一个batch的数据
            optimizer.zero_grad()                   # 梯度清0
            x , target = data[0].to(device), data[1].to(device)       # 将数据放到设备上
            pred = model(x)                          # 用模型预测数据
            bat_loss = loss(pred, target, model)       # 计算loss
            bat_loss.backward()                        # 梯度回传, 反向传播。
            optimizer.step()                            #用优化器更新模型。
            train_loss += bat_loss.detach().cpu().item()             #记录loss和

        plt_train_loss. append(train_loss/trainloader.dataset.__len__())
        #记录loss到列表。注意是平均的loss ,因此要除以数据集长度。

        model.eval()                 # 模型设置为验证状态
        with torch.no_grad():                    # 模型不再计算梯度
            for data in valloader:                      # 从验证集取一个batch的数据
                val_x , val_target = data[0].to(device), data[1].to(device)          # 将数据放到设备上
                val_pred = model(val_x)                 # 用模型预测数据
                val_bat_loss = loss(val_pred, val_target, model)          # 计算loss
                val_loss += val_bat_loss.detach().cpu().item()                  # 计算loss
                val_rel.append(val_pred)                 #记录预测结果
        if val_loss < min_val_loss:
            torch.save(model, save_)               #如果loss比之前的最小值小, 说明模型更优, 保存这个模型

        plt_val_loss.append(val_loss/valloader.dataset.__len__())  #记录loss到列表。注意是平均的loss ,因此要除以数据集长度。
        #
        print('[%03d/%03d] %2.2f sec(s) TrainLoss : %.6f | valLoss: %.6f' % \
              (i, epoch, time.time()-start_time, plt_train_loss[-1], plt_val_loss[-1])
              )              #打印训练结果。 注意python语法, %2.2f 表示小数位为2的浮点数, 后面可以对应。
        # print('[%03d/%03d] %2.2f sec(s) TrainLoss : %3.6f | valLoss: %.6f' % \
        #       (i, epoch, time.time()-start_time, 2210.2255411, plt_val_loss[-1])
        #       )              #打印训练结果。 注意python语法, %2.2f 表示小数位为2的浮点数, 后面可以对应。
    plt.plot(plt_train_loss)              # 画图, 向图中放入训练loss数据
    plt.plot(plt_val_loss)                # 画图, 向图中放入训练loss数据
    plt.title('loss')                      # 画图, 标题
    plt.legend(['train', 'val'])             # 画图, 图例

    plt.show()                                 # 画图, 展示
# 通俗解释:
# 这个函数是 “教模型学知识” 的全过程,分两大步:
# 训练:模型用训练集学规律,每学一批数据就更新一次参数(相当于做一次练习题,订正一次答案);
# 验证:每学完一轮(epoch),用验证集测一测模型学得好不好,若效果更好就保存模型;
# 最后画 loss 曲线:训练 loss 和验证 loss 都下降,说明模型学得好;如果验证 loss 上升,说明模型 “学傻了”(过拟合)。
# 核心关键点(新手先记):
# model.train()/model.eval():训练 / 验证模式切换(验证时不更新模型);
# optimizer.zero_grad():梯度清零(PyTorch 必须做的步骤);
# bat_loss.backward():反向传播(模型 “反思” 哪里错了);
# optimizer.step():更新模型(模型 “改正错误”);
# torch.save(model, save_):保存效果最好的模型(后续预测用)。




def evaluate(model_path, testset, rel_path ,device):
    model = torch.load(model_path).to(device)                     # 模型放到设备上。
    testloader = DataLoader(testset,batch_size=1,shuffle=False)         # 将验证数据放入loader 验证时, 一般batch为1
    val_rel = []
    model.eval()               # 模型设置为验证状态
    with torch.no_grad():               # 模型不再计算梯度
        for data in testloader:                 # 从测试集取一个batch的数据
            x = data.to(device)                # 将数据放到设备上
            pred = model(x)                        # 用模型预测数据
            val_rel.append(pred.item())                #记录预测结果
    print(val_rel)                                     #打印预测结果
    with open(rel_path, 'w') as f:                        #打开保存的文件
        csv_writer = csv.writer(f)                           #初始化一个写文件器 writer
        csv_writer.writerow(['id','tested_positive'])         #在第一行写上 “id” 和 “tested_positive”
        for i in range(len(testset)):                           # 把测试结果的每一行放入输出的excel表中。
            csv_writer.writerow([str(i),str(val_rel[i])])
    print("rel已经保存到"+ rel_path)
# 通俗解释:
# 这个函数是 “用训练好的模型干活”—— 加载之前保存的最优模型,对测试集(没有标签的数据)做预测,然后把预测结果保存到 CSV 文件里(格式是 id + 预测的感染数)。
# 核心作用:
# torch.load(model_path):加载最优模型;
# model.predict(x):预测测试集的每个样本;
# 保存结果:把预测值写成 CSV,方便查看 / 提交。

all_col = False              #是否使用所有的列
device = 'cuda' if torch.cuda.is_available() else 'cpu'       #选择使用cpu还是gpu计算。
print(device)
train_path = 'covid.train.csv'                     # 训练数据路径
test_path = 'covid.test.csv'              # 测试数据路径

file = pd.read_csv(train_path)
file.head()                    # 用pandas 看看数据长啥样

if all_col == True:
    feature_dim = 93
else:
    feature_dim = 6              #是否使用所有的列

trainset = covidDataset(train_path,'train',feature_dim=feature_dim, all_feature=all_col)
valset = covidDataset(train_path,'val',feature_dim=feature_dim, all_feature=all_col)
testset = covidDataset(test_path,'test',feature_dim=feature_dim, all_feature=all_col)   #读取训练, 验证,测试数据

通俗解释:
这是 “准备数据” 的主流程:
选计算设备(有 GPU 用 GPU,没 GPU 用 CPU);
设定用 6 个特征(不是 93 个);
用自定义的covidDataset类,加载训练集、验证集、测试集。
核心作用:
把数据文件转成模型能直接用的数据集对象。

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           # 定义mseloss 即 平方差损失,

config = {
    'n_epochs': 50,                # maximum number of epochs
    'batch_size': 256,               # mini-batch size for dataloader
    'optimizer': 'SGD',              # optimization algorithm (optimizer in torch.optim)
    'optim_hparas': {                # hyper-parameters for the optimizer (depends on which optimizer you are using)
        'lr': 0.0001,                 # learning rate of SGD
        'momentum': 0.9              # momentum for SGD
    },
    'early_stop': 200,               # early stopping epochs (the number epochs since your model's last improvement)
    'save_path': 'model_save/model.pth',  # your model will be saved here
}

model = myNet(feature_dim).to(device)                      # 实例化模型
optimizer = optim.SGD(model.parameters(), lr=0.001,momentum=0.9)             # 定义优化器
trainloader = DataLoader(trainset, batch_size=config['batch_size'],shuffle=True)
valloader = DataLoader(valset, batch_size=config['batch_size'],shuffle=True)  # 将数据装入loader 方便取一个batch的数据

train_val(model, trainloader, valloader, optimizer, loss, config['n_epochs'], device,save_=config['save_path'])  # 训练


evaluate(config['save_path'], testset, 'pred.csv', device)           # 验证
通俗解释:
这是整个项目的 “最后一步”—— 把所有部件拼起来运行:
定义损失函数(MSE:预测值和真实值的均方误差,越小越好),加了 L2 正则(防止模型 “死记硬背” 训练集,不会举一反三);
设定训练参数(训练 50 轮,每批 256 个样本);
初始化模型和优化器(SGD:随机梯度下降,更新模型参数);
调用train_val训练模型,调用evaluate做预测。
核心作用:
启动训练和预测,输出最终的预测结果文件pred.csv。

更多推荐