10주차 TIL - 과적합

비트코인과 같은 금융 시계열 데이터는 변동성이 크고 노이즈가 많기 때문에, 모델의 일반화 성능을 높이기 위해 **임의 데이터 증강(Randomized Data Augmentation)**을 활용하는 건 매우 유효합니다.

import pandas as pd
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import random
from sklearn.preprocessing import MinMaxScaler
import matplotlib.pyplot as plt

# 1. 데이터 불러오기
df = pd.read_csv("btc_5min_last_60days.csv")
df = df[['close']]
df.dropna(inplace=True)

# 정규화
scaler = MinMaxScaler()
scaled_data = scaler.fit_transform(df)

# 시계열 생성 함수
def create_sequences(data, seq_len=60):
    xs, ys = [], []
    for i in range(len(data) - seq_len):
        x = data[i:i+seq_len]
        y = data[i+seq_len]
        xs.append(x)
        ys.append(y)
    return np.array(xs), np.array(ys)

X, y = create_sequences(scaled_data)

# 2. 데이터 증강 함수
def augment_sequence(seq):
    if random.random() < 0.3:
        seq = seq + np.random.normal(0, 0.01, seq.shape)
    if random.random() < 0.3:
        factor = np.random.uniform(0.9, 1.1)
        seq = seq * factor
    if random.random() < 0.3:
        indices = np.arange(len(seq))
        np.random.shuffle(indices)
        seq = seq[indices]
    return seq

# 3. Dataset 클래스 정의
class BTC_Dataset(Dataset):
    def __init__(self, X, y, augment=False):
        self.X = X
        self.y = y
        self.augment = augment

    def __len__(self):
        return len(self.X)

    def __getitem__(self, idx):
        x = self.X[idx]
        if self.augment:
            x = augment_sequence(x)
        return torch.FloatTensor(x), torch.FloatTensor(self.y[idx])

# 4. 모델 정의
class LSTMModel(nn.Module):
    def __init__(self, input_size=1, hidden_size=64, num_layers=2):
        super().__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, 1)

    def forward(self, x):
        out, _ = self.lstm(x)
        return self.fc(out[:, -1, :])

# 5. 학습 설정
model = LSTMModel()
loss_fn = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

dataset = BTC_Dataset(X, y, augment=True)
loader = DataLoader(dataset, batch_size=64, shuffle=True)

# 6. 학습 루프
for epoch in range(20):
    for batch_x, batch_y in loader:
        output = model(batch_x)
        loss = loss_fn(output, batch_y)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    print(f"Epoch {epoch+1}, Loss: {loss.item():.6f}")

# 7. 예측 및 시각화
model.eval()
test_input = torch.FloatTensor(X[-1:])
with torch.no_grad():
    pred = model(test_input).numpy()
    pred = scaler.inverse_transform(pred)

real = scaler.inverse_transform(y[-1:])
print(f"실제 종가: {real[0][0]:.2f}, 예측 종가: {pred[0][0]:.2f}")

실제 종가: 82987.23, 예측 종가: 83110.54