回归预测代码实战

引言

在机器学习领域，回归分析是一种非常重要的方法，常用于预测连续型变量。本次实战项目以新冠模型预测(ML2021Spring-hw1)为例，展示如何进行数据处理、模型定义、训练以及预测。

项目环境与依赖库

1
2
3
4
5
6
7
8
9
10
11
12
import time

import torch
import matplotlib.pyplot as plt
import csv
import numpy as np
import pandas as pd
from torch.utils.data import DataLoader, Dataset
import torch.nn as nn  # 引入NN模型
import torch.optim as optim

from sklearn.feature_selection import SelectKBest, chi2
  • torch：深度学习框架，用于构建和训练神经网络模型。
  • matplotlib.pyplot：用于数据可视化。
  • csv：用于处理 CSV 文件。
  • numpy：用于数值计算。
  • pandas：用于数据处理和分析。
  • torch.utils.data：提供数据加载和处理工具。
  • torch.nn：包含神经网络的各种层和损失函数。
  • torch.optim：包含优化器。
  • sklearn.feature_selection：用于特征选择。

数据处理部分

数据集划分

数据集的合理划分对于模型的训练和评估至关重要。两种划分方式：

数据集划分一：训练集占 80%，测试集 (不用于训练) 占 20%

这种划分方法是将整个数据集按照 8:2 的比例划分为训练集和测试集。

  • 训练集：用于模型的训练，即通过训练集的数据和标签来调整模型的参数，会累计梯度更新模型
  • 测试集：不参与模型的训练过程，主要用于评估模型在未见过的数据上的泛化能力。通过将模型在测试集上进行预测，并与测试集的真实标签进行比较，可以得到模型的性能指标，如准确率、均方误差等。
    这种划分方法比较简单直接，适用于数据量不是特别小的情况。但它没有专门的验证集，可能无法很好地进行模型选择和超参数调整。

数据集划分二：训练集占 70%，验证集 (用于训练之后验证，不更新模型) 占 10%，测试集 (用于最后验证) 占 20%

这种划分方法将整个数据集按照 7:1:2 的比例划分为训练集、验证集和测试集。

  • 训练集：用于模型的训练
  • 验证集：在模型训练过程中，用于验证模型的性能。在训练过程中，可以每隔一定的训练轮次（epoch），使用验证集来评估模型的性能，根据验证集上的性能指标来调整模型的超参数（如学习率、正则化参数等），选择最优的模型。验证集的作用是避免模型在训练集上过度拟合，帮助我们找到泛化能力较好的模型。
  • 测试集：和第一种划分方法中的测试集作用相同，用于最终评估模型在未见过的数据上的性能。测试集的数据在整个模型训练和验证过程中都不会被使用，这样可以得到一个相对客观的模型性能评估结果。
    这种划分方法更加细致，通过引入验证集，可以更好地进行模型选择和超参数调整，提高模型的泛化能力。适用于数据量相对较大的情况，因为划分出验证集后，训练集的数据量会相应减少，如果数据量本身就很小，可能会导致训练不充分。

