一个简单的深度学习模型，通常包含三部分：

1、数据集Dataset：一般输入是文件地址，或者数据内容， 输出是一个 存储了数据X,Y的数据结构。 torch中一般用dataloader 来装载。

基础功能包含下面三个：

1）init（初始化）

2）getitem（切片）

3）len（得到长度）

2、模型Model：定义自己的模型 输入X， 输出预测值

基础功能包含下面两个：

1）init（初始化）

2）forward（前向过程）

3、超参hyperPara：除模型外的超参 一般包含： 学习率，优化器， 损失函数等

一、自定义数据集

        在数据部分，我们迎来了第一个可以优化的点，原数据集是有很多特征的，但是不同的特征的影响差别是很大的，如果我们考虑了所有的特征，这也意味着我们加入了许多无关、冗余、甚至带有噪声的特征，会干扰模型的学习性能。而且当特征太多而样本太少时，容易导致过拟合的问题，减少特征可以提高泛化能力。

        由于我也是一个初学者，在这里记录一下选取k个最相关特征的方法。

        这里我们的任务是预测 tested_positive，这是一个回归问题，因此采取F回归的方法

def get_feature_importance(feature_data, label_data, k=4, column=None):
    """
    此处省略 feature_data, label_data 的生成代码。
    如果是 CSV 文件，可通过 read_csv() 函数获得特征和标签。
    这个函数的目的是， 找到所有的特征种， 比较有用的k个特征， 并打印这些列的名字。
    """
    model = SelectKBest(f_regression, 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]                  # 返回选中列的特征和他们的下标。

        下面开始写读取数据的代码。

        我们需要做的是首先读取csv文件；然后需要划分出训练模式、验证模式、测试模式下不同的数据集 。把一部分数据用于训练，一部分用于验证，在这两种模式下需要有特征x和标签y。所有的数据只保留特征作为测试模式下的数据；下面对数据进行标准化。下面进行具体的介绍。

        这里我们把这个类定义为CovidDataset，继承Data的属性和方法。还记得我们最开始说的三个基础功能吗？下面就开始写这几个函数了。

首先是初始化函数__init__，下面是函数内部细节

1、读取csv文件

        with open(file_path, "r") as f:
            ori_data = list(csv.reader(f))
            column = ori_data[0]
            csv_data = np.array(ori_data[1:])[:, 1:].astype(float)

        我们用只读的方式打开该路径下的文件，同时把返回的文件对象赋值给变量 f 。

        然后以二维列表的形式读取原始数据ori_data，其中的ori_data[0]就是原始数据的第一行，即表头部分（所有列的名称），后面就是熟悉的操作，之前我们在手写线性模型中也写过的操作，通过删掉第一行，再转换为矩阵，再将矩阵的第一列（序号）删掉，就得到了真正的数据部分。

        这里值得注意的是，csv文件中默认存储的是字符串，但是pytorch只能处理数值型数据，而且要求是浮点数（不能是字符串和整数），所以最后我们必须通过 .astype(float) 的方式，把数据转化为浮点数。

2、分离特征和标签（用于特征选择）

        feature = np.array(ori_data[1:])[:, 1:-1]
        label_data = np.array(ori_data[1:])[:, -1]

        原始数据去掉第一行（列名）、第一列（id）、最后一列（标签），得到纯特征feature

        去掉第一行（列名），只保留最后一列，得到标签label_data

3、特征选择（使用之前写的函数，得到最相关特征）

        if all_feature:
            col = np.array([i for i in range(0, 93)])
        else:
            _, col = get_feature_importance(feature, label_data, feature_dim, column)
        col = col.tolist()

        这里我们支持两种模式，一种是使用所有的93个特征，一种是使用最相关的k个特征（默认是6）.

        这里需要注意，我们调用 get_feature_importance 函数的唯一目的是：拿到最重要的特征列的索引（即 col），以便后续从原始数据中选出这些列。但是函数的返回值有两个，一个是经过特征选择之后的新特征矩阵（只保留 top-k 特征的数据），另一个是被选中的 k 个特征在原始数据中的列索引。其中前者我们这里用不到，所有我们用 _, col 的形式来接收，表示第一个数据我们不关心（忽略），只想要col。这是一种忽略不需要的返回值的惯用写法。

