•【Hung-yi Lee 機器學習 – L1 : COVID-19 Cases Prediction (Regression)】

參考資料

這系列是實作李宏毅的課程，每堂課都很有料！一起來當他腦粉

• 參考 2021 Spring、2022 Spring、2023 Spring、2024 Spring、2025
• Lecture 1 : Deep Learing Introduction
  • 機器學習基本概念簡介
  • 深度學習基本概念簡介
• HW 1 : COVID-19 Cases Prediction

Lecture 1 : Deep Learing Introduction

【機器學習2021】預測本頻道觀看人數 (上) – 機器學習基本概念簡介

當遇到一組資料，假設我們想使用【回歸】

第一步: 猜測符合raw_data的函式

第二步: 定義 誤差(Loss)

誤差(Loss) = 真實的值(label) – 預測的值(y)
Loss 相加平均越小，誤差越小，該組的(b,w)為目前最佳解

第三步: 優化

斜率

訓練出來後，發現測試集的誤差 > 訓練集

從趨勢中觀察出有週期性，周五、六觀看人數較低

修改模型

改成抓取前7天、前28天、前56天，找到目前最佳解
bias + 前＿天的 各天權重＊觀看人數

【機器學習2021】預測本頻道觀看人數 (下) – 深度學習基本概念簡介

【機器學習2021】類神經網路訓練不起來怎麼辦 (二)： 批次 (batch) 與動量 (momentum)

【機器學習2021】類神經網路訓練不起來怎麼辦 (三)：自動調整學習速率 (Learning Rate)

【機器學習2021】類神經網路訓練不起來怎麼辦 (四)：損失函數 (Loss) 也可能有影響

第一步: 使用 Sigmoid, ReLU

下一層的權重和偏差會在訓練的過程中根據從上一層處理後的激活輸出來進行調整

訓練時，會把數據分成 batch(批次)，分批次處理
可以讓 GPU 有限的記憶體逐步計算、透過隨機性增加模型的泛化能力，幫助跳出局部最小值

通常會使用大小適中（如 32、64、128）的批次，梯度更新較穩定，訓練收斂更快

PS 為什麼訓練時要用Batch？
大的Batch : 假設全部資料當一個Batch，看完所有資料才update一次，很久才能更新一次但更新的很穩
小的Batch，假設每１個資料當一個Batch，看完一個資料就update一次，很快就能更新一次但更新的不穩

Batch Size為1至1000，所花的時間都差不多，因為GPU有平行運算的能力，但當Batch Size很大很大的時候，時間還是會增加
Batch Size越小：一次update很快，但因為要update次數多，跑一次Epoch愈慢。如果有考慮平行運算，小的Batch要update多次，反而比較慢
Batch Size越大：一次update很慢，但因為要update次數少，跑一次Epoch愈快。如果沒有考慮平行運算，大的Batch比較慢

*shuffle=True 會讓每一圈epoch，就重新分batch

ReLU (Rectified Linear Unit，”修正線性單元”)
計算效率高，與其他激活函數activation function（例如 sigmoid 或 tanh）相比，ReLU 的計算只需要一個簡單的比較和選擇
非線性： 雖然定義看起來很簡單，但 ReLU 引入了非線性，這讓神經網絡能夠學習更複雜的模型

比較

• Sigmoid，當 x 很大時（正無限大），輸出會接近 1；當 x 很小時（負無限大），輸出會接近 0 。可能會導致梯度消失問題，尤其在深層網絡的反向傳播過程中，梯度隨著層數增多而持續變小，從而影響到權重更新
• ReLU，當 x > 1，輸出為 1；當 x <= 0，輸出為 0 。由於在 x > 0 的情況下梯度保持為 1，所以在這些區域中，不會出現梯度消失的問題

第二步: 定義 誤差(Loss)

第三步: 優化

其中線性有可能會出現 Linear Model Bias
線性的規律為X變大，Y也會變大，有時無法符合真實情況，偏差（Bias）就會過大，而出現欠擬合（Underfitting）現象

測驗結果

Loss 高時，並非都是over fitting，有許多不同的情況

• 當 training data Loss 高時 (underfitting)

  → 可能是 model bias（模型偏差大），代表模型過於簡單，無法有效學習數據模式，這時候需要讓模型變得更複雜，增加參數

→ 要做 optimization，此門課只會使用Gradient Descent，找不到 = 大海撈針，撈不到，更改batch size、增加訓練次數（epochs）等

PS 怎麼分辨兩者？
在訓練的時候，越複雜的模型應該要越契合訓練資料，甚至會overfitting，因此
model bias : 56 layer v.s. 20 layer的測試資料，20層的Loss<56層的Loss **optimization** : 56 layer v.s. 20 layer 的訓練資料，20層的Loss>56層的Loss

• 當 training data Loss 低，但 testing data Loss 高時

  → 可能是 overfitting，這時候可以增加 raw_data 筆數、簡化模型（減少參數、增加正則化 regularization，如 L1/L2 正則化）、early stopping、Dropout(隨機讓部分神經元在訓練時失效，權重變為0)、數據增強 augmentation (在不收集新數據的情況下，透過對原始數據進行變換或擴增，生成新的樣本)、給model一些限制，最好是跟題目要求有關的限制。ex：已知題目可能會是2次函數，就把model限制在2次，給模型太大的限制，可能會回到Model Bias的問題

→ 數據分佈問題，可能是測試集與訓練集數據分佈不同

正則化 (Regularization)，主要用來防止Overfitting

常見的正則化方法有：
1️⃣ L1 正則化（Lasso Regression）
2️⃣ L2 正則化（Ridge Regression）
3️⃣ L1 + L2 正則化（Elastic Net）
4️⃣ Dropout（適用於神經網絡）
5️⃣ Early Stopping（提前停止訓練）
6️⃣ Batch Normalization（批次正規化）

N-fold Cross Validation

假設我有三個數據，五個ReLU

假設我有十五個數據，五個ReLU，三個batch

HW 1 : COVID-19 Cases Prediction

以前兩天各種指標，預測第三天測出陽性的比例

# Numerical Operations
import math
import numpy as np

# Reading/Writing Data
import pandas as pd
import os
import csv

# For Progress Bar
from tqdm import tqdm

# Pytorch
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader, random_split

# For plotting learning curve
from torch.utils.tensorboard import SummaryWriter
import matplotlib.pyplot as plt
from matplotlib.pyplot import figure

def get_device():
    ''' Get device (if GPU is available, use GPU) '''
    return 'cuda' if torch.cuda.is_available() else 'cpu'

# 隨機種子
def same_seed(seed):
    '''Fixes random number generator seeds for reproducibility'''
    torch.backends.cudnn.deterministic = True # 在相同輸入下輸出固定結果
    torch.backends.cudnn.benchmark = False # 防止 CuDNN 根據輸入大小自動調整算法
    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):
    '''Split provided training data into training set and validation set'''
    # 確保 valid_ratio 在 0~1
    if not (0 < valid_ratio < 1):
        raise ValueError("valid_ratio 必須在 0 和 1 之間")

    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)

# 切分訓練和驗證集*2
def train_2valid_split(data_set, valid_ratio1, valid_ratio2, seed):
    if not (0 < valid_ratio1 < 1 and 0 < valid_ratio2 < 1 and valid_ratio1 + valid_ratio2 < 1):
        raise ValueError("valid_ratio1、valid_ratio2 必須在 0 和 1 之間，且總和不能超過 1")
    valid1_size = int(valid_ratio1 * len(data_set))   # 第一個驗證集比例
    valid2_size = int(valid_ratio2 * len(data_set))   # 第二個驗證集比例
    train_size = len(data_set) - valid1_size - valid2_size  # 剩下的作為訓練集

    train_set, valid1_set, valid2_set = random_split(data_set, [train_size, valid1_size, valid2_size], generator=torch.Generator().manual_seed(seed))  
    return (np.array(train_set),  np.array(valid1_set), np.array(valid2_set))


# 切分訓練和驗證集和測試集
def train_valid_test_split(data_set, valid_ratio, test_ratio, seed):
    if not (0 < valid_ratio < 1 and 0 < test_ratio < 1 and valid_ratio + test_ratio < 1):
        raise ValueError("valid_ratio 和 test_ratio 必須在 0 和 1 之間，且總和不能超過 1")    
    test_size = int(test_ratio * len(data_set))   # 測試集比例
    valid_size = int(valid_ratio * len(data_set)) # 驗證集比例
    train_size = len(data_set) - valid_size - test_size  # 剩下的作為訓練集
    train_set, valid_set, test_set = random_split(data_set, [train_size, valid_size, test_size], generator=torch.Generator().manual_seed(seed)) 
    return np.array(train_set), np.array(valid_set), np.array(test_set)


