파이토치(Pytorch): LSTM 기반 GasRateCO2 시계열 예측 모델 구현

• 수업 내용 리마인드 및 아카이빙 목적의 업로드

---

---

 

이번 글에서는 GasRateCO2 데이터셋을 사용해 시계열 예측을 진행하는 과정을 소개할게요. 우리는 파이토치 기반의 LSTM(Long Short-Term Memory) 모델을 사용해서 예측 작업을 수행할 거예요. 데이터를 불러오고 전처리한 뒤, 모델을 학습시키고 평가하는 전 과정을 함께 진행해 보겠습니다.

 

1. 필요 모듈 임포트

우선, 필요한 모듈을 임포트해볼게요. torch, darts, torchinfo 등의 모듈을 사용하고, GPU가 사용 가능한지 확인해볼게요.

!pip install darts
!pip install torchinfo

import torch
import torch.nn as nn
import torch.optim as optim
from torchinfo import summary
from torch.utils.data import Dataset, DataLoader
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from darts.datasets import GasRateCO2Dataset
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error, r2_score
import warnings
from tqdm import tqdm

# 모든 경고 무시
warnings.simplefilter("ignore")

# 시드 설정 (재현 가능성 보장)
torch.manual_seed(10)
torch.cuda.manual_seed(10)

# CUDA가 사용 가능한 경우 GPU를 사용하고, 그렇지 않으면 CPU를 사용합니다.
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"사용 중인 장치: {device}")

 

2. 데이터 불러오기 및 분할

이번에는 GasRateCO2 데이터를 불러와서 훈련, 검증, 테스트 세트로 나눌 거예요. 데이터를 스케일링하고, 타겟 피처인 'GasRate(ft3/min)' 값을 예측 대상으로 설정할 거예요.

# 데이터 로드
data = GasRateCO2Dataset().load().pd_dataframe()

# 예측할 타겟 피처 설정
target_feature = "GasRate(ft3/min)"
seq_length = 12

# 데이터셋 분할 (Train: 70%, Val: 15%, Test: 15%)
train_size = int(0.7 * len(data))
val_size = int(0.15 * len(data))
test_size = len(data) - train_size - val_size

train_data = data.iloc[:train_size]
val_data = data.iloc[train_size:train_size + val_size]
test_data = data.iloc[train_size + val_size:]

print(f"train : {len(train_data)}, validation : {len(val_data)}, test : {len(test_data)}")

# 스케일링 (Train 데이터에만 fit)
scaler = MinMaxScaler(feature_range=(0, 1))
train_data_scaled = train_data.copy()
train_data_scaled[:] = scaler.fit_transform(train_data)

val_data_scaled = val_data.copy()
val_data_scaled[:] = scaler.transform(val_data)

test_data_scaled = test_data.copy()
test_data_scaled[:] = scaler.transform(test_data)

# 타겟 피처 시각화
plt.figure(figsize=(10, 5))
plt.plot(train_data[target_feature], label='Train')
plt.plot(val_data[target_feature], label='Validation')
plt.plot(test_data[target_feature], label='Test')
plt.title(f'{target_feature} Over Time')
plt.xlabel('Time')
plt.ylabel(f'{target_feature} Value')
plt.legend()
plt.show()

 

3. 데이터셋 생성 및 Dataloader 설정

torch.utils.data.Dataset을 상속받아 커스텀 데이터셋 클래스를 정의하고, DataLoader로 배치 단위로 데이터를 나누어 모델에 전달합니다.

# 커스텀 데이터셋 정의
class TimeSeriesDataset(Dataset):
    def __init__(self, data, target_feature, seq_length):
        self.data = data
        self.target_feature = target_feature
        self.seq_length = seq_length

    def __len__(self):
        return len(self.data) - self.seq_length

    def __getitem__(self, idx):
        x = self.data.iloc[idx:idx + self.seq_length].values
        y = self.data[self.target_feature].iloc[idx + self.seq_length]
        return torch.tensor(x, dtype=torch.float32), torch.tensor(y, dtype=torch.float32).unsqueeze(-1)

# 데이터셋 생성
train_dataset = TimeSeriesDataset(train_data_scaled, target_feature, seq_length)
val_dataset = TimeSeriesDataset(val_data_scaled, target_feature, seq_length)
test_dataset = TimeSeriesDataset(test_data_scaled, target_feature, seq_length)