4、根据模式mode划分数据

        这里就很简单了，按照我们之前讲过的逻辑即可，不做过多介绍。

        if mode == "train":        #逢5取1.
            indices = [i for i in range(len(csv_data)) if i % 5 != 0]
            data = torch.tensor(csv_data[indices, :-1])
            self.y = torch.tensor(csv_data[indices, -1])
        elif mode == "val":
            indices = [i for i in range(len(csv_data)) if i % 5 == 0]
            data = torch.tensor(csv_data[indices, :-1])
            self.y = torch.tensor(csv_data[indices, -1])
        else:
            indices = [i for i in range(len(csv_data))]
            data = torch.tensor(csv_data[indices])

5、只保留选中的特征列

        data = data[:, col]

6、标准化（Z-score Normalization）

        由于原始数据的不同特征值有着不同的量度，所有我们需要通过标准化把他们的数值范围变得差不多（均值为0、标准差为1），防止某些特征因为数值大而“主导”模型。具体操作就是让数据减去均值再除标准差。

        self.data = (data- data.mean(dim=0, keepdim=True))/data.std(dim=0, keepdim=True)
        self.mode = mode

        这里我们还捎带着把标准化后的数据以及模式都变成self的一个属性（也可以说是这个对象的一个数据），保存到这个对象身上，以便我们再其它函数上可以直接使用。

下面是第二个函数__getitem__，用于按索引取样本，例如我们写 dataset[0]时，就是在调用这个函数。

        这里我们需要做的就是按照不同的模式，给不同的样本，因为训练和验证模式下需要 x 和 y，而测试模式下只需要 x 。

    def __getitem__(self, idx):
        if self.mode != "test":
            return self.data[idx].float(), self.y[idx].float()     
        else:
            return self.data[idx].float()

        我们之前不是已经转化为float类型了吗？为什么又要转化呢？是否多此一举呢？

        不是的！需要注意的是，在 csv_data 那里，我们是在NumPy下的矩阵中转换，而在NumPy中默认是float64，在torch.tensor中，如果不指定dtype，会保留NumPy的dtype。而PyTorch 模型（尤其是 GPU 上的模型）默认使用 float32（也叫 torch.float）。所有我们这里再次指定类型，是确保返回的数据都是 float32 。

下面是最后一个函数__len__，用于返回数据集的样本总数

    def __len__(self):
        return len(self.data)

二、模型定义

        先定义一个类MyModel继承自nn.Module，下面开始实现我们的两个基础功能。

        第一部分：模型定义（__init__）：

        这里我们定义一个简单的神经网络用于完成回归任务。采用最经典的结构：

输入 → 全连接层 → 激活函数 → 全连接层 → 输出

    def __init__(self, inDim):
        super(MyModel, self).__init__()
        self.fc1 = nn.Linear(inDim, 64)
        self.relu1 = nn.ReLU()
        self.fc2 = nn.Linear(64, 1)

        第二部分：前向传播（forward）

    def forward(self, x):   
        x = self.fc1(x)
        x = self.relu1(x)
        x = self.fc2(x)
        if len(x.size()) > 1:
            return x.squeeze(1)     
        return x

        这里需要注意，虽然模型的最后一层是nn.Linear(64, 1)，但其实输出的是一个二维的张量。什么意思呢？对于n个样本来说，我们输出的结果是一个n行1列的张量。而不是简单的n个预测值。

        具体来说，我们模型输出的是形状是 [ batch , outputs ] ，但我们要的预测标签的形状是           [ batch ] 。所以我们需要通过 .squeeze(1) 实现去掉第1维，只留第0维。

三、超参部分

这里我们实现了损失函数的定义，并且通过加入正则项，实现对损失函数的优化。

什么是正则化呢？正则化（Regularization） 是机器学习和深度学习中一种防止模型过拟合、提升泛化能力的核心技术。

具体我们在这里不去深究（后面会另写一篇文章中集中解释我对正则化的理解），只需要知道，我们这里通过L2正则化，防止模型过拟合，提高模型泛化能力。

定义一个带正则项的均方误差 mseLoss_with_reg 

1、创建标准损失，也就是pytorch中内置的均方误差

    loss = nn.MSELoss(reduction='mean')

2、初始化正则项

    regularization_loss = 0                  

3、遍历所以模型参数，计算L2正则项

    for param in model.parameters():
        regularization_loss += torch.sum(param ** 2)                  

其中 model.parameters() 是 PyTorch 中的一个方法，它会返回模型中所有可学习的参数（张量）。

公式就是：     

循环遍历每一曾的参数的（weight和bias），把他们的平方和加起来。

4、组合总损失

    return loss(pred, target) + 0.00075 * regularization_loss

这里的 λ 取0.00075，λ即正则化强度，越大就越惩罚大权重，模型就越平滑越简单。

