Before All

今天先完成下HW1，阅读理解下代码。

代码来源李宏毅老师的blog：ML 2022 Spring

# 导入必要的库
import math
import numpy as np
import pandas as pd
import os
import csv
from tqdm import tqdm
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader, random_split
from torch.utils.tensorboard import SummaryWriter

# 固定随机数种子以确保结果可复现
def same_seed(seed):
    '''设置随机数生成器种子，确保实验可复现。'''
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

# 划分训练集和验证集
def train_valid_split(data_set, valid_ratio, seed):
    '''将提供的训练数据集划分为训练集和验证集'''
    valid_set_size = int(valid_ratio * len(data_set))  # 计算验证集大小
    train_set_size = len(data_set) - valid_set_size  # 计算训练集大小
    train_set, valid_set = random_split(data_set, [train_set_size, valid_set_size], generator=torch.Generator().manual_seed(seed))
    return np.array(train_set), np.array(valid_set)

# 模型预测
def predict(test_loader, model, device):
    '''使用训练好的模型进行预测'''
    model.eval()  # 设置模型为评估模式
    preds = []
    for x in tqdm(test_loader):  # 遍历测试集
        x = x.to(device)  # 将数据移至计算设备
        with torch.no_grad():  # 不计算梯度，减少内存消耗
            pred = model(x)  # 获取模型预测结果
            preds.append(pred.detach().cpu())  # 将结果从GPU移至CPU
    preds = torch.cat(preds, dim=0).numpy()  # 合并所有预测结果
    return preds

# 自定义数据集类，用于Pytorch的数据加载
class COVID19Dataset(Dataset):
    '''定义一个COVID-19数据集类'''
    def __init__(self, x, y=None):
        if y is None:
            self.y = y
        else:
            self.y = torch.FloatTensor(y)  # 目标变量y
        self.x = torch.FloatTensor(x)  # 特征x

    def __getitem__(self, idx):
        '''返回指定索引的数据'''
        if self.y is None:
            return self.x[idx]
        else:
            return self.x[idx], self.y[idx]

    def __len__(self):
        '''返回数据集的大小'''
        return len(self.x)

# 自定义神经网络模型
class My_Model(nn.Module):
    def __init__(self, input_dim):
        super(My_Model, self).__init__()
        '''定义一个简单的前馈神经网络模型'''
        #递减神经元数量并采用ReLU激活函数。
        self.layers = nn.Sequential(
            nn.Linear(input_dim, 16),
            nn.ReLU(),
            nn.Linear(16, 8),
            nn.ReLU(),
            nn.Linear(8, 1)
        )

    def forward(self, x):
        '''定义前向传播过程'''
        x = self.layers(x)
        x = x.squeeze(1)  # (B, 1) -> (B)
        return x

# 选择特征用于回归分析
def select_feat(train_data, valid_data, test_data, select_all=True):
    '''选择合适的特征用于回归模型训练'''
    y_train, y_valid = train_data[:,-1], valid_data[:,-1]
    raw_x_train, raw_x_valid, raw_x_test = train_data[:,:-1], valid_data[:,:-1], test_data

    if select_all:
        feat_idx = list(range(raw_x_train.shape[1]))  # 使用所有特征
    else:
        feat_idx = [0,1,2,3,4]  # 选择合适的特征列
        
    return raw_x_train[:,feat_idx], raw_x_valid[:,feat_idx], raw_x_test[:,feat_idx], y_train, y_valid

# 训练与验证模型
def trainer(train_loader, valid_loader, model, config, device):
    '''训练模型并进行验证'''
    criterion = nn.MSELoss(reduction='mean')  # 定义均方误差损失函数

    # 定义优化算法
    optimizer = torch.optim.SGD(model.parameters(), lr=config['learning_rate'], momentum=0.9)  # 使用SGD优化器

    writer = SummaryWriter()  # TensorBoard的Writer

    if not os.path.isdir('./models'):
        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 = []
        train_pbar = tqdm(train_loader, position=0, leave=True)  # 进度条显示

        for x, y in train_pbar:
            optimizer.zero_grad()  # 将梯度清零
            x, y = x.to(device), y.to(device)  # 将数据移至计算设备
            pred = model(x)  # 模型预测
            loss = criterion(pred, y)  # 计算损失
            loss.backward()  # 反向传播
            optimizer.step()  # 更新参数
            step += 1
            loss_record.append(loss.detach().item())  # 记录损失

            # 更新进度条显示
            train_pbar.set_description(f'Epoch [{epoch+1}/{n_epochs}]')
            train_pbar.set_postfix({'loss': loss.detach().item()})

        mean_train_loss = sum(loss_record) / len(loss_record)
        writer.add_scalar('Loss/train', mean_train_loss, step)

        # 验证集评估
        model.eval()  # 设置模型为评估模式
        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}')
        writer.add_scalar('Loss/valid', mean_valid_loss, step)

        # 如果验证集损失较低，则保存模型
        if mean_valid_loss < best_loss:
            best_loss = mean_valid_loss
            torch.save(model.state_dict(), config['save_path'])  # 保存最佳模型
            print(f'Saving model with loss {best_loss:.3f}...')
            early_stop_count = 0
        else:
            early_stop_count += 1

        # 如果早停次数超过设定值，则停止训练
        if early_stop_count >= config['early_stop']:
            print('\nModel is not improving, halting the training process.')
            return

# 配置参数
config = {
    'seed': 5201314,      # 随机种子
    'select_all': True,   # 是否使用所有特征
    'valid_ratio': 0.2,   # 验证集占比
    'n_epochs': 3000,     # 训练周期
    'batch_size': 256,    # 批次大小
    'learning_rate': 1e-5,  # 学习率
    'early_stop': 400,    # 早停策略
    'save_path': './models/model.ckpt'  # 模型保存路径
}

# 设置随机数种子
same_seed(config['seed'])

# 加载训练数据和测试数据
train_data, test_data = pd.read_csv('./covid.train.csv').values, pd.read_csv('./covid.test.csv').values
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)

# 创建数据加载器
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)

# 创建并训练模型
model = My_Model(input_dim=x_train.shape[1]).to(device)  # 创建模型并将其移动到计算设备
trainer(train_loader, valid_loader, model, config, device)

# 保存预测结果
def save_pred(preds, file):
    '''保存预测结果到指定文件'''
    with open(file, 'w') as fp:
        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')  # 保存预测结果

代码流程图如下

开始
│
├── 设置随机种子
│   ├── 确保实验可重复性
│
├── 加载数据
│   ├── 加载训练数据
│   ├── 加载测试数据
│   └── 分割数据集（训练集、验证集）
│
├── 选择特征
│   ├── 根据配置选择特征
│   └── 准备输入数据
│
├── 定义数据集类
│   ├── 定义如何获取数据
│   └── 返回数据集长度
│
├── 定义模型
│   ├── 输入层、隐藏层、输出层
│   └── 激活函数
│
├── 训练与验证
│   ├── 使用训练数据训练模型
│   ├── 使用验证数据评估性能
│   └── 保存最佳模型
│
├── 预测
│   ├── 加载最佳模型
│   ├── 使用测试集进行预测
│   └── 保存预测结果
│
└── 输出结果
    ├── 保存结果到 `pred.csv`
    └── 格式：`id, tested_positive`