# 切分訓練和驗證集*2和測試集
def train_2valid_test_split(data_set, valid_ratio1, valid_ratio2, test_ratio, seed):
    if not (0 < valid_ratio1 < 1 and 0 < valid_ratio2 < 1 and 0 < test_ratio < 1 and valid_ratio1 + valid_ratio2 + test_ratio < 1):
        raise ValueError("valid_ratio1、valid_ratio2 和 test_ratio 必須在 0 和 1 之間，且總和不能超過 1")
    test_size = int(test_ratio * len(data_set))       # 測試集比例
    valid1_size = int(valid_ratio1 * len(data_set))   # 第一個驗證集比例
    valid2_size = int(valid_ratio2 * len(data_set))   # 第二個驗證集比例
    train_size = len(data_set) - valid1_size - valid2_size - test_size  # 剩下的作為訓練集

    train_set, valid1_set, valid2_set, test_set = random_split(data_set, [train_size, valid1_size, valid2_size, test_size], generator=torch.Generator().manual_seed(seed))  
    return (np.array(train_set),  np.array(valid1_set), np.array(valid2_set), np.array(test_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()) # 轉換到 CPU，避免 GPU 內存累積
    preds = torch.cat(preds, dim=0).numpy()
    return preds

# 選擇訓練、驗證和測試數據中的特徵，以便於後續的回歸分析

def select_feat(train_data, valid_data, test_data, select_all=True):
    '''Selects useful features to perform regression'''
    # 目標變數
    y_train, y_valid = train_data[:,-1], valid_data[:,-1]
    # 所有特徵，最後一列 (-1 列) 以外的所有列，通常最後一列為目標變數 (target label)
    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] # TODO: Select suitable feature columns.

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

# 資料集
# 處理因變數、自變數

class COVID19Dataset(Dataset):
    '''
    x: Features.
    y: Targets, if none, do prediction.
    '''
    def __init__(self, x, y=None):
        if y is None:
            self.y = y
        else:
            self.y = torch.FloatTensor(y) # 轉換為 FloatTensor
        self.x = torch.FloatTensor(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, 100),  # 第一層隱藏層，將輸入特徵映射到 100 維
            nn.ReLU(),
            nn.Linear(100, 100),  # 第二層隱藏層，100 維
            nn.ReLU(),
            nn.Linear(100, 100),  # 第三層隱藏層，100 維
            nn.ReLU(),
            nn.Linear(100, 64),  # 第三層隱藏層，100 維
            nn.ReLU(),
            nn.Linear(64, 32),  # 第四層隱藏層，100 維
            nn.ReLU(),
            nn.Linear(32, 16),  # 第五層隱藏層，100 維
            nn.ReLU(),
            nn.Linear(16, 1)  # 輸出 1 維結果
        )

    def forward(self, x):
        x = self.layers(x)
        x = x.squeeze(1)  # (B, 1) -> (B)
        return x

# 2022

# Training

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

    criterion = nn.MSELoss(reduction='mean') # Define your loss function, do not modify this. # 定義均方誤差損失函數

    # Define your optimization algorithm.
    # TODO: Please check https://pytorch.org/docs/stable/optim.html to get more available algorithms.
    # TODO: L2 regularization (optimizer(weight decay...) or implement by your self).

    # 定義隨機梯度下降優化器
    optimizer = torch.optim.Adam(model.parameters(), lr=config['learning_rate']) # , momentum=0.9

    writer = SummaryWriter() # Writer of tensoboard. # 用於記錄訓練過程，方便可視化

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

    # 每 step_size=10 個 epoch，學習率 lr 會乘以 gamma=0.1，讓學習率逐漸降低，提高收斂穩定性
    scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.1)

    # 訓練 n_epochs 次、保存訓練過程中最低的驗證損失、計算訓練過程中沒有提升的輪數（epochs）
    # 如果模型在一定數量的輪次中沒有改進（即驗證損失不再減少），則進行早停（early stopping），
    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)

        # 記錄train data損失
        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() # 設置模型為評估模式

        # 記錄預測結果
        preds = []
        for x, _ in valid_loader:  # 用valid_loader代替test_loader
          with torch.no_grad():  # 不計算梯度
              x = x.to(device)
              pred = model(x)
              preds.append(pred.cpu().numpy())  # 將預測結果轉移到 CPU 並保存
        preds = np.concatenate(preds, axis=0)  # 合併所有預測結果

        # 記錄valid data損失
        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)

        # 如果驗證損失數值較低，保存數據，否則 停止的數字＋1
        if mean_valid_loss < best_loss:
            best_loss = mean_valid_loss
            torch.save(model.state_dict(), config['save_path']) # best model
            print('loss {:.3f}...'.format(best_loss))
            early_stop_count = 0
        else:
            early_stop_count += 1

        # 如果超過提前停止的次數，停止訓練
        if early_stop_count >= config['early_stop']:
            print('\nEarly stop')
            return
    return preds

Colab

# 安裝 TensorFlow 和 PyTorch
!pip install tensorflow
!pip install torch
!pip install torch torchvision