四、训练部分

需要实现整个深度学习训练流程的核心骨架，实现“训练 + 验证 + 模型保存 + 可视化” 的完整闭环。

1、定义函数并初始化

def train_val(model, train_loader, val_loader, device, epochs, optimizer, loss, save_path):
    model = model.to(device)

    plt_train_loss = []
    plt_val_loss = []

    min_val_loss = 9999999999999

• 将模型移到指定设备（GPU/CPU），确保后续计算在正确设备上进行。
• 初始化两个列表，用于记录每轮的平均训练/验证损失（画图用）
• min_val_loss 初始化为极大值，用于跟踪历史最低验证损失（实现“保存最佳模型”）

2、主训练循环

吹响胜利的号角！！！

    for epoch in range(epochs):

每一轮做两件事“训练+验证”

1）训练阶段

首先重置本轮训练的损失和计时器并把模式调整为训练模式，后面就是一系列小连招

        train_loss = 0.0    
        start_time = time.time()
        model.train()   #模型调整为训练模式
        for batch_x, batch_y in train_loader:
            x, target = batch_x.to(device), batch_y.to(device)    #把数据搬迁到训练设备上
            pred = model(x)    #前向传播
            train_bat_loss = loss(pred, target, model)    #计算损失
            train_bat_loss.backward()    #反向传播
            optimizer.step()        #更新模型
            optimizer.zero_grad()    #清零梯度
            train_loss += train_bat_loss.cpu().item()    #记录损失把 loss 标量从 GPU 移回 CPU，转成普通 Python 数字，累加到 train_loss
        plt_train_loss.append(train_loss / train_loader.__len__())    #计算本轮平均每个 batch 的训练损失（近似 epoch 平均损失）

2）验证阶段

整个过程和训练阶段类似，但是不需要计算梯度，因此没有反向传播、更新模型、清零梯度的过程，只计算损失，用于评估模型在未见数据上的表现。

        val_loss = 0.0
        model.eval()    #验证模式
        with torch.no_grad():    #验证过程不需要计算梯度
            for batch_x, batch_y in val_loader:
                x, target = batch_x.to(device), batch_y.to(device)    #把数据搬迁到验证设备上
                pred = model(x)    #前向过程
                val_bat_loss = loss(pred, target, model)    #计算每一批数据的验证损失
                val_loss += val_bat_loss.cpu().item()    #把当前验证损失放到cpu上，转化为普通 Python 数字，并累加
        plt_val_loss.append(val_loss / val_loader.__len__())    #记录平均损失

3）保存最佳模型

如果当前验证损失比历史最低还低，就保存这个模型。

这是防止过拟合的关键：即使后面训练 loss 继续下降，只要 val loss 上升，就不保存

        if val_loss < min_val_loss:
            torch.save(model, save_path)
            min_val_loss = val_loss

        print("[%03d/%03d] %2.2f sec(s) Trainloss： %.6f |Valloss: %.6f" % \
              (epoch, epochs, time.time()-start_time, plt_train_loss[-1], plt_val_loss[-1]))    #输出当前 epoch 的进度、耗时、训练/验证损失，方便实时监控训练状态

4）可视化损失曲线

    plt.plot(plt_train_loss)
    plt.plot(plt_val_loss)
    plt.title("loss")
    plt.legend(["train", "val"])
    plt.show()

通过曲线可以判断：

• 是否收敛（loss 是否下降）
• 是否过拟合（val loss 上升而 train loss 下降）
• 是否欠拟合（两者都高）

五、开始训练！！！

首先，我们需要配置一些参数。

使用特征选取，选取6个最相关的特征

all_feature = False
if all_feature:
    feature_dim = 93
else:
    feature_dim = 6

给出数据集的路径

train_file = "covid.train.csv"
test_file = "covid.test.csv"

给出训练的一批数据量

batchsize = 16

配置参数

config = {
    "lr": 0.001,    #学习率
    "epochs": 20,    #训练轮数
    "momentum": 0.9,    #动量
    "save_path":"model_save/best_model.pth",    #最优模型保存路径
    "rel_path":"pred.csv"    #预测值保存路径
}

调用函数获得各个模式下的数据集

train_dataset = CovidDataset(train_file, "train", feature_dim=feature_dim, all_feature=all_feature)
val_dataset = CovidDataset(train_file, "val", feature_dim=feature_dim, all_feature=all_feature)
test_dataset = CovidDataset(test_file, "test", feature_dim=feature_dim, all_feature=all_feature)

   

