Transformer 학습 및 평가 시 training data와 test data를 다르게 처리 하는 이유
평가 시에 rollout 방식을 사용하는 것은 미래 예측을 위해 이전 예측값을 기반으로 순차적으로 진행하는 반면, 학습 단계에서는 그런 방식을 사용하지 않습니다. 그 이유는 다음과 같습니다.
1. Teacher Forcing 사용
학습 단계에서는 모델에게 실제 정답(ground truth)을 제공하는 teacher forcing 방식을 사용합니다.
이렇게 하면 모델이 올바른 시퀀스 패턴을 학습하는 데 도움이 되고, 오차가 누적되는 문제를 방지할 수 있습니다.
매 스텝마다 실제 값으로부터 학습함으로써, 모델은 더 안정적으로 수렴할 수 있습니다.
2. 병렬 처리 효율성
DataLoader로 독립적인 시퀀스를 배치 처리하면, 여러 샘플을 동시에 처리할 수 있어 학습 속도와 메모리 사용 효율이 크게 향상됩니다.
반면, rollout 방식은 순차적인 예측을 필요로 하므로 병렬 처리가 어려워집니다.
즉, 학습 시에는 전체 데이터셋에 대해 빠르고 안정적인 병렬 처리가 중요합니다.
3. 오차 누적 방지
rollout 방식에서는 모델이 이전 예측을 바탕으로 다음 예측을 생성하기 때문에, 초기 작은 오차가 계속 누적되어 전체 예측 품질에 영향을 줄 수 있습니다.
학습 단계에서는 이러한 오차 누적 문제를 피하기 위해 실제 데이터를 사용합니다.
요약하면, 학습 시에는 teacher forcing을 통해 실제 데이터를 활용하여 안정적이고 효율적으로 학습하고, 평가나 실제 예측 단계에서만 rollout 방식을 사용해 모델의 장기 예측 성능을 확인하는 방식으로 진행합니다.
(출처: chatGPT)
세부 코드
데이터 로딩 설정 값 및 하이퍼 파라미터 설정 값
Hyper Parameters: 모델의 학습 및 예측 프로세스에 큰 영향을 미치는 사전 설정 값
ticker = "TSLA"
start_date = "2020-01-01"
end_date = "2023-01-01"
seq_length = 50 # 입력 시퀀스 길이 (예: 50일)
batch_size = 32
epochs = 100
num_prediction = 50 # 예측할 미래 시점 개수
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")데이터 로딩과 전처리를 위해 load_and_preprocess_data() 함수를 호출하여 train_data, test_data, scaler를 리턴 받는다.
load_and_preprocess_data() 함수에서는 dataframe 형태로 다운로드되며 Close 피처만 사용하고 MinMaxScaler를 이용해 Clase data 기준으로 정규화한다.
전체 data 중 0.8만큼은 training data에 사용g하도록 분리하여 최종적으로 train_data, test_data, scaler을 리턴한다.
train_data, test_data, scaler = load_and_preprocess_data(
ticker, start_date, end_date, feature='Close', split_ratio=0.8
df = yf.download(ticker, start=start_date, end=end_date)
data = df[[feature]].values # shape: [num_samples, 1]
scaler = MinMaxScaler(feature_range=(0, 1))
data_scaled = scaler.fit_transform(data)
split_idx = int(len(data_scaled) * split_ratio)
train_data = data_scaled[:split_idx]
test_data = data_scaled[split_idx:]training data만 DataLoader()를 사용하여 로더한다. 위에서 말한 것과 같이 Transformer는 rollout 방식으로 초기 시퀀스 이후부터는 디코더의 출력이 다시 인코더의 입력으로 사용되므로 DataLoader를 통해 생성된 Data쌍 형태로 만들지 않아도 된다. DataLoader를 통해 생성된 입력-출력(정답) 형태의 데이터는 효율적인 학습을 위해 training data에만 이용된다.
# DataLoader 생성
train_loader = create_dataloader(train_data, seq_length, batch_size)def create_dataloader(data, seq_length, batch_size, shuffle=True):
dataset = TimeSeriesDataset(data, seq_length)
loader = DataLoader(dataset, batch_size=batch_size, shuffle=shuffle)
return loaderclass TimeSeriesDataset(Dataset):
def __init__(self, data, seq_length):
self.data = torch.tensor(data, dtype=torch.float32)
self.seq_length = seq_length
def __len__(self):
return len(self.data) - self.seq_length
def __getitem__(self, idx):
# src: seq_length 시퀀스, tgt: 그 다음 시점의 값
src = self.data[idx:idx + self.seq_length] # [seq_length, input_dim]
tgt = self.data[idx + self.seq_length] # [input_dim]
return src, tgt전체 코드
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset
import numpy as np
import matplotlib.pyplot as plt
import yfinance as yf
from sklearn.preprocessing import MinMaxScaler
# 1. 데이터 로딩 및 전처리
def load_and_preprocess_data(ticker, start_date, end_date, feature='Close', split_ratio=0.8):
"""
yfinance를 사용하여 주가 데이터를 다운로드하고, 지정한 feature를 선택 후 정규화 및 학습/테스트 분할을 수행합니다.
"""
df = yf.download(ticker, start=start_date, end=end_date)
data = df[[feature]].values # shape: [num_samples, 1]
scaler = MinMaxScaler(feature_range=(0, 1))
data_scaled = scaler.fit_transform(data)
split_idx = int(len(data_scaled) * split_ratio)
train_data = data_scaled[:split_idx]
test_data = data_scaled[split_idx:]
return train_data, test_data, scaler
# 2. Dataset 및 DataLoader 구성
class TimeSeriesDataset(Dataset):
def __init__(self, data, seq_length):
self.data = torch.tensor(data, dtype=torch.float32)
self.seq_length = seq_length
def __len__(self):
return len(self.data) - self.seq_length
def __getitem__(self, idx):
# src: seq_length 시퀀스, tgt: 그 다음 시점의 값
src = self.data[idx:idx + self.seq_length] # [seq_length, input_dim]
tgt = self.data[idx + self.seq_length] # [input_dim]
return src, tgt
def create_dataloader(data, seq_length, batch_size, shuffle=True):
dataset = TimeSeriesDataset(data, seq_length)
loader = DataLoader(dataset, batch_size=batch_size, shuffle=shuffle)
return loader
# 3. Transformer 기반 시계열 예측 모델 정의
class TimeSeriesTransformer(nn.Module):
def __init__(self, input_dim, d_model, nhead, num_layers, dropout=0.1):
super(TimeSeriesTransformer, self).__init__()
self.embedding = nn.Linear(input_dim, d_model)
self.transformer = nn.Transformer(
d_model=d_model,
nhead=nhead,
num_encoder_layers=num_layers,
num_decoder_layers=num_layers,
dropout=dropout
)
self.fc_out = nn.Linear(d_model, input_dim)
def forward(self, src, tgt):
# src: [S, N, input_dim], tgt: [T, N, input_dim]
src_emb = self.embedding(src) # [S, N, d_model]
tgt_emb = self.embedding(tgt) # [T, N, d_model]
output = self.transformer(src_emb, tgt_emb) # [T, N, d_model]
return self.fc_out(output) # [T, N, input_dim]
# 4. 모델 학습 함수
def train_model(model, train_loader, device, epochs, learning_rate=0.001):
model.train()
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)
for epoch in range(epochs):
epoch_loss = 0
for src, tgt in train_loader:
src = src.to(device) # [batch, seq_length, input_dim]
tgt = tgt.to(device) # [batch, input_dim]
# Transformer 입력은 [sequence_length, batch, feature]
src = src.transpose(0, 1) # [seq_length, batch, input_dim]
# 디코더 입력: 단일 스텝 예측을 위해 0 벡터 사용 (shape: [1, batch, input_dim])
tgt_decoder = torch.zeros(1, src.size(1), src.size(2)).to(device)
optimizer.zero_grad()
output = model(src, tgt_decoder) # [1, batch, input_dim]
output = output.squeeze(0) # [batch, input_dim]
loss = criterion(output, tgt)
loss.backward()
optimizer.step()
epoch_loss += loss.item()
print(f"Epoch {epoch+1}/{epochs}, Loss: {epoch_loss / len(train_loader):.6f}")
# 5. 미래 주가 예측 (rollout 방식)
def predict_future(model, test_data, seq_length, num_prediction, device):
model.eval()
# 초기 입력: 테스트 데이터의 처음 seq_length 값
test_input = torch.tensor(test_data[:seq_length], dtype=torch.float32).to(device)
predictions = []
with torch.no_grad():
for _ in range(num_prediction):
src = test_input.unsqueeze(1) # [seq_length, 1, input_dim]
tgt_decoder = torch.zeros(1, 1, test_input.size(-1)).to(device)
out = model(src, tgt_decoder) # [1, 1, input_dim]
pred = out.squeeze(0).squeeze(0).item()
predictions.append(pred)
# 시퀀스 업데이트: 가장 오래된 값 제거 후 예측값 추가
test_input = torch.cat([test_input[1:], torch.tensor([[pred]], dtype=torch.float32).to(device)], dim=0)
return predictions
# 6. 결과 시각화 함수
def plot_predictions(actual, predictions, seq_length):
plt.figure(figsize=(12,6))
plt.plot(actual, label="Actual Close Price")
plt.plot(range(seq_length, seq_length + len(predictions)), predictions,
label="Predicted Close Price", linestyle="dashed")
plt.xlabel("Time Step")
plt.ylabel("Price")
plt.title("Tesla Stock Price Prediction using Transformer")
plt.legend()
plt.show()
# 7. 메인 실행 함수
def main():
# 설정값
ticker = "TSLA"
start_date = "2020-01-01"
end_date = "2023-01-01"
seq_length = 50 # 입력 시퀀스 길이 (예: 50일)
batch_size = 32
epochs = 100
num_prediction = 50 # 예측할 미래 시점 개수
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 데이터 로딩 및 전처리
train_data, test_data, scaler = load_and_preprocess_data(
ticker, start_date, end_date, feature='Close', split_ratio=0.8
)
# DataLoader 생성
train_loader = create_dataloader(train_data, seq_length, batch_size)
# 모델 생성
model = TimeSeriesTransformer(input_dim=1, d_model=64, nhead=4, num_layers=3).to(device)
# 모델 학습
train_model(model, train_loader, device, epochs, learning_rate=0.001)
# 미래 주가 예측
predictions = predict_future(model, test_data, seq_length, num_prediction, device)
# 역정규화 (원래 주가 스케일로 변환)
predictions_inverse = scaler.inverse_transform(np.array(predictions).reshape(-1, 1))
actual_test = scaler.inverse_transform(test_data)
# 결과 시각화
plot_predictions(actual_test, predictions_inverse, seq_length)
if __name__ == "__main__":
main()
AI Study