# 顯示版本
import tensorflow as tf
import torch

print("TensorFlow version:", tf.__version__)
print("PyTorch version:", torch.__version__)

torch.cuda.is_available()

!nvidia-smi

下載資料

!gdown --id '1kLSW_-cW2Huj7bh84YTdimGBOJaODiOS' --output covid.train.csv
!gdown --id '1iiI5qROrAhZn-o4FPqsE97bMzDEFvIdg' --output covid.test.csv

df_train_covid_train = pd.read_csv('covid.train.csv')

columns_name = []
for c in df_train_covid_train.columns:
  columns_name.append(c)

print(columns_name)
print(df_train_covid_train.shape)
df_train_covid_train

最後一列為目標變數，要預測的結果
AL, AK, AZ, …, WA（美國各州縮寫）
下面都以％表示
cli（COVID-Like Illness）：類似 COVID-19 的症狀
ili（Influenza-Like Illness）：類似流感的症狀
hh_cmnty_cli（Household Community CLI）：家戶內社區感染數
nohh_cmnty_cli（Non-Household Community CLI）：非家戶內社區感染數
wearing_mask：是否戴口罩
travel_outside_state：是否有跨州旅行
work_outside_home：是否在家工作（1 代表在家工作，0 代表需外出）
shop：是否去過超市或商店
restaurant：是否去過餐廳
spent_time：在外停留時間
large_event：是否參加大型活動
public_transit：是否使用大眾運輸（如公車、捷運）
anxious：是否感到焦慮
depressed：是否感到沮喪
worried_finances：是否擔心財務狀況
tested_positive 是否確診 COVID-19
tested_positive.1 代表前一天是否確診
tested_positive.2 代表前兩天是否確診
tested_positive.3 代表前三天是否確診
tested_positive.4 代表前四天是否確診

因為載的dataset沒有test的解答，這裡不使用
改成先切出 10% test data，剩下再分 80% train data + 20% vaild data

device = 'cuda' if torch.cuda.is_available() else 'cpu'
config = {
    'seed': 77777,
    'select_all': True,
    'valid_ratio': 0.2,   # validation_size = train_size * valid_ratio
    'n_epochs': 3000,
    'batch_size': 256,
    'learning_rate': 1e-5, # 較小的學習率有助於穩定收斂，但可能會使訓練變慢
    'early_stop': 400,
    'save_path': './models/model.ckpt'
}

# same_seed
same_seed(config['seed'])

#
train_data_ori = pd.read_csv('./covid.train.csv').values
# train_data_ori = df_train_covid_train_new.values
np.random.seed(777)
np.random.shuffle(train_data_ori)

# 先把 10% 切出作為測試數據
test_data_size = int(len(train_data_ori) * 0.1)
test_data = train_data_ori[:test_data_size, :-1]
# 10%的正確結果先存
test_data_y_hat = train_data_ori[:test_data_size, -1]
test_data_y_hat_df = pd.DataFrame(test_data_y_hat, columns=['tested_positive'])
test_data_y_hat_df.to_csv('test_data_y_hat.csv', index=False)

train_data = train_data_ori[test_data_size:]

# 剩下再分 80% train data + 20% vaild data
train_data, valid_data = train_valid_split(train_data, config['valid_ratio'], config['seed'])

print(f"""
train_data_ori size: {train_data_ori.shape}
train_data size: {train_data.shape}
valid_data size: {valid_data.shape}
test_data size: {test_data.shape}
test_data_y_hat size: {test_data_y_hat.shape}
      """)

# Select features
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]}')

# COVID19Dataset
train_dataset, valid_dataset, test_dataset = COVID19Dataset(x_train, y_train), \
                                            COVID19Dataset(x_valid, y_valid), \
                                            COVID19Dataset(x_test)

# Pytorch
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)

開始訓練

# 2022
model = My_Model(input_dim=x_train.shape[1]).to(device)
trainer(train_loader, valid_loader, model, config, device)

測試並存檔

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')

準確率

# 檔案對檔案對比

import pandas as pd
from sklearn.metrics import mean_squared_error, r2_score

# 預測結果
pred_df = pd.read_csv('pred.csv')
preds = pred_df['tested_positive'].values

# 真實標籤
test_data_y_hat_df = pd.read_csv('test_data_y_hat.csv')
test_data_y_hat = test_data_y_hat_df['tested_positive'].values

# MSE
mse = mean_squared_error(test_data_y_hat, preds)
print(f'Mean Squared Error (MSE): {mse}')

# R² (決定係數)
r2 = r2_score(test_data_y_hat, preds)
print(f"R²: {r2}")

改成Adam，需要時間比較多

在這裡結果沒有比較好