使用 PyTorch 的 DataLoader 类，为训练集、验证集和测试集分别创建了数据加载器（DataLoader），用于在训练和评估过程中高效、批量地读取数据

train_loader = DataLoader(train_dataset, batch_size=batchsize, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=batchsize, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=1, shuffle=False)     #测试集的数据不能打乱

        注意训练集和验证集都是按批次准备数据，并且为了让每个batch的样本具有随机性和代表性，需要打乱数据，但是测试集必须一个一个来，而且由于我们要按顺序输出预测结果，所有测试集也不能打乱。

定义设备、创建模型并把模型放到设备上、创建优化器

device = "cuda" if torch.cuda.is_available() else "cpu"
print(device)
model = MyModel(inDim=feature_dim).to(device)
optimizer = optim.SGD(model.parameters(), lr= config["lr"],momentum= config["momentum"])

定义损失函数

loss = mseLoss_with_reg

启动训练过程

train_val(model, train_loader, val_loader, device, config["epochs"], optimizer, loss, config["save_path"])

至此，整个模型就训练完成了。

六、用训练好的模型对测试集进行预测，并生成标准格式的提交文件

定义一个预测函数

def evaluate(save_path, test_loader, device, rel_path):

1、加载模型，并移动到设备上。

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

• 从磁盘加载整个模型对象（包括网络结构 + 权重参数）
• weights_only=False 是 PyTorch ≥2.0 的安全选项，表示允许加载完整模型（而不仅是权重）

torch.load()可以用来加载任何用 torch.save() 保存的 Python 对象（如张量、优化器状态、字典等）。

2、预测过程

初始化一个空列表，用于存储预测结果。

禁用梯度计算，把数据搬迁到设备上，然后开始前向过程得到预测值pred。

注意pred得到的仍然是张量，只不过是一个形状为 [1] 的张量。我们再把pred放回cpu并转化为普通python数字，最后把预测值加入到结果列表中。

    rel = []
    with torch.no_grad():
        for x in test_loader:
            pred = model(x.to(device))
            rel.append(pred.cpu().item())

3、保存为 CSV 提交文件

以“写”模式打开csv文件，注意这里有一个参数 newline=' ' ，他的作用是关闭了 Python 的自动换行符转换，防止在写入 CSV 文件时，在 Windows 系统上产生多余的空行。

    print(rel)
    with open(rel_path, "w", newline='')as f:
        csvWriter = csv.writer(f)
        csvWriter.writerow(["id", "tested_positive"])    #写入表头
        for i, value in enumerate(rel):                  #写入每行的预测结果
            csvWriter.writerow([str(i), str(value)])
    print("文件已经保存到{}".format(rel_path))

最后启动预测函数就可以了

evaluate(config["save_path"], test_loader, device, config["rel_path"])

下面给出完整实现代码，代码顺序与文章顺序稍有不同，但内容一致。

import matplotlib
matplotlib.use("QtAgg")
import matplotlib.pyplot as plt
import torch
import numpy as np
import csv
from torch.utils.data import DataLoader, Dataset
import torch.nn as nn
from torch import optim
import time
from sklearn.feature_selection import SelectKBest, chi2, f_regression


def get_feature_importance(feature_data, label_data, k=4, column=None):
    """
    此处省略 feature_data, label_data 的生成代码。
    如果是 CSV 文件，可通过 read_csv() 函数获得特征和标签。
    这个函数的目的是， 找到所有的特征种， 比较有用的k个特征， 并打印这些列的名字。
    """
    model = SelectKBest(f_regression, 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]                  # 返回选中列的特征和他们的下标。

class CovidDataset(Dataset):
    def __init__(self, file_path, mode="train", all_feature=True, feature_dim=6):
        with open(file_path, "r") as f:
            ori_data = list(csv.reader(f))
            column = ori_data[0]
            csv_data = np.array(ori_data[1:])[:, 1:].astype(float)
        feature = np.array(ori_data[1:])[:, 1:-1]
        label_data = np.array(ori_data[1:])[:, -1]
        if all_feature:
            col = np.array([i for i in range(0, 93)])
        else:
            _, col = get_feature_importance(feature, label_data, feature_dim, column)
        col = col.tolist()
        if mode == "train":        #逢5取1.
            indices = [i for i in range(len(csv_data)) if i % 5 != 0]
            data = torch.tensor(csv_data[indices, :-1])
            self.y = torch.tensor(csv_data[indices, -1])
        elif mode == "val":
            indices = [i for i in range(len(csv_data)) if i % 5 == 0]
            data = torch.tensor(csv_data[indices, :-1])
            self.y = torch.tensor(csv_data[indices, -1])
        else:
            indices = [i for i in range(len(csv_data))]
            data = torch.tensor(csv_data[indices])
        data = data[:, col]
        self.data = (data- data.mean(dim=0, keepdim=True))/data.std(dim=0, keepdim=True)
        self.mode = mode
    def __getitem__(self, idx):
        if self.mode != "test":
            return self.data[idx].float(), self.y[idx].float()      #改变为32位float，减少模型消耗
        else:
            return self.data[idx].float()

    def __len__(self):
        # 返回数据集的样本总数
        return len(self.data)