CovidDataSet类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class CovidDataset(Dataset):  
    def __init__(self, file_path, mode="train"):  
        with open(file_path, 'r') as f:  
            ori_data = list(csv.reader(f))  
            # 去掉第一行第一列  
            csv_data = np.array(ori_data[1:])[:, 1:].astype(float)  
        self.mode = mode  
        # 划分数据集  
        if mode == "train":  # 逢5取1  
            # 筛选训练集  
            indices = [i for i in range(len(csv_data)) if i % 5 != 0]  
            # 训练集标签  
            self.y = torch.tensor(csv_data[indices, -1])  
            data = torch.tensor(csv_data[indices, :-1])  
        elif mode == "val":  
            # 筛选验证集  
            indices = [i for i in range(len(csv_data)) if i % 5 != 0]  
            # 验证集标签  
            self.y = torch.tensor(csv_data[indices, -1])  
            data = torch.tensor(csv_data[indices, :-1])  
        else:  
            # 筛选测试集  
            indices = [i for i in range(len(csv_data))]  
            data = torch.tensor(csv_data[indices])  
        # 归一化: 消除部分平行数据之间量纲的影响  
        # （x - x的平均值） / x的标准差  
        self.data = (data - data.mean(dim=0, keepdim=True)) / data.std(dim=0, keepdim=True)  
  
    def __getitem__(self, idx):  
        if self.mode != "test":  
            # 训练集和验证集要返回标签和数据 加入.float() 是因为data的类型时float64将其转化为float32 64位消耗大  
            return self.data[idx].float(), self.y[idx].float()  
        else:  
            # 测试集只返回数据  
            return self.data[idx].float()  
  
    def __len__(self):  
        return len(self.data)
  • 数据读取：使用csv.reader读取 CSV 文件内容，将其转换为列表形式的ori_data。之后，利用np.array将数据转换为 NumPy 数组，并去掉第一行第一列(第一行为列名，第一列为id，与数据无关需要删除，部分数据如下图所示)，其次NumPy数组元素是字符串，使用.astype(float)转化为浮点型，
  • 数据集划分：mode为train时，选择下标为index % 5 != 0的数据项作为训练集，mode为val时，选择下标为index % 5 == 0的数据项作为验证集，mode为test时，选择全部数据作为测试集
  • 数据归一化：使用公式(data - data.mean(dim=0, keepdim=True)) / data.std(dim=0, keepdim=True)对数据进行归一化处理。计算数据在每个特征维度上的均值和标准差，将每个数据点减去对应维度的均值后再除以标准差，从而使数据具有统一的尺度，消除量纲影响，提升模型训练效果。

数据加载

1
2
3
4
5
6
7
8
9
10
11
12
train_file = "covid.train.csv"
test_file = "covid.test.csv"

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

# 批次训练
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)
  • 注意测试集不可以打乱数据

定义模型

Model类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
### 定义模型部分
class Model(nn.Module):
    # in_dim 输入维度, out_dim 输出维度此处固定为1
    def __init__(self, in_dim):
        # 使用父类初始化
        super(Model, self).__init__()
        # 定义全连接层
        self.fc1 = nn.Linear(in_dim, 64)
        self.relu1 = nn.ReLU()
        self.fc2 = nn.Linear(64, 1)

    # 模型前向过程
    def forward(self, x):
        # 不建议这样写 x1 x1_act 占用太大的空间
        x1 = self.fc1(x)
        x1_act = self.relu1(x1)
        x = self.fc2(x1_act)  # 此时x维度为(16, 1) 而 y_pred为(16,) 不可以直接相减
        # x.size()返回x的维度(tuple):(16, 1) len(x.size())返回x的维数
        if len(x.size()) > 1:
            return x.squeeze(1)  # 分离dim=1的维度 此时返回值形状为(16,)
        return x

模型的结构与参数量为：
![[/img/post/Pasted image 20250226085046.png]]

超参设置与模型训练

超参设置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 调用GPU or CPU
device = "cuda" if torch.cuda.is_available() else 'cpu'

config = {
    "lr": 0.001,  # 学习率
    "epochs": 20,  # 训练次数
    "momentum": 0.9,  # 动量参数
    "save_path": "model_save/Best_model.dat",  # 模型保存路径
    "out_path": "pred.csv"  # 输出路径
}

model = Model(93).to(device)
loss = nn.MSELoss()  # 损失函数
optimizer = optim.SGD(model.parameters(), lr=config["lr"], momentum=config["momentum"])  # 采用动量梯度变化优化器momentum
  • 选择均方误差损失函数(nn.MSELoss())衡量模型预测值与真实值之间的差异。在回归问题中，均方误差能直观反映预测值偏离真实值的程度，通过最小化该损失函数来优化模型参数。优化器采用随机梯度下降(optim.SGD)
  • 在梯度下降中添加momentum参数优点：加速收敛，抑制震荡，跳出局部最优
  • 传统 SGD 更新公式为$\omega = \omega - \eta \nabla J(\omega)$，其中$\eta$是学习率，是$\nabla J(\omega)$当前梯度。引入momentum后，更新公式变为$v=\gamma v-\eta \nabla J(v) \quad w=w+v$，，这里$v$是速度变量，$\gamma$就是momentum参数，通常取值在 0 - 1 之间，代码中设为 0.9。

