柚子快報邀請碼778899分享：深度學習回歸任務訓練代碼模版

http://yzkb.51969.com/

深度學習回歸任務訓練代碼模版

文章目錄

深度學習回歸任務訓練代碼模版參數設置功能函數數據加載自定義數據集加載類特征選擇（可選）數據讀取

定義模型訓練模型訓練迭代＋驗證迭代使用 `tensorboard` 輸出模型訓練過程和指標可視化(可選)

結果預測參考

參數設置

超參設置：config 包含所有訓練需要的超參數（便于后續(xù)的調參），以及模型需要存儲的位置

device = 'cuda' if torch.cuda.is_available() else 'cpu'

config = {

'seed': 5201314, # 隨機種子，可以自己填寫. :)

'select_all': False, # 是否選擇全部的特征

'valid_ratio': 0.2, # 驗證集大小(validation_size) = 訓練集大小(train_size) * 驗證數據占比(valid_ratio)

'n_epochs': 3000, # 數據遍歷訓練次數

'batch_size': 256,

'learning_rate': 1e-5,

'early_stop': 400, # 如果early_stop輪損失沒有下降就停止訓練.

'save_path': './models/model.ckpt' # 模型存儲的位置

}

功能函數

導入需要的Python包

# 數值、矩陣操作

import math

import numpy as np

# 數據讀取與寫入

import pandas as pd

import os

import csv

# 進度條

# from tqdm import tqdm

# 如果是使用notebook 推薦使用以下（顏值更高 : ) ）

from tqdm.notebook import tqdm

# Pytorch 深度學習張量操作框架

import torch

import torch.nn as nn

from torch.utils.data import Dataset, DataLoader, random_split

# 繪制pytorch的網絡

from torchviz import make_dot

# 學習曲線繪制

from torch.utils.tensorboard import SummaryWriter

一些重要的方法（隨機種子設置、數據拆分、模型預測）

# 定義一個函數來設置隨機種子，以確保實驗的可復現(xiàn)性

def same_seed(seed):

"""

設置隨機種子(便于復現(xiàn))

"""

# 設置CUDA的確定性，確保每次運行的結果是確定的

torch.backends.cudnn.deterministic = True

# 關閉CUDA的benchmark模式，因為這與確定性運行模式沖突

torch.backends.cudnn.benchmark = False

# 設置NumPy的隨機種子

np.random.seed(seed)

# 設置PyTorch的隨機種子

torch.manual_seed(seed)

# 如果CUDA可用，則為GPU設置隨機種子

if torch.cuda.is_available():

torch.cuda.manual_seed_all(seed)

# 打印設置的種子值

print(f'Set Seed = {seed}')

# 定義一個函數來將數據集隨機拆分為訓練集和驗證集

def train_valid_split(data_set, valid_ratio, seed):

"""

數據集拆分成訓練集（training set）和 驗證集（validation set）

"""

# 計算驗證集的大小

valid_set_size = int(valid_ratio * len(data_set))

# 訓練集的大小是數據集總大小減去驗證集大小

train_set_size = len(data_set) - valid_set_size

# 使用PyTorch的random_split函數來拆分數據集，傳入隨機種子以確保可復現(xiàn)性

train_set, valid_set = random_split(data_set, [train_set_size, valid_set_size], generator=torch.Generator().manual_seed(seed))

# 將拆分得到的數據集轉換為NumPy數組格式并返回

return np.array(train_set), np.array(valid_set)

# 定義一個函數來進行模型的預測

def predict(test_loader, model, device):

# 將模型設置為評估模式

model.eval()

# 初始化一個列表來存儲預測結果

preds = []

# 遍歷測試數據集

for x in tqdm(test_loader):

# 將數據移動到指定的設備上（CPU或GPU）

x = x.to(device)

# 使用with torch.no_grad()來禁止計算梯度，因為在預測模式下不需要計算梯度

with torch.no_grad():

# 進行前向傳播以獲得預測結果

pred = model(x)

# 將預測結果從GPU移回CPU，并將其從計算圖中分離出來

preds.append(pred.detach().cpu())

# 將所有批次的預測結果拼接成一個NumPy數組，并返回

preds = torch.cat(preds, dim=0).numpy()

return preds

數據加載

自定義數據集加載類

# 定義一個COVID19數據集類，繼承自PyTorch的Dataset類

class COVID19Dataset(Dataset):

"""

x: np.ndarray 特征矩陣.

y: np.ndarray 目標標簽, 如果為None,則是預測的數據集

"""

def __init__(self, x, y=None):

# 如果y不是None，則將y轉換為PyTorch的FloatTensor類型，否則y保持為None

if y is None:

self.y = y

else:

self.y = torch.FloatTensor(y)

# 將x轉換為PyTorch的FloatTensor類型

self.x = torch.FloatTensor(x)

def __getitem__(self, idx):

# 根據索引idx獲取數據項

# 如果y是None，表示這是一個預測數據集，只返回x

if self.y is None:

return self.x[idx]

# 否則，返回一個包含x和y的元組

return self.x[idx], self.y[idx]

def __len__(self):

# 返回數據集中x的數量，即數據集的大小

return len(self.x)

特征選擇（可選）

觀察數據，選擇更有效的數據

df = pd.read_csv('./covid.train.csv')

df.describe()

利用Pearson相關系數分析不同feature與label的相關性強弱。

df.corr()['tested_positive'].sort_values(ascending=False)

# 定義一個函數來選擇特征，用于擬合回歸模型

def select_feat(train_data, valid_data, test_data, select_all=True):

"""

特征選擇

選擇較好的特征用來擬合回歸模型

"""

# 從訓練數據中分離出目標變量y

y_train, y_valid = train_data[:, -1], valid_data[:, -1]

'''

[:, -1]：這是一個NumPy的切片操作，用于選擇數組中的特定行和列。

:表示選擇所有行，即選擇整個數據集。

-1表示選擇最后一列。在Python中，使用負數索引可以從數組的末尾開始計數，-1就是數組中的最后一個元素，對于二維數組來說，就是最后一列。

'''

# 從訓練數據、驗證數據和測試數據中分離出特征矩陣x

raw_x_train, raw_x_valid, raw_x_test = train_data[:, :-1], valid_data[:, :-1], test_data

'''

[:, :-1]：這是一個NumPy的切片操作，用于選擇數組中的特定行和列。

:表示選擇所有行，即選擇整個數據集。

:-1表示選擇從第一列開始直到倒數第二列的所有列。在Python中，使用負數索引可以從數組的末尾開始計數，-1就是數組中的最后一個元素之前的所有元素，對于二維數組來說，就是除了最后一列之外的所有列。

'''

# 如果select_all為True，則選擇所有特征

if select_all:

feat_idx = list(range(raw_x_train.shape[1]))

else:

# 否則，根據某些標準（需要自行調研特征選擇方法）選擇特征

# 根據Pearson系數降序排列的結果，我們重新選擇特征。

# 去掉第一列 id 列

feat_idx = list(range(1, 38)) + [53, 69, 85, 101]

# 返回選定的特征矩陣和目標變量

return raw_x_train[:, feat_idx], raw_x_valid[:, feat_idx], raw_x_test[:, feat_idx], y_train, y_valid

數據讀取

從文件中讀取數據pd.read_csv數據拆分成三份 訓練（training）、驗證（validation）、測試（testing）

train_valid_split： 分成訓練、驗證select_feat：拆分特征和label，并進行特征選擇COVID19Dataset: 分別將訓練、驗證、測試集的特征和label組合成可以用于快速迭代訓練的數據集train_dataset, valid_dataset, test_dataset

# 設置隨機種子便于復現(xiàn)

same_seed(config['seed'])

# 訓練集大小(train_data size) : 2699 x 118 (id + 37 states + 16 features x 5 days)

# 測試集大小(test_data size）: 1078 x 117 (沒有l(wèi)abel (last day's positive rate))

pd.set_option('display.max_column', 200) # 設置顯示數據的列數

train_df, test_df = pd.read_csv('./covid.train.csv'), pd.read_csv('./covid.test.csv')

display(train_df.head(3)) # 顯示前三行的樣本

train_data, test_data = train_df.values, test_df.values

del train_df, test_df # 刪除數據減少內存占用

train_data, valid_data = train_valid_split(train_data, config['valid_ratio'], config['seed'])

# 打印數據的大小

print(f"""train_data size: {train_data.shape}

valid_data size: {valid_data.shape}

test_data size: {test_data.shape}""")

# 特征選擇

x_train, x_valid, x_test, y_train, y_valid = select_feat(train_data, valid_data, test_data, config['select_all'])

# 打印出特征數量

print(f'number of features: {x_train.shape[1]}')

train_dataset, valid_dataset, test_dataset = COVID19Dataset(x_train, y_train), \

COVID19Dataset(x_valid, y_valid), \

COVID19Dataset(x_test)

# 使用Pytorch中Dataloader類按照Batch將數據集加載

train_loader = DataLoader(train_dataset, batch_size=config['batch_size'], shuffle=True, pin_memory=True)

valid_loader = DataLoader(valid_dataset, batch_size=config['batch_size'], shuffle=True, pin_memory=True)

test_loader = DataLoader(test_dataset, batch_size=config['batch_size'], shuffle=False, pin_memory=True)

定義模型

# 定義一個自定義的模型類My_Model，繼承自nn.Module

class My_Model(nn.Module):

def __init__(self, input_dim):

super(My_Model, self).__init__()

# 定義模型的結構為一個順序模型（Sequential），包含線性層（Linear）和激活函數（ReLU）

self.layers = nn.Sequential(

# 第一層線性層，輸入維度為input_dim，輸出維度為16

nn.Linear(input_dim, 16),

# ReLU激活函數

nn.ReLU(),

# 第二層線性層，輸入維度為16，輸出維度為8

nn.Linear(16, 8),

# ReLU激活函數

nn.ReLU(),

# 第三層線性層，輸入維度為8，輸出維度為1

nn.Linear(8, 1)

)

def forward(self, x):

# 在前向傳播中，輸入x通過定義的層（layers）

x = self.layers(x)

# 使用squeeze函數移除輸出張量的一個維度，使其從形狀(B, 1)變?yōu)?B)

x = x.squeeze(1)

# 返回模型的輸出

return x

訓練模型

訓練迭代＋驗證迭代

def trainer(train_loader, valid_loader, model, config, device):

criterion = nn.MSELoss(reduction='mean') # 損失函數的定義

# 定義優(yōu)化器

# TODO: 可以查看學習更多的優(yōu)化器 https://pytorch.org/docs/stable/optim.html

# TODO: L2 正則( 可以使用optimizer(weight decay...) )或者 自己實現(xiàn)L2正則.

optimizer = torch.optim.SGD(model.parameters(), lr=config['learning_rate'], momentum=0.9)

# tensorboard 的記錄器

writer = SummaryWriter()

if not os.path.isdir('./models'):

# 創(chuàng)建文件夾-用于存儲模型

os.mkdir('./models')

n_epochs, best_loss, step, early_stop_count = config['n_epochs'], math.inf, 0, 0

for epoch in range(n_epochs):

model.train() # 訓練模式

loss_record = []

# tqdm可以幫助我們顯示訓練的進度

train_pbar = tqdm(train_loader, position=0, leave=True)

# 設置進度條的左邊 ： 顯示第幾個Epoch了

train_pbar.set_description(f'Epoch [{epoch+1}/{n_epochs}]')

for x, y in train_pbar:

optimizer.zero_grad() # 將梯度置0.

x, y = x.to(device), y.to(device) # 將數據一到相應的存儲位置(CPU/GPU)

pred = model(x) # 前向傳播

loss = criterion(pred, y) # 計算損失

loss.backward() # 反向傳播 計算梯度.

optimizer.step() # 更新網絡參數

step += 1

loss_record.append(loss.detach().item())

# 訓練完一個batch的數據，將loss 顯示在進度條的右邊

train_pbar.set_postfix({'loss': loss.detach().item()})

mean_train_loss = sum(loss_record)/len(loss_record)

# 每個epoch,在tensorboard 中記錄訓練的損失（后面可以展示出來）

writer.add_scalar('Loss/train', mean_train_loss, step)

model.eval() # 將模型設置成 evaluation 模式.

loss_record = []

for x, y in valid_loader:

x, y = x.to(device), y.to(device)

with torch.no_grad():

pred = model(x)

loss = criterion(pred, y)

loss_record.append(loss.item())

mean_valid_loss = sum(loss_record)/len(loss_record)

print(f'Epoch [{epoch+1}/{n_epochs}]: Train loss: {mean_train_loss:.4f}, Valid loss: {mean_valid_loss:.4f}')

# 每個epoch,在tensorboard 中記錄驗證的損失（后面可以展示出來）

writer.add_scalar('Loss/valid', mean_valid_loss, step)

# 如果當前驗證損失優(yōu)于最佳損失，則保存模型

if mean_valid_loss < best_loss:

best_loss = mean_valid_loss

torch.save(model.state_dict(), config['save_path']) # 模型保存

print('Saving model with loss {:.3f}...'.format(best_loss))

early_stop_count = 0

else:

early_stop_count += 1

# 如果連續(xù)多次驗證損失沒有改善，則停止訓練

if early_stop_count >= config['early_stop']:

print('\nModel is not improving, so we halt the training session.')

return

開始訓練

model = My_Model(input_dim=x_train.shape[1]).to(device) # 將模型和訓練數據放在相同的存儲位置(CPU/GPU)

trainer(train_loader, valid_loader, model, config, device)

使用 tensorboard 輸出模型訓練過程和指標可視化(可選)

tensorboard 可視化工具：可以記錄并展現(xiàn)模型的訓練過程中的各種指標，這里我們是記錄模型的損失

%reload_ext tensorboard

%tensorboard --logdir=./runs/ --port=6007

結果預測

測試集的預測結果保存到pred.csv.

def save_pred(preds, file):

""" 將模型保存到指定位置 """

with open(file, 'w', newline='') as fp: # 添加 newline='' 防止在Windows上出現(xiàn)額外的空行

writer = csv.writer(fp)

writer.writerow(['id', 'tested_positive'])

for i, p in enumerate(preds):

writer.writerow([i, p])

model = My_Model(input_dim=x_train.shape[1]).to(device)

model.load_state_dict(torch.load(config['save_path']))

preds = predict(test_loader, model, device)

save_pred(preds, 'pred.csv')

參考

完整代碼見：Hongyi_Lee_dl_homeworks/HW1_Regression at master · huaiyuechusan/Hongyi_Lee_dl_homeworks (github.com)

參考文章：

【李宏毅《機器學習》2022】作業(yè)1：COVID 19 Cases Prediction (Regression)_李宏毅2022作業(yè)-CSDN博客

Hongyi_Lee_dl_homeworks/Warmup/Pytorch_Tutorial_2.pdf at master · huaiyuechusan/Hongyi_Lee_dl_homeworks (github.com)

e/master/HW1_Regression)

參考文章：

【李宏毅《機器學習》2022】作業(yè)1：COVID 19 Cases Prediction (Regression)_李宏毅2022作業(yè)-CSDN博客

Hongyi_Lee_dl_homeworks/Warmup/Pytorch_Tutorial_2.pdf at master · huaiyuechusan/Hongyi_Lee_dl_homeworks (github.com)

柚子快報邀請碼778899分享：深度學習回歸任務訓練代碼模版

http://yzkb.51969.com/

精彩內容