# 데이터로더 생성
batch_size = 32
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=False)
val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

# 데이터 확인
first_batch_x, first_batch_y = next(iter(train_loader))
print(f"batch_x shape : {first_batch_x.shape}, batch_y shape : {first_batch_y.shape}")

 

4. LSTM 모델 정의

이제 LSTM 모델을 정의해보겠습니다. LSTM은 시계열 데이터 예측에 효과적인 모델로, 은닉 상태를 기억하며 여러 타임스텝에 걸친 의존성을 학습합니다.

class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, output_size, num_layers, dropout):
        super(LSTMModel, self).__init__()
        self.lstm = nn.LSTM(input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, dropout=dropout, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)

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

# 모델 초기화
input_size = 2
hidden_size = 64
output_size = 1
num_layers = 2
dropout = 0.2

model = LSTMModel(input_size, hidden_size, output_size, num_layers, dropout).to(device)

# 모델 요약 정보
model_info = summary(model, input_size=first_batch_x.shape, col_names=["input_size", "output_size", "num_params"])
model_info

 

5. 하이퍼파라미터 설정 및 학습

이제 손실 함수와 옵티마이저를 설정하고 모델 학습을 시작할 거예요. 학습 과정에서는 검증 손실이 가장 낮을 때의 모델을 저장하도록 할게요.

# 손실 함수 및 옵티마이저 정의
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 학습 및 평가 함수
def train(model, dataloader, criterion, optimizer):
    model.train()
    total_loss = 0
    for inputs, targets in dataloader:
        inputs, targets = inputs.to(device), targets.to(device)
        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, targets)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    return total_loss / len(dataloader)

def evaluate(model, dataloader, criterion):
    model.eval()
    total_loss = 0
    with torch.no_grad():
        for inputs, targets in dataloader:
            inputs, targets = inputs.to(device), targets.to(device)
            outputs = model(inputs)
            loss = criterion(outputs, targets)
            total_loss += loss.item()
    return total_loss / len(dataloader)

# 학습 루프
Epoch = 500
best_valid_loss = float('inf')
best_epoch = 0

with tqdm(total=Epoch, desc="Training Loop", unit="epoch") as pbar:
    for epoch in range(1, Epoch + 1):
        train_loss = train(model, train_loader, criterion, optimizer)
        valid_loss = evaluate(model, val_loader, criterion)

        pbar.set_postfix({'Train Loss': f'{train_loss:.4f}', 'Validation Loss': f'{valid_loss:.4f}'})
        pbar.update(1)

        # 검증 손실이 가장 낮을 때 모델 저장
        if valid_loss < best_valid_loss:
            best_valid_loss = valid_loss
            best_epoch = epoch
            torch.save({'epoch': best_epoch, 'model_state_dict': model.state_dict(), 'optimizer_state_dict': optimizer.state_dict(), 'val_loss': best_valid_loss}, "best_model.pth")

print(f"훈련 완료! 가장 낮은 검증 손실은 {best_epoch}번째 에포크에서 발생했습니다.")

 

6. 성능 지표 및 결과 시각화

모델 성능을 테스트하고, 예측된 값과 실제값을 비교해봅시다. RMSE와 R² 점수도 계산해서 모델 성능을 평가할 거예요.

# 최종 모델 평가
checkpoint = torch.load('best_model.pth')
model.load_state_dict(checkpoint['model_state_dict'])

model.eval()
test_loss = 0.0
predictions = []
actuals = []

with torch.no_grad():
    for inputs, targets in test_loader:
        inputs, targets = inputs.to(device), targets.to(device)
        outputs = model(inputs)
        test_loss += criterion(outputs, targets).item()
        predictions.extend(outputs.cpu().numpy())
        actuals.extend(targets.cpu().numpy())

test_loss /= len(test_loader)
rmse = np.sqrt(mean_squared_error(actuals, predictions))
r2 = r2_score(actuals, predictions)

print(f'Test Loss: {test_loss:.4f}, RMSE: {rmse:.4f}, R² Score: {r2:.4f}')

# 결과 시각화
plt.figure(figsize=(10, 5))
plt.plot(actuals, label='Actual')
plt.plot(predictions, label='Predicted')
plt.title('Actual vs Predicted GasRate')
plt.legend()
plt.show()