浅析深度学习（二）–回归实战（新冠人数预测）
任务
过程：
一.调包
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 即 平方差损失，
八.开始训练
-
设置参数
-
设置模型，优化器，batch
-
万事具备，开始训练
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) # 验证
本图文内容来源于网友网络收集整理提供，作为学习参考使用，版权属于原作者。
THE END
二维码