#模型部分
class MyModel(nn.Module):
    def __init__(self, inDim):
        super(MyModel, self).__init__()
        self.fc1 = nn.Linear(inDim, 64)
        self.relu1 = nn.ReLU()
        self.fc2 = nn.Linear(64, 1)

    def forward(self, x):   #前向过程
        x = self.fc1(x)
        x = self.relu1(x)
        x = self.fc2(x)
        if len(x.size()) > 1:
            return x.squeeze(1)     #去掉1维
        return x

def train_val(model, train_loader, val_loader, device, epochs, optimizer, loss, save_path):
    model = model.to(device)

    plt_train_loss = []
    plt_val_loss = []

    min_val_loss = 9999999999999

    for epoch in range(epochs):
        train_loss = 0.0
        val_loss = 0.0
        start_time = time.time()

        model.train()   #模型调整为训练模式
        for batch_x, batch_y in train_loader:
            x, target = batch_x.to(device), batch_y.to(device)
            pred = model(x)
            train_bat_loss = loss(pred, target, model)
            train_bat_loss.backward()
            optimizer.step()        #更新模型
            optimizer.zero_grad()
            train_loss += train_bat_loss.cpu().item()

        plt_train_loss.append(train_loss / train_loader.__len__())

        model.eval()    #验证模式
        with torch.no_grad():
            for batch_x, batch_y in val_loader:
                x, target = batch_x.to(device), batch_y.to(device)
                pred = model(x)
                val_bat_loss = loss(pred, target, model)
                val_loss += val_bat_loss.cpu().item()
        plt_val_loss.append(val_loss / val_loader.__len__())
        if val_loss < min_val_loss:
            torch.save(model, save_path)
            min_val_loss = val_loss

        print("[%03d/%03d] %2.2f sec(s) Trainloss： %.6f |Valloss: %.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()

def evaluate(save_path, test_loader, device, rel_path):
    model = torch.load(save_path, weights_only=False).to(device)
    rel = []
    with torch.no_grad():
        for x in test_loader:
            pred = model(x.to(device))
            rel.append(pred.cpu().item())
    print(rel)
    with open(rel_path, "w", newline='')as f:
        csvWriter = csv.writer(f)
        csvWriter.writerow(["id", "tested_positive"])
        for i, value in enumerate(rel):
            csvWriter.writerow([str(i), str(value)])
    print("文件已经保存到{}".format(rel_path))

all_feature = False
if all_feature:
    feature_dim = 93
else:
    feature_dim = 6


train_file = "covid.train.csv"
test_file = "covid.test.csv"

train_dataset = CovidDataset(train_file, "train", feature_dim=feature_dim, all_feature=all_feature)
val_dataset = CovidDataset(train_file, "val", feature_dim=feature_dim, all_feature=all_feature)
test_dataset = CovidDataset(test_file, "test", feature_dim=feature_dim, all_feature=all_feature)


batchsize = 16
train_loader = DataLoader(train_dataset, batch_size=batchsize, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=batchsize, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=1, shuffle=False)     #测试集的数据不能打乱
# for batch_x, batch_y in train_loader:
#     print(batch_x ,batch_y)




device = "cuda" if torch.cuda.is_available() else "cpu"
print(device)

def mseLoss_with_reg(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             # 返回损失。



config = {
    "lr": 0.001,
    "epochs": 20,
    "momentum": 0.9,
    "save_path":"model_save/best_model.pth",
    "rel_path":"pred.csv"
}

model = MyModel(inDim=feature_dim).to(device)
loss = mseLoss_with_reg
optimizer = optim.SGD(model.parameters(), lr= config["lr"],momentum= config["momentum"])

train_val(model, train_loader, val_loader, device, config["epochs"], optimizer, loss, config["save_path"])

evaluate(config["save_path"], test_loader, device, config["rel_path"])