训练模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
# 训练模型
def train_val(model, train_loader, val_loader, device, epochs, optimizer, loss, save_path):
    model = model.to(device)

    # 用于画图的 train loss 和 val loss
    plt_train_loss = []
    plt_val_loss = []
    # 记录最小的val loss
    min_val_loss = 65535

    # 开始训练
    for epoch in range(epochs):
        train_loss = 0.0
        val_loss = 0.0
        # 记录时间
        start_time = time.time()
        model.train()  # 模型调整为训练模式
        for x_batch, y_batch in train_loader:
            # 将数据存储到GPU上
            x, y = x_batch.to(device), y_batch.to(device)
            y_pred = model(x)
            train_batch_loss = loss(y_pred, y)
            train_batch_loss.backward()
            optimizer.step()  # 更新模型
            optimizer.zero_grad()
            train_loss += train_batch_loss.cpu().item()
        plt_train_loss.append(train_loss / train_loader.__len__())  # 记录每轮批次计算的平均值

        # 验证模型
        model.eval()  # 调整为验证模式
        with torch.no_grad():
            for x_batch, y_batch in val_loader:
                x, y = x_batch.to(device), y_batch.to(device)
                y_pred = model(x)
                val_batch_loss = loss(y_pred, y)
                val_loss += val_batch_loss.cpu().item()
        plt_val_loss.append(val_loss / val_loader.__len__())
        if min_val_loss > val_loss:
            # 只保存模型的状态字典, 不保存模型的定义
            torch.save(model.state_dict(), save_path)
        print("[%03d/%03d], time: %2.2f sec(s), TrainLoss:%.6f | ValLoss:%.6f" % \
              (epoch + 1, 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, device, config["epochs"], optimizer, loss, config["save_path"])

预测结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 预测结果
def evaluate(model, save_path, test_loader, device, out_path):
    model.load_state_dict(torch.load(config["save_path"]))
    out = []
    with torch.no_grad():
        for x in test_loader:
            y_pred = model(x.to(device))
            out.append(y_pred.cpu().item())
    with open(out_path, "w", newline='') as f:  # newline=''不会换行
        csv_writer = csv.writer(f)
        csv_writer.writerow(["id", "tested_positive"])
        for i in range(len(out)):
            csv_writer.writerow([str(i), str(out[i])])
    print(f"文件已经保存到{out_path}中")


evaluate(model, config["save_path"], test_loader, device, config["out_path"])

项目优化内容

损失函数正则化(mseLoss with regulate)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
'''  
正则化: loss = loss + W * W  
让loss和参数都越小越好，让函数平滑，缓解函数过拟合  
'''  
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  # 返回损

相关系数：线性相关(Select K Best)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
'''  
相关系数：线性相关(select K best)  
不确定每一列都非常重要，有些列可能完全没有用，只起到噪声的作用  
选择相关系数(k)高的那几列留下  
'''  
def get_feature_importance(feature_data, label_data, k =4,column = None):  
    """  
    feature_data, label_data 要求字符串形式  
    k为选择的特征数量  
    如果需要打印column，需要传入行名  
    此处省略 feature_data, label_data 的生成代码。  
    如果是 CSV 文件，可通过 read_csv() 函数获得特征和标签。  
    这个函数的目的是， 找到所有的特征种， 比较有用的k个特征， 并打印这些列的名字。  
    """    model = SelectKBest(chi2, k=k)      #定义一个选择k个最佳特征的函数  
    feature_data = np.array(feature_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] for i in indices[0:k].tolist()]         # 选中这些列 打印  
        print('k best features are: ',k_best_features)  
    return X_new, indices[0:k]                  # 返回选中列的特征和他们的下标。

主成分分析(PCA)

1
# Todo

总结

