สัปดาห์ที่ 11
PyTorch: RNN, LSTM, และ Sequence Models
1. Timeseries Data คืออะไร
2. Windowing และ Sliding Window
3. PyTorch Dataset สำหรับ Timeseries
4. RNN และ LSTM
5. Training Loop สำหรับ Sequence
6. ตัวอย่างจริง: Temperature Prediction
- Time series: ข้อมูลที่เรียงตามเวลา
- ตัวอย่าง: ราคาหุ้น, อุณหภูมิ, ยอดขาย, sensor data
- แต่ละค่ามี dependency กับค่าก่อนหน้า
- ต่างจาก tabular data ที่ไม่เรียงตามลำดับ
[สำคัญ] ห้ามสุ่ม (shuffle) ข้อมูล time series เพราะลำดับเวลาสำคัญ!
Time: 0 1 2 3 4 5
Value: [10] [12] [15] [13] [18] [20]
| | | | | |
v v v v v v
trend upward
# Python representation
import pandas as pd
data = [10, 12, 15, 13, 18, 20]
ts = pd.Series(data,
index=pd.date_range('2024-01-01',
periods=6, freq='D'))
print(ts)
# 2024-01-01 10
# 2024-01-02 12
# 2024-01-03 15
# ...
- Trend: แนวโน้มระยะยาว (up/down/flat)
- Seasonality
- Noise: ความผันผวนแบบสุ่ม - Stationary: ค่าเฉลี่ยและ variance คงที่ [เคล็ดลับ] Stationary data ทำนายง่ายกว่า อาจต้อง differencing หรือ log transform ก่อน
# Decompose time series
from statsmodels.tsa.seasonal import seasonal_decompose
result = seasonal_decompose(ts, model='additive',
period=7)
result.plot()
# Components:
# - trend: [10, 12, 14, 16, 18, 20, 22]
# - seasonal: [0, +2, +1, -2, +1, +3, -1]
# - residual: [0, -1, 0, +1, -1, 0, +1]
# Make stationary
ts_diff = ts.diff().dropna() # differencing
- Windowing: ใช้ค่าที่ผ่านมาทำนายค่าในอนาคต
- lookback: จำนวน step ย้อนหลัง
- delay: จำนวน step ที่จะทำนาย
- stride: ระยะห่างระหว่าง sample
[ตัวอย่าง] lookback=5 หมายถึง ใช้ 5 วันก่อนหน้าทำนายวันถัดไป
# Data: [10, 12, 15, 13, 18, 20, 22, 25]
# lookback=3, delay=1
# Sample 1: [10, 12, 15] -> 13
# Sample 2: [12, 15, 13] -> 18
# Sample 3: [15, 13, 18] -> 20
# Sample 4: [13, 18, 20] -> 22
# Sample 5: [18, 20, 22] -> 25
def create_dataset(data, lookback, delay):
X, y = [], []
for i in range(len(data) - lookback - delay):
X.append(data[i:i+lookback])
y.append(data[i+lookback+delay-1])
return np.array(X), np.array(y)
X, y = create_dataset(data, 3, 1)
# X = [[10,12,15], [12,15,13], ...]
# y = [13, 18, ...]
import torch
from torch.utils.data import Dataset, DataLoader
import numpy as np
class TimeSeriesDataset(Dataset):
def __init__(self, data, lookback):
self.data = torch.FloatTensor(data)
self.lookback = lookback
def __len__(self):
return len(self.data) - self.lookback
def __getitem__(self, idx):
x = self.data[idx:idx+self.lookback]
y = self.data[idx+self.lookback]
return x, y
# Usage
data = np.array([10,12,15,13,18,20,22,25,28,30])
dataset = TimeSeriesDataset(data, lookback=3)
loader = DataLoader(dataset, batch_size=2,
shuffle=False) # ไม่ shuffle!
for X_batch, y_batch in loader:
print(X_batch.shape, y_batch.shape)
# torch.Size([2, 3]) torch.Size([2])
- Dataset: กำหนดวิธีดึงข้อมูลแต่ละ sample
- DataLoader: จัด batching และ iteration
- ห้าม shuffle! ต้องรักษาลำดับเวลา
- ข้อมูลต้องเป็น Tensor หรือ numpy array
[สำคัญ] ต้อง normalize ก่อนสร้าง Dataset (MinMaxScaler หรือ StandardScaler)
from sklearn.preprocessing import MinMaxScaler
# Normalize data
scaler = MinMaxScaler()
data_scaled = scaler.fit_transform(
data.reshape(-1, 1)).flatten()
# Split train/test (ห้ามสุ่ม!)
train_size = int(len(data_scaled) * 0.8)
train_data = data_scaled[:train_size]
test_data = data_scaled[train_size - lookback:]
# Create datasets
train_dataset = TimeSeriesDataset(train_data, lookback)
test_dataset = TimeSeriesDataset(test_data, lookback)
# Denormalize predictions
pred_original = scaler.inverse_transform(
pred_scaled.reshape(-1, 1))
- MinMaxScaler: ปรับค่าอยู่ระหว่าง [0, 1]
- StandardScaler: ปรับค่าให้ mean=0, std=1
- ห้าม fit บน test data! ต้อง fit บน train เท่านั้น
- เก็บ scaler ไว้ denormalize ผลทำนาย
[สำคัญ] Data leakage = ใช้ข้อมูลอนาคตตอน train ทำให้ผลทำนายดีเกินจริง
- RNN มี memory (hidden state) ที่ส่งต่อระหว่าง time step
- \(h_t = \tanh(W_{ih} x_t + b_{ih} + W_{hh} h_{t-1} + b_{hh})\)
- Vanilla RNN: มีปัญหา vanishing gradient
- จำได้ระยะสั้น ลืมระยะยาว
[ปัญหา] RNN จำ long-term dependencies ได้ไม่ดี = ต้องใช้ LSTM/GRU
import torch.nn as nn
class SimpleRNN(nn.Module):
def __init__(self, input_size, hidden_size,
output_size):
super().__init__()
self.rnn = nn.RNN(input_size, hidden_size,
batch_first=True)
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# x: (batch, seq_len, input_size)
out, hn = self.rnn(x)
# out: (batch, seq_len, hidden_size)
out = self.fc(out[:, -1, :])
return out
model = SimpleRNN(input_size=1, hidden_size=32,
output_size=1)
print(model)
# SimpleRNN(
# (rnn): RNN(1, 32, batch_first=True)
# (fc): Linear(32, 1)
# )
class LSTMModel(nn.Module):
def __init__(self, input_size, hidden_size,
num_layers, output_size):
super().__init__()
self.lstm = nn.LSTM(input_size, hidden_size,
num_layers=num_layers,
batch_first=True)
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# x: (batch, seq_len, input_size)
out, (hn, cn) = self.lstm(x)
# out: (batch, seq_len, hidden_size)
out = self.fc(out[:, -1, :])
return out
model = LSTMModel(input_size=1, hidden_size=64,
num_layers=2, output_size=1)
print(model)
# LSTMModel(
# (lstm): LSTM(1, 64, num_layers=2,
# batch_first=True)
# (fc): Linear(64, 1)
# )
- LSTM มี cell state ที่ช่วยจำ long-term dependencies
- มี 3 gates: forget, input, output
- num_layers: จำนวน LSTM layers ซ้อนกัน
- out[:, -1, :]: ใช้ hidden state ตัวสุดท้าย
[เคล็ดลับ] LSTM ดีกว่า RNN สำหรับข้อมูลที่มี long-term dependencies
model = LSTMModel(input_size=1, hidden_size=64,
num_layers=2, output_size=1)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
for epoch in range(100):
model.train()
train_loss = 0
for X_batch, y_batch in train_loader:
X_batch = X_batch.unsqueeze(-1) # (B, T, 1)
y_batch = y_batch.unsqueeze(-1) # (B, 1)
pred = model(X_batch)
loss = criterion(pred, y_batch)
optimizer.zero_grad()
loss.backward()
optimizer.step()
train_loss += loss.item()
# Evaluate
model.eval()
with torch.no_grad():
test_loss = 0
for X_batch, y_batch in test_loader:
X_batch = X_batch.unsqueeze(-1)
pred = model(X_batch)
test_loss += criterion(pred, y_batch.unsqueeze(-1)).item()
if epoch % 10 == 0:
print(f"Epoch {epoch}: "
f"train={train_loss/len(train_loader):.4f} "
f"test={test_loss/len(test_loader):.4f}")
- unsqueeze(-1): เพิ่มมิติ input_size = 1
- ห้าม shuffle ใน DataLoader
- ใช้ MSELoss สำหรับ regression
- ต้อง model.eval() ตอน evaluate
[สำคัญ] เปรียบเทียบ train/test loss เพื่อดู overfitting
- GRU: เวอร์ชันที่ง่ายกว่า LSTM
- มี 2 gates: reset และ update
- เร็วกว่า LSTM, ผลลัพธ์ใกล้เคียงกัน
- ใช้เมื่อ dataset ไม่ใหญ่มาก
[เปรียบเทียบ] LSTM จำได้ดีกว่า / GRU เร็วกว่า ขึ้นกับปัญหา
class GRUModel(nn.Module):
def __init__(self, input_size, hidden_size,
num_layers, output_size):
super().__init__()
self.gru = nn.GRU(input_size, hidden_size,
num_layers=num_layers,
batch_first=True)
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
out, hn = self.gru(x)
out = self.fc(out[:, -1, :])
return out
# Comparison:
# RNN: 1 gate, fastest, worst memory
# GRU: 2 gates, fast, good balance
# LSTM: 3 gates, slower, best memory
- ข้อมูลอุณหภูมิจาก Max Planck Institute, Germany
- 14 features, ทุก 10 นาที, ปี 2009-2016
- ทำนาย temperature จากค่าที่ผ่านมา
- ตัวอย่างจริงที่ใช้บ่อยใน textbooks
[ข้อมูล] 420,485 ช่วงเวลา, contiene NaN values
import pandas as pd
df = pd.read_csv('jena_climate_2009_2016.csv')
print(df.head())
# Date Time p (mbar) T (degC) ...
# 0 01.01.2009 00:10 996.52 -8.02
# 1 01.01.2009 00:20 996.51 -8.41
# ...
# Select temperature column
temp = df['T (degC)'].values
# Visualize
import matplotlib.pyplot as plt
plt.figure(figsize=(12, 4))
plt.plot(temp[:1440]) # 10 days
plt.xlabel('Time (10 min intervals)')
plt.ylabel('Temperature (C)')
plt.title('Jena Temperature')
plt.show()
import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader
from sklearn.preprocessing import MinMaxScaler
# Load and preprocess
df = pd.read_csv('jena_climate_2009_2016.csv')
temp = df['T (degC)'].values.astype(np.float32)
# Normalize
scaler = MinMaxScaler()
temp_scaled = scaler.fit_transform(
temp.reshape(-1, 1)).flatten()
# Train/test split (chronological!)
train_size = 200000
train_data = temp_scaled[:train_size]
test_data = temp_scaled[train_size-60:]
# Dataset
class TimeSeriesDataset(Dataset):
def __init__(self, data, lookback):
self.data = torch.FloatTensor(data)
self.lookback = lookback
def __len__(self):
return len(self.data) - self.lookback
def __getitem__(self, i):
return (self.data[i:i+self.lookback],
self.data[i+self.lookback])
LOOKBACK = 60
train_ds = TimeSeriesDataset(train_data, LOOKBACK)
test_ds = TimeSeriesDataset(test_data, LOOKBACK)
train_loader = DataLoader(train_ds, 256, shuffle=False)
test_loader = DataLoader(test_ds, 256, shuffle=False)
- lookback=60: ใช้ 60 step (10 ชั่วโมง) ทำนาย step ถัดไป
- ข้อมูล train 200K, test ~220K
- shuffle=False สำหรับ train และ test
- ต้องทำ normalization ก่อน split
[สำคัญ] ห้าม fit scaler บน test data!
class LSTM(nn.Module):
def __init__(self):
super().__init__()
self.lstm = nn.LSTM(1, 64, 2, batch_first=True)
self.fc = nn.Linear(64, 1)
def forward(self, x):
out, _ = self.lstm(x)
return self.fc(out[:, -1, :])
model = LSTM()
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
for epoch in range(20):
model.train()
train_loss = 0
for X_b, y_b in train_loader:
X_b = X_b.unsqueeze(-1) # (B, T, 1)
pred = model(X_b)
loss = criterion(pred, y_b.unsqueeze(-1))
optimizer.zero_grad()
loss.backward()
optimizer.step()
train_loss += loss.item()
model.eval()
test_loss = 0
with torch.no_grad():
for X_b, y_b in test_loader:
X_b = X_b.unsqueeze(-1)
pred = model(X_b)
test_loss += criterion(pred, y_b.unsqueeze(-1)).item()
print(f"Epoch {epoch}: "
f"train={train_loss/len(train_loader):.5f} "
f"test={test_loss/len(test_loader):.5f}")
- LSTM(1, 64, 2): input=1, hidden=64, layers=2
- ใช้ Adam optimizer, lr=0.001
- 20 epochs พอสำหรับ demo
- ดู train/test loss เทียบกัน
[เคล็ดลับ] ถ้า test loss สูงกว่า train มาก = overfitting ลองเพิ่ม dropout
model.eval()
with torch.no_grad():
X_test = test_loader.dataset.data
X_test = X_test.unsqueeze(-1).unsqueeze(0)
pred_scaled = model(X_test).squeeze().numpy()
# Denormalize
pred = scaler.inverse_transform(
pred_scaled.reshape(-1, 1)).flatten()
actual = scaler.inverse_transform(
test_loader.dataset.data.numpy()[
LOOKBACK:].reshape(-1, 1)).flatten()
# Plot
plt.figure(figsize=(12, 4))
plt.plot(actual, label='Actual')
plt.plot(pred, label='Predicted')
plt.legend()
plt.xlabel('Time')
plt.ylabel('Temperature (C)')
plt.title('LSTM Prediction vs Actual')
plt.show()
# Metrics
from sklearn.metrics import mean_absolute_error
mae = mean_absolute_error(actual, pred)
print(f"MAE: {mae:.2f} C")
- ต้อง model.eval() ก่อน evaluate
- ใช้ torch.no_grad() ประหยัด memory
- Denormalize กลับเป็นค่าจริงก่อน plot
- ใช้ MAE (Mean Absolute Error) เป็นเมตริก
[เมตริก] MAE ยิ่งต่ำยิ่งดี, 0 หมายถึงทำนายตรงเป๊ะ
- Single-step: ทำนาย 1 ค่าถัดไป
- Multi-step: ทำนายหลายค่าล่วงหน้า
- ใช้ผลทำนายป้อนกลับเป็น input
- ยิ่งนาน ยิ่งผิดพลาดมากขึ้น
[ปัญหา] Error accumulation: ความผิดพลาดสะสมตามจำนวน step
# Single-step: predict next 1
pred_1 = model(last_window) # 1 value
# Multi-step: predict next N
def predict_multi(model, last_window, n_steps):
preds = []
current = last_window.clone()
for _ in range(n_steps):
with torch.no_grad():
pred = model(current.unsqueeze(0)
.unsqueeze(-1))
preds.append(pred.item())
# Shift and append prediction
current = torch.cat([
current[1:], pred.squeeze()
])
return preds
preds = predict_multi(model, test_window, 10)
print(preds)
# [12.3, 12.5, 12.8, 13.1, 13.4, ...]
uv add django torch scikit-learn
uv run django-admin startproject wk11 .
uv run manage.py startapp dashboard
# wk11/urls.py
from django.urls import path, include
urlpatterns = [
path('', include('dashboard.urls')),
]
# dashboard/views.py
import json, torch, torch.nn as nn
from django.http import StreamingHttpResponse
MODELS = {
'rnn': lambda: nn.RNN(1, 32, 1, batch_first=True),
'lstm': lambda: nn.LSTM(1, 32, 1, batch_first=True),
'gru': lambda: nn.GRU(1, 32, 1, batch_first=True),
}
def train_stream(request, model_type):
def event_stream():
model = MODELS[model_type]()
fc = nn.Linear(32, 1)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(
list(model.parameters()) + list(fc.parameters()), lr=0.01)
for epoch in range(51):
# ... training loop ...
yield f"data: {json.dumps({'epoch': epoch, 'loss': loss.item()})}\n\n"
return StreamingHttpResponse(event_stream(), content_type='text/event-stream')
- Django: web framework สำหรับ backend
- SSE (Server-Sent Events): real-time streaming
- เลือก model type: RNN, LSTM, GRU
- แต่ละ model train แยกกัน แสดงผล real-time
[โครงสร้าง] manage.py -> settings.py -> urls.py -> views.py -> templates/
from django.http import StreamingHttpResponse
import json, time
def train(request, model_type):
def event_stream():
# Load data (simplified)
X = torch.tensor([[0,0],[0,1],[1,0],[1,1]],
dtype=torch.float32)
y = torch.tensor([[0],[1],[1],[0]],
dtype=torch.float32)
# Build model based on selection
if model_type == 'rnn':
net = nn.RNN(2, 16, 1, batch_first=True)
elif model_type == 'lstm':
net = nn.LSTM(2, 16, 1, batch_first=True)
else:
net = nn.GRU(2, 16, 1, batch_first=True)
fc = nn.Linear(16, 1)
criterion = nn.BCELoss()
optimizer = torch.optim.Adam(
list(net.parameters()) + list(fc.parameters()),
lr=0.05)
for epoch in range(201):
out, _ = net(X.unsqueeze(0))
pred = torch.sigmoid(fc(out[:, -1, :]))
loss = criterion(pred, y)
optimizer.zero_grad()
loss.backward()
optimizer.step()
yield f"data: {json.dumps({'epoch': epoch, 'loss': loss.item(), 'pred': pred.squeeze().tolist()})}\n\n"
time.sleep(0.02)
yield f"data: {json.dumps({'type': 'done'})}\n\n"
resp = StreamingHttpResponse(event_stream(),
content_type='text/event-stream')
resp['Cache-Control'] = 'no-cache'
return resp
- StreamingHttpResponse: ส่ง data ทีละ epoch
- event_stream(): generator function
- แต่ละ line เป็น data: {...}\n\n format
- Client ใช้ EventSource รับข้อมูล
[สำคัญ] SSE = one-way (server -> client), ไม่ต้อง bidirectional
<!-- dashboard/templates/index.html -->
<div x-data="{ model: 'lstm', training: false }">
<select x-model="model">
<option value="rnn">RNN</option>
<option value="lstm">LSTM</option>
<option value="gru">GRU</option>
</select>
<button @click="startTrain()">
Start Training
</button>
<canvas id="chart" width="420" height="200">
</canvas>
</div>
<script>
function startTrain() {
const es = new EventSource('/train/' + model);
es.onmessage = function(e) {
const m = JSON.parse(e.data);
drawLoss(m.epoch, m.loss);
drawPred(m.pred);
};
}
</script>
- ใช้ <select> เลือก model type
- ใช้ <canvas> แสดงผล loss curve
- ใช้ EventSource รับ SSE updates
- วาด loss ทุก epoch แบบ real-time
[เคล็ดลับ] ใช้ Alpine.js สำหรับ reactive UI, ไม่ต้อง React/Vue
สร้าง LSTM model สำหรับทำนายอุณหภูมิ: