浅析深度学习--回归实战（新冠人数预测）

Posted 2023-01-16 qianzhipeng123

任务

过程：

一.调包

from sklearn.feature_selection import SelectKBest
from sklearn.feature_selection import chi2
import csv #读csv文件的
import numpy as np
import time
import matplotlib.pyplot as plt
import pandas as pd #分析数据的

from torch import optim
import torch.nn as nn
import torch
from torch.utils.data import Dataset,DataLoader

二.创建函数selectkbest--挑选重要参数

slectkbest原理：通过给特征进行打分，然后从高到底选取特征，要用到皮尔森系数和单变量线性回归，

皮尔森系数可以表明y与各个特征的相关度

单变量线性回归可以用来表明特征的回归效果

#selectkbest
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个最佳特征的函数
    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是从小到大排序
    # 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]                  # 返回选中列的特征和他们的下标。

X_new选择了k个最佳特征，用indices记录下标，并输出打印

x_new打印结果

三.创建类cobidDataset--数据集

通常包含三个任务

init--初始化（包括列表读取后转矩阵，取关键列，划分训练集和验证集，对数据的列进行归一化

getitem--返回值

len--返回长度

#数据集：完成三个工作：1.初始化 2.getitem返回x【idx】，y【idx】 3.返回x的长度
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]    #用column第一行：类别
            x = np.array(csv_data)[1:,1:-1]  #转了矩阵，第一行不要了。第一列也不要了
            y = np.array(csv_data)[1:,-1]     
            # 若全选，则选中所有列0-93的列表。
            if all_feature:
                col_indices = np.array([i for i in range(0,93)]) #下标为0-93
            # 选重要的dim列。
            else:
                _, col_indices = get_feature_importance(x, y, 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]   #12346789
                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]
                #如果选重要的dim
                # 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))]     # 测试机只有数据
                #如果选重要的dim
                # 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                                   # 表示当前数据集的模式
            # 对数据进行列归一化，防止由于某些数据过大引起loss过大
            self.data = (self.data - self.data.mean(dim=0,keepdim=True)) / self.data.std(dim=0,keepdim=True)  
            #assert为不相等就报错
            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)                 # 返回数据长度。

四.创建类myNet--用于构造模型

通常包含两个任务

init--转换维度，利用Linear函数转换维度

forward--模型前向过程，不断缩小维度

class myNet(nn.Module):     #模型
    def __init__(self,inDim):
        super(myNet,self).__init__()
        self.fc1 = nn.Linear(inDim, 64)              # 全连接
        self.relu = nn.ReLU()                        # 激活函数，添加非线性ReLU
        self.fc2 = nn.Linear(64,1)                     # 全连接

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

        x = self.fc2(x)
        if len(x.size()) > 1:
            return x.squeeze(1)
        else:
            return x

激活函数：

缩小维度：

缩小维度的过程

缩小维度后

五.创建函数train_val--用于训练

训练时用模型预测数据，与真实值做对比，得到train_loss，梯度回传，利用优化器更新模型

验证时用模型预测得到的数据与真实值做对比，得到val_loss，当loss值越来越小，则表明模型良好

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和

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

        model.eval()                 # 模型设置为验证状态
        with torch.no_grad():                    # 模型不再计算梯度
            for data in valloader:                      # 从验证集取一个batch的数据
                # 将数据放到设备上，验证不计算，放cpu上
                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比之前的最小值小， 说明模型更优， 保存这个模型
        
        #记录loss到列表。注意是平均的loss ，因此要除以数据集长度。
        plt_val_loss.append(val_loss/valloader.dataset.__len__())  
        
        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()                                 # 画图， 展示

六.创建函数evaluate--用于预测

利用模型计算预测值，然后打印出要求格式的excel表

def evaluate(model_path, testset, rel_path ,device):        #测试验证
    model = torch.load(model_path).to(device)                     # 模型放到设备上。加载模型
    # 将验证数据放入loader 验证时， 一般batch为1
    testloader = DataLoader(testset,batch_size=1,shuffle=False)         
    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)

七.设置参数

设置关键特征值，设置路径，设置数据集

#是否使用所有的列
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)      #读入excel表
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)   #读取训练， 验证，测试数据

定义loss函数

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 即 平方差损失，

八.开始训练

1. 设置参数

2. 设置模型，优化器，batch

3. 万事具备，开始训练

config = 
    'n_epochs': 50,                # maximum number of epochs
    'batch_size': 32,               # 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)             # 定义优化器，momentum表动量
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)           # 验证