完整代码

  • 未优化的代码
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    import time  
      
    import torch  
    import matplotlib.pyplot as plt  
    import csv  
    import numpy as np  
    import pandas as pd  
    from torch.utils.data import DataLoader, Dataset  
    import torch.nn as nn  # 引入NN模型  
    import torch.optim as optim  
      
    from sklearn.feature_selection import SelectKBest, chi2  
      
    # 新冠模型预测实战  
      
    ### 数据处理部分  数据处理越好，模型越优良  
    # 数据集划分一：训练集占80%,测试集(不用于训练)占20%  
    # 数据集划分二：训练集占70%，验证集(用于训练之后验证，不更新模型)占10%，测试集(用于最后验证)占20%  
    class CovidDataset(Dataset):  
        def __init__(self, file_path, mode="train"):  
            with open(file_path, 'r') as f:  
                ori_data = list(csv.reader(f))  
                # 去掉第一行第一列  
                csv_data = np.array(ori_data[1:])[:, 1:].astype(float)  
            self.mode = mode  
            # 划分数据集  
            if mode == "train":  # 逢5取1  
                # 筛选训练集  
                indices = [i for i in range(len(csv_data)) if i % 5 != 0]  
                # 训练集标签  
                self.y = torch.tensor(csv_data[indices, -1])  
                data = torch.tensor(csv_data[indices, :-1])  
            elif mode == "val":  
                # 筛选验证集  
                indices = [i for i in range(len(csv_data)) if i % 5 != 0]  
                # 验证集标签  
                self.y = torch.tensor(csv_data[indices, -1])  
                data = torch.tensor(csv_data[indices, :-1])  
            else:  
                # 筛选测试集  
                indices = [i for i in range(len(csv_data))]  
                data = torch.tensor(csv_data[indices])  
            # 归一化: 消除部分平行数据之间量纲的影响  
            # （x - x的平均值） / x的标准差  
            self.data = (data - data.mean(dim=0, keepdim=True)) / data.std(dim=0, keepdim=True)  
      
        def __getitem__(self, idx):  
            if self.mode != "test":  
                # 训练集和验证集要返回标签和数据 加入.float() 是因为data的类型时float64将其转化为float32 64位消耗大  
                return self.data[idx].float(), self.y[idx].float()  
            else:  
                # 测试集只返回数据  
                return self.data[idx].float()  
      
        def __len__(self):  
            return len(self.data)  
      
      
    train_file = "covid.train.csv"  
    test_file = "covid.test.csv"  
      
    train_dataset = CovidDataset(train_file, "train")  
    val_dataset = CovidDataset(train_file, "val")  
    test_dataset = CovidDataset(test_file, "test")  
      
    # 批次训练  
    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 data, y in train_dataset:  
    #     print(data, y)  
      
    # 验证数据读取  
    # file = pd.read_csv(train_file)  
    # print(file.head())  
      
    # 验证批次取数据  
    # for x_batch, y_batch in train_loader:  
    #     print(x_batch, y_batch)  
    # x_batch shape (16, 93)  
      
    ### 定义模型部分  
    class Model(nn.Module):  
        # in_dim 输入维度, out_dim 输出维度此处固定为1  
        def __init__(self, in_dim):  
            # 使用父类初始化  
            super(Model, self).__init__()  
            # 定义全连接层  
            self.fc1 = nn.Linear(in_dim, 64)  
            self.relu1 = nn.ReLU()  
            self.fc2 = nn.Linear(64, 1)  
      
        # 模型前向过程  
        def forward(self, x):  
            # 不建议这样写 x1 x1_act 占用太大的空间  
            x1 = self.fc1(x)  
            x1_act = self.relu1(x1)  
            x = self.fc2(x1_act)  # 此时x维度为(16, 1) 而 y_pred为(16,) 不可以直接相减  
            # x.size()返回x的维度(tuple):(16, 1) len(x.size())返回x的维数  
            if len(x.size()) > 1:  
                return x.squeeze(1)  # 分离dim=1的维度 此时返回值形状为(16,)  
            return x  
      
      
    # 验证模型  
    # model = Model(93)  
    # for x_batch, y_batch in train_loader:  
    #     y_pred = model(x_batch)         # 此时返回y_pred维度为(16, 1)  
      
    ### 超参: 学习率，优化器(SGD, Adam, Adamw)，损失函数(MAE)部分  
    # 调用GPU or CPU  
    device = "cuda" if torch.cuda.is_available() else 'cpu'  
      
    config = {  
        "lr": 0.001,  # 学习率  
        "epochs": 20,  # 训练次数  
        "momentum": 0.9,  # 动量参数  
        "save_path": "model_save/Best_model.dat",  # 模型保存路径  
        "out_path": "pred.csv"  
    }  
      
    model = Model(93).to(device)  
    loss = nn.MSELoss()  # 损失函数  
    optimizer = optim.SGD(model.parameters(), lr=config["lr"], momentum=config["momentum"])  # 采用动量梯度变化优化器momentum  
      
    # 训练模型  
    def train_val(model, train_loader, val_loader, device, epochs, optimizer, loss, save_path):  
        model = model.to(device)  
      
        # 用于画图的 train loss 和 val loss    plt_train_loss = []  
        plt_val_loss = []  
        # 记录最小的val loss  
        min_val_loss = 65535  
      
        # 开始训练  
        for epoch in range(epochs):  
            train_loss = 0.0  
            val_loss = 0.0  
            # 记录时间  
            start_time = time.time()  
            model.train()  # 模型调整为训练模式  
            for x_batch, y_batch in train_loader:  
                # 将数据存储到GPU上  
                x, y = x_batch.to(device), y_batch.to(device)  
                y_pred = model(x)  
                train_batch_loss = loss(y_pred, y)  
                train_batch_loss.backward()  
                optimizer.step()  # 更新模型  
                optimizer.zero_grad()  
                train_loss += train_batch_loss.cpu().item()  
            plt_train_loss.append(train_loss / train_loader.__len__())  # 记录每轮批次计算的平均值  
      
            # 验证模型  
            model.eval()  # 调整为验证模式  
            with torch.no_grad():  
                for x_batch, y_batch in val_loader:  
                    x, y = x_batch.to(device), y_batch.to(device)  
                    y_pred = model(x)  
                    val_batch_loss = loss(y_pred, y)  
                    val_loss += val_batch_loss.cpu().item()  
            plt_val_loss.append(val_loss / val_loader.__len__())  
            if min_val_loss > val_loss:  
                # 只保存模型的状态字典, 不保存模型的定义  
                torch.save(model.state_dict(), save_path)  
            print("[%03d/%03d], time: %2.2f sec(s), TrainLoss:%.6f | ValLoss:%.6f" % \  
                  (epoch + 1, 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, device, config["epochs"], optimizer, loss, config["save_path"])  
      
      
    # 预测结果  
    def evaluate(model, save_path, test_loader, device, out_path):  
        model.load_state_dict(torch.load(config["save_path"]))  
        out = []  
        with torch.no_grad():  
            for x in test_loader:  
                y_pred = model(x.to(device))  
                out.append(y_pred.cpu().item())  
        with open(out_path, "w", newline='') as f:  # newline=''不会换行  
            csv_writer = csv.writer(f)  
            csv_writer.writerow(["id", "tested_positive"])  
            for i in range(len(out)):  
                csv_writer.writerow([str(i), str(out[i])])  
        print(f"文件已经保存到{out_path}中")  
      
      
    evaluate(model, config["save_path"], test_loader, device, config["out_path"])
  • 优化后的代码
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200
    201
    202
    203
    204
    205
    206
    207
    208
    209
    210
    211
    212
    213
    214
    215
    216
    217
    218
    219
    220
    221
    222
    223
    224
    225
    226
    227
    228
    229
    230
    231
    232
    233
    234
    235
    236
    237
    238
    239
    240
    241
    242
    243
    244
    245
    246
    247
    248
    249
    250
    251
    252
    253
    254
    255
    256
    257
    258
    259
    260
    261
    262
    263
    264
    265
    266
    267
    268
    269
    270
    271
    272
    273
    274
    275
    276
    277
    278
    279
    280
    281
    282
    283
    import time  
      
    import torch  
    import matplotlib.pyplot as plt  
    import csv  
    import numpy as np  
    import pandas as pd  
    from torch.utils.data import DataLoader, Dataset  
    import torch.nn as nn  # 引入NN模型  
    import torch.optim as optim  
      
    from sklearn.feature_selection import SelectKBest, chi2  
      
    # 新冠模型预测实战  
      
    ### 数据处理部分  数据处理越好，模型越优良  
    # 数据集划分一：训练集占80%,测试集(不用于训练)占20%  
    # 数据集划分二：训练集占70%，验证集(用于训练之后验证，不更新模型)占10%，测试集(用于最后验证)占20%  
      
    '''  
    优化内容2：相关系数：线性相关(select K best)  
    不确定每一列都非常重要，有些列可能完全没有用，只起到噪声的作用  
    选择相关系数(k)高的那几列留下  
    '''  
      
      
    def get_feature_importance(feature_data, label_data, k=4, column=None):  
        """  
        feature_data, label_data 要求字符串形式  
        k为选择的特征数量  
        如果需要打印column，需要传入行名  
        此处省略 feature_data, label_data 的生成代码。  
        如果是 CSV 文件，可通过 read_csv() 函数获得特征和标签。  
        这个函数的目的是， 找到所有的特征种， 比较有用的k个特征， 并打印这些列的名字。  
        column 输入列的名称 会打印出类列的名称  
        """    model = SelectKBest(chi2, k=k)  # 定义一个选择k个最佳特征的函数  
        feature_data = np.array(feature_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]表示逆置一个列表或者矩阵。      例：[1, 2, 3] -> [3, 2, 1]  
        # sort 函数将向量直接排序 例：3.5 2.0 6.1 sort输出 2.0 3.5 6.1  argsort输出 1 0 2    # argsort这个函数， 可以矩阵排序后的下标。 比如 indices[0]表示的是，scores中最小值的下标。  
      
        if column:  # 如果需要打印选中的列名字  
            k_best_features = [column[i] 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):  
            '''  
            :param file_path: 文件读取路径  
            :param mode: 划分数据集 只可为 train val test        :param all_feature: 是否选取所有列作为数据集，默认为 True        :param feature_dim: 当 all feature=False时有效，选取相关系数比较大的几个列的列数  
            '''        with open(file_path, 'r') as f:  
                ori_data = list(csv.reader(f))  
                # 去掉第一行第一列  
                csv_data = np.array(ori_data[1:])[:, 1:].astype(float)  
                column = ori_data[0]  
            self.mode = mode  
      
            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(93)])  
            else:  
                # col保存重要的4列 类型为array  
                _, col = get_feature_importance(feature, label_data, k=feature_dim, column=column)  
      
            # 划分数据集  
            if mode == "train":  # 逢5取1  
                # 筛选训练集  
                indices = [i for i in range(len(csv_data)) if i % 5 != 0]  
                # 训练集标签  
                self.y = torch.tensor(csv_data[indices, -1])  
                data = torch.tensor(csv_data[indices, :-1])  
            elif mode == "val":  
                # 筛选验证集  
                indices = [i for i in range(len(csv_data)) if i % 5 != 0]  
                # 验证集标签  
                self.y = torch.tensor(csv_data[indices, -1])  
                data = torch.tensor(csv_data[indices, :-1])  
            else:  
                # 筛选测试集  
                indices = [i for i in range(len(csv_data))]  
                data = torch.tensor(csv_data[indices, :])  
      
            # 取出重要的最相关的4列  
            col = col.tolist()  
            data = data[:, col]  
      
            # 归一化: 消除部分平行数据之间量纲的影响  
            # （x - x的平均值） / x的标准差  
            self.data = (data - data.mean(dim=0, keepdim=True)) / data.std(dim=0, keepdim=True)  
      
        def __getitem__(self, idx):  
            if self.mode != "test":  
                # 训练集和验证集要返回标签和数据 加入.float() 是因为data的类型时float64将其转化为float32 64位消耗大  
                return self.data[idx].float(), self.y[idx].float()  
            else:  
                # 测试集只返回数据  
                return self.data[idx].float()  
      
        def __len__(self):  
            return len(self.data)  
      
      
    train_file = "covid.train.csv"  
    test_file = "covid.test.csv"  
      
    all_feature = False  
    if all_feature:  
        feature_dim = 93  
    else:  
        feature_dim = 6  
      
    train_dataset = CovidDataset(train_file, "train", all_feature, feature_dim)  
    val_dataset = CovidDataset(train_file, "val", all_feature, feature_dim)  
    test_dataset = CovidDataset(test_file, "test", all_feature, feature_dim)  
      
    # 批次训练  
    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 data, y in train_dataset:  
    #     print(data, y)  
      
    # 验证数据读取  
    # file = pd.read_csv(train_file)  
    # print(file.head())  
      
    # 验证批次取数据  
    # for x_batch, y_batch in train_loader:  
    #     print(x_batch, y_batch)  
    # x_batch shape (16, 93)  
      
    ### 定义模型部分  
    class Model(nn.Module):  
        # in_dim 输入维度, out_dim 输出维度此处固定为1  
        def __init__(self, in_dim):  
            # 使用父类初始化  
            super(Model, self).__init__()  
            # 定义全连接层  
            self.fc1 = nn.Linear(in_dim, 64)  
            self.relu1 = nn.ReLU()  
            self.fc2 = nn.Linear(64, 1)  
      
        # 模型前向过程  
        def forward(self, x):  
            # 不建议这样写 x1 x1_act 占用太大的空间  
            x1 = self.fc1(x)  
            x1_act = self.relu1(x1)  
            x = self.fc2(x1_act)  # 此时x维度为(16, 1) 而 y_pred为(16,) 不可以直接相减  
            # x.size()返回x的维度(tuple):(16, 1) len(x.size())返回x的维数  
            if len(x.size()) > 1:  
                return x.squeeze(1)  # 分离dim=1的维度 此时返回值形状为(16,)  
            return x  
      
      
    # 验证模型  
    # model = Model(93)  
    # for x_batch, y_batch in train_loader:  
    #     y_pred = model(x_batch)         # 此时返回y_pred维度为(16, 1)  
      
    ### 超参: 学习率，优化器(SGD, Adam, Adamw)，损失函数(MAE)部分  
    # 调用GPU or CPU  
    device = "cuda" if torch.cuda.is_available() else 'cpu'  
      
    config = {  
        "lr": 0.001,  # 学习率  
        "epochs": 20,  # 训练次数  
        "momentum": 0.9,  # 动量参数  
        "save_path": "model_save/Best_model.dat",  # 模型保存路径  
        "out_path": "pred.csv"  
    }  
      
    '''  
    优化内容1：正则化: loss = loss + W * W  
    让loss和参数都越小越好，让函数平滑，缓解函数过拟合  
    '''  
      
      
    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  # 返回损失值  
      
      
    '''正则化优化定义结束'''  
      
    model = Model(feature_dim).to(device)  
    loss = mseLoss_with_reg  
    optimizer = optim.SGD(model.parameters(), lr=config["lr"], momentum=config["momentum"])  # 采用动量梯度变化优化器momentum  
      
      
    # 训练模型  
    def train_val(model, train_loader, val_loader, device, epochs, optimizer, loss, save_path):  
        model = model.to(device)  
      
        # 用于画图的 train loss 和 val loss    plt_train_loss = []  
        plt_val_loss = []  
        # 记录最小的val loss  
        min_val_loss = 65535  
      
        # 开始训练  
        for epoch in range(epochs):  
            train_loss = 0.0  
            val_loss = 0.0  
            # 记录时间  
            start_time = time.time()  
            model.train()  # 模型调整为训练模式  
            for x_batch, y_batch in train_loader:  
                # 将数据存储到GPU上  
                x, y = x_batch.to(device), y_batch.to(device)  
                y_pred = model(x)  
                train_batch_loss = loss(y_pred, y, model)  
                train_batch_loss.backward()  
                optimizer.step()  # 更新模型  
                optimizer.zero_grad()  
                train_loss += train_batch_loss.cpu().item()  
            plt_train_loss.append(train_loss / train_loader.__len__())  # 记录每轮批次计算的平均值  
      
            # 验证模型  
            model.eval()  # 调整为验证模式  
            with torch.no_grad():  
                for x_batch, y_batch in val_loader:  
                    x, y = x_batch.to(device), y_batch.to(device)  
                    y_pred = model(x)  
                    val_batch_loss = loss(y_pred, y, model)  
                    val_loss += val_batch_loss.cpu().item()  
            plt_val_loss.append(val_loss / val_loader.__len__())  
            if min_val_loss > val_loss:  
                # 只保存模型的状态字典, 不保存模型的定义  
                torch.save(model.state_dict(), save_path)  
                min_val_loss = val_loss  
      
            print("[%03d/%03d], time: %2.2f sec(s), TrainLoss:%.6f | ValLoss:%.6f" % \  
                  (epoch + 1, 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, device, config["epochs"], optimizer, loss, config["save_path"])  
      
      
    # 预测结果  
    def evaluate(model, save_path, test_loader, device, out_path):  
        model.load_state_dict(torch.load(config["save_path"]))  
        out = []  
        with torch.no_grad():  
            for x in test_loader:  
                y_pred = model(x.to(device))  
                out.append(y_pred.cpu().item())  
        with open(out_path, "w", newline='') as f:  # newline=''不会换行  
            csv_writer = csv.writer(f)  
            csv_writer.writerow(["id", "tested_positive"])  
            for i in range(len(out)):  
                csv_writer.writerow([str(i), str(out[i])])  
        print(f"文件已经保存到{out_path}中")  
      
      
    evaluate(model, config["save_path"], test_loader, device, config["out_path"])  
      
    '''  
    TODO: 主成分分析 PCA 数据降维  
    将 93 维数据改造为 4 维  
    跟相关系数挑选4维数据不相同  
    '''

工具

  • 输出模型参数以及模型结构，原文链接
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    # 输出模型参数及模型结构
    def model_structure(model):
        blank = ' '
        print('-' * 90)
        print('|' + ' ' * 11 + 'weight name' + ' ' * 10 + '|' \
              + ' ' * 15 + 'weight shape' + ' ' * 15 + '|' \
              + ' ' * 3 + 'number' + ' ' * 3 + '|')
        print('-' * 90)
        num_para = 0
        type_size = 1  # 如果是浮点数就是4

        for index, (key, w_variable) in enumerate(model.named_parameters()):
            if len(key) <= 30:
                key = key + (30 - len(key)) * blank
            shape = str(w_variable.shape)
            if len(shape) <= 40:
                shape = shape + (40 - len(shape)) * blank
            each_para = 1
            for k in w_variable.shape:
                each_para *= k
            num_para += each_para
            str_num = str(each_para)
            if len(str_num) <= 10:
                str_num = str_num + (10 - len(str_num)) * blank

            print('| {} | {} | {} |'.format(key, shape, str_num))
        print('-' * 90)
        print('The total number of parameters: ' + str(num_para))
        print('The parameters of Model {}: {:4f}M'.format(model._get_name(), num_para * type_size / 1000 / 1000))
        print('-' * 90)
  • (在线LaTeX公式编辑器-编辑器)
  • 动量梯度下降优点解释：
    • 跳出局部最优：在复杂的损失函数空间中，模型可能陷入局部最优解，此时传统 SGD 难以跳出。而momentum赋予了模型一定的 “惯性”，当模型陷入局部最优时，积累的 “速度” 可能帮助它跳出这个局部最优区域，继续寻找全局最优解。就好比小球在一个坑洼的地形上滚动，动量能让小球有机会越过一些小坑，找到更低的地方。
    • 抑制震荡：在训练过程中，若梯度方向频繁变化，传统 SGD 会导致参数更新方向不稳定，出现震荡现象，影响收敛效率。momentum参数可对这种震荡起到抑制作用。因为它会综合考虑过往梯度，使得更新方向更平滑。比如在一个二维的损失函数曲面上，如果梯度在两个维度上频繁变化，有了momentum，更新方向不会随着每次梯度的微小变化而大幅改变，从而减少不必要的震荡，让模型更稳定地朝着最优解前进。
    • 加速收敛：在传统 SGD 中，参数更新仅依据当前批次数据计算的梯度。而引入momentum后，参数更新不仅考虑当前梯度，还结合之前梯度的累积信息。就像跑步时，运动员凭借之前积累的速度（过往梯度的影响）能跑得更快。数学上，参数更新公式在原有基础上增加了与之前梯度相关的项。