[PyTorch]Vanilla RNN을 활용한 코스피 예측 문제

목차

 

지난 글에선 CNN과 CIFAR-10 데이터셋을 이용한 이미지 분류 문제를 살펴보았다.

 

[PyTorch]CNN을 활용한 이미지 분류 문제(CIFAR-10)

[PyTorch]CNN을 활용한 이미지 분류 문제(CIFAR-10)

 

이번 글에서는 순환 신경망RNN(Recurrent Neural Network)을 사용해

 

코스피 데이터를 기반으로 주식 가격을 예측하는 문제를 살펴본다.

 

RNN은 시계열 데이터와 같은 순차적 데이터에서 매우 유용한데,

 

주식 가격은 과거 데이터가 현재와 미래의 예측에 중요한 영향을 미친다.

 

따라서 RNN이 이러한 문제에 적합하다.

 

순환 신경망(Recurrent Neural Network, RNN)

순환 신경망(Recurrent Neural Network, RNN)

 

선 요약

 

먼저 오늘 뜯어볼 코드와 결과는 다음과 같다:

import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
import torch.optim as optim

import matplotlib.pyplot as plt
from tqdm import trange

df = pd.read_csv("./data/kospi.csv")

df.head()

scaler = MinMaxScaler()
df[["Open", "High", "Low", "Close", "Volume"]] = scaler.fit_transform(
    df[["Open", "High", "Low", "Close", "Volume"]]
)

df.head()

df.info()

device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(f"{device} is available.")

X = df[["Open", "High", "Low", "Volume"]].values
y = df["Close"].values


def seq_data(x, y, sequence_length):

    x_seq = []
    y_seq = []
    for i in range(len(x) - sequence_length):
        x_seq.append(
            x[i : i + sequence_length].tolist()
        )
        y_seq.append(y[i + sequence_length])

    return torch.FloatTensor(x_seq).to(device), torch.FloatTensor(y_seq).to(
        device
    ).view([-1, 1])


split = 200
sequence_length = 5

x_seq, y_seq = seq_data(X, y, sequence_length)
x_train_seq = x_seq[:split]
y_train_seq = y_seq[:split]
x_test_seq = x_seq[split:]
y_test_seq = y_seq[split:]
print(x_train_seq.size(), y_train_seq.size())
print(x_test_seq.size(), y_test_seq.size())

train = TensorDataset(x_train_seq, y_train_seq)
test = TensorDataset(x_test_seq, y_test_seq)

batch_size = 20
train_loader = DataLoader(dataset=train, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(dataset=test, batch_size=batch_size, shuffle=False)

# RNN
input_size = x_seq.size(2)
num_layers = 2
hidden_size = 8


class VanillaRNN(nn.Module):

    def __init__(self, input_size, hidden_size, sequence_length, num_layers, device):
        super(VanillaRNN, self).__init__()
        self.device = device
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.rnn = nn.RNN(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Sequential(
            nn.Linear(hidden_size * sequence_length, 1), nn.Sigmoid()
        )

    def forward(self, x):
        h0 = torch.zeros(self.num_layers, x.size()[0], self.hidden_size).to(
            self.device
        )  # 초기 hidden state 설정
        out, _ = self.rnn(
            x, h0
        )
        out = out.reshape(out.shape[0], -1)
        out = self.fc(out)
        return out


model = VanillaRNN(
    input_size=input_size,
    hidden_size=hidden_size,
    sequence_length=sequence_length,
    num_layers=num_layers,
    device=device,
).to(device)

criterion = nn.MSELoss()

lr = 1e-3
num_epochs = 100
optimizer = optim.Adam(model.parameters(), lr=lr)

loss_graph = []
n = len(train_loader)
pbar = trange(num_epochs)

for epoch in pbar:
    running_loss = 0.0

    for data in train_loader:

        seq, target = data
        out = model(seq)
        loss = criterion(out, target)

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        running_loss += loss.item()

    l = running_loss / n
    loss_graph.append(l)
    pbar.set_postfix({"epoch": epoch + 1, "loss": l})

plt.figure(figsize=(16, 8))
plt.plot(loss_graph)
plt.ylabel("loss")
plt.xlabel("epoch")
plt.show()


def plotting(train_loader, test_loader, actual):
    with torch.no_grad():
        train_pred = []
        test_pred = []
        model.eval()
        for data in train_loader:
            seq, target = data
            out = model(seq)
            train_pred += out.cpu().numpy().tolist()

        for data in test_loader:
            seq, target = data
            out = model(seq)
            test_pred += out.cpu().numpy().tolist()

    total = train_pred + test_pred

    plt.figure(figsize=(16, 8))
    plt.plot(
        np.ones(100) * len(train_pred), np.linspace(0, 1, 100), "--", linewidth=0.6
    )
    plt.plot(actual, "--")
    plt.plot(total, "b", linewidth=0.6)
    plt.legend(["train boundary", "actual", "prediction"])
    plt.show()


train_loader = DataLoader(dataset=train, batch_size=batch_size, shuffle=False)
plotting(train_loader, test_loader, df["Close"][sequence_length:].values)

 이어서 핵심 워크플로우는 다음과 같다.

 

1. 데이터 준비
   
   
   • 데이터를 불러오고(read_csv), 주가 관련 열(Open, High, Low, Close, Volume)을 정규화([0, 1]로 스케일링)한다.
2. 시퀀스 데이터 생성
   
   
   • 일정 기간(sequence_length)의 데이터를 입력으로 묶어, 다음 날의 Close 값을 예측할 수 있도록 데이터셋을 구성한다.
3. 데이터 분할
   
   
   • 학습(train)과 테스트(test) 데이터로 분리한다.
4. 데이터셋 및 데이터 로더 생성
   
   
   • PyTorch의 TensorDataset과 DataLoader를 사용해 배치 단위로 데이터를 처리할 수 있도록 설정한다.
5. RNN 모델 정의
   
   
   • RNN 레이어와 Fully Connected 레이어를 포함한 모델을 정의한다.
   • RNN은 시계열 데이터를 처리하고, FC 레이어는 최종 예측 값을 생성한다.
6. 모델 학습
   
   
   • 손실 함수(MSELoss)와 옵티마이저(Adam)를 사용해 모델을 학습한다.
   • 여러 epoch 동안 데이터를 반복 학습하며 손실을 최소화한다.
7. 결과 시각화
   
   
   • 학습 과정의 손실 변화를 시각화해 학습 상태를 점검한다.
8. 모델 평가 및 예측
   
   
   • 테스트 데이터를 모델에 입력해 예측 값을 계산한다.
   • 예측 값과 실제 값을 시각화해 모델 성능을 확인한다.

 

 이제 코드를 살펴보자. 이미 익숙한 코드라도 최대한 다시 설명하려 노력해 보았다.

 

라이브러리와 데이터 불러오기

# 데이터 처리 라이브러리
import numpy as np  # 수학적 계산과 배열 연산을 위한 라이브러리
import pandas as pd  # 데이터프레임 형태로 데이터를 다루기 위한 라이브러리

# 데이터 전처리용 라이브러리
from sklearn.preprocessing import MinMaxScaler  # 데이터를 [0, 1] 범위로 정규화

# 딥러닝 관련 PyTorch 모듈
import torch  # PyTorch 프레임워크
import torch.nn as nn  # 신경망 레이어 설계를 위한 모듈
from torch.utils.data import DataLoader, TensorDataset  # 데이터셋 및 배치 처리를 위한 모듈
import torch.optim as optim  # 최적화 알고리즘을 제공하는 모듈

# 결과 시각화 및 진행 상태 확인
import matplotlib.pyplot as plt  # 시각화를 위한 라이브러리
from tqdm import trange  # 진행 상태를 보여주는 진행 바

 

numpy

 

• 수학적 계산과 다차원 배열 처리를 지원하는 라이브러리이다.
• 딥러닝 모델 학습 전 데이터 전처리와 변환 과정에서 사용된다.
• 벡터와 행렬 연산을 최적화하여 빠른 계산을 가능하게 한다.
• 예를 들어, 주식 데이터의 행렬 계산이나 시퀀스 데이터를 구성할 때 활용된다. 

 

pandas

 

• 구조화된 데이터를 처리하고 분석하는 데 사용되는 라이브러리이다.
• 데이터프레임 형태로 데이터를 다룰 수 있어 CSV 파일과 같은 표 형식 데이터를 효율적으로 처리한다.
• 주식 데이터와 같은 시계열 데이터를 다룰 때 날짜와 값을 효과적으로 관리할 수 있다.
• 예를 들어, 코스피 데이터를 읽고 열별로 정규화하거나 필요한 열만 추출하는 데 사용된다.

 

sklearn.preprocessing.MinMaxScaler

 

• 입력 데이터를 정규화(normalization)하여 [0, 1] 범위로 변환한다.
• 딥러닝 모델은 입력 데이터의 스케일에 민감하기 때문에 정규화를 통해 안정적인 학습을 돕는다.
• 주식 가격은 보통 큰 값을 가지므로, 이를 정규화하여 모델 학습 과정에서 값의 스케일 차이가 발생하지 않도록 한다.
• 예를 들어, Open, High, Low, Close, Volume 값을 정규화하여 모델에 입력한다.

 

torch와 관련 모듈

 

• PyTorch는 딥러닝 모델을 정의하고 학습하는 데 사용하는 프레임워크이다.
  
  
• GPU 가속을 활용하여 대규모 데이터를 효율적으로 처리한다.
  
  
• 주요 모듈 설명:
  
  
  • torch.nn
    신경망을 정의하기 위한 모듈이다. RNN, LSTM, GRU 등 다양한 레이어를 지원하며, 모델 구조를 설계할 때 사용된다.
  • torch.utils.data
    데이터셋을 효율적으로 관리하고 모델에 배치 단위로 데이터를 제공하는 데 사용된다.
  • torch.optim
    모델의 파라미터를 업데이트하는 최적화 알고리즘(예: SGD, Adam)을 제공한다.

 

matplotlib

 

• 학습 과정에서 손실(loss)의 변화를 시각화하거나, 예측 결과를 실제 값과 비교하는 데 사용된다.
• 주식 예측 문제에서는 학습 후 예측된 주가 흐름을 시각적으로 확인할 때 유용하다.

 

tqdm

 

• 코드 실행 상태를 시각적으로 확인할 수 있는 진행 바(progress bar)를 제공한다.
• 딥러닝 모델 학습은 시간이 오래 걸리므로, trange를 사용해 각 에포크(epoch)의 진행 상태를 확인할 수 있다.
• 학습 중 tqdm을 사용하면 실시간으로 학습 진행도를 확인할 수 있어 디버깅과 개발 생산성을 높인다.

 

데이터 정규화

df = pd.read_csv("./data/kospi.csv")
df[["Open", "High", "Low", "Close", "Volume"]] = scaler.fit_transform(
    df[["Open", "High", "Low", "Close", "Volume"]]
)

 

데이터 읽기

df = pd.read_csv("./data/kospi.csv")

 

• pd.read_csv 함수를 사용해 kospi.csv 파일을 읽어온다.
• 읽어온 데이터는 DataFrame 형식으로 저장된다.
• 이 데이터는 코스피(KOSPI) 주식 시장의 가격 정보를 포함하고 있을 가능성이 높다.
  
  
  • 열(column): Open(시가), High(고가), Low(저가), Close(종가), Volume(거래량) 등.

 

데이터 정규화

df[["Open", "High", "Low", "Close", "Volume"]] = scaler.fit_transform(
    df[["Open", "High", "Low", "Close", "Volume"]]
)

• MinMaxScaler 사용:
  
  
  • MinMaxScaler는 데이터를 [0, 1] 범위로 변환한다.
  • 정규화(normalization)는 딥러닝 모델 학습에서 중요한 단계이다.
  • 데이터가 서로 다른 범위를 가지면 모델이 학습하는 데 어려움을 겪을 수 있다.
• fit_transform:
  
  
  • 입력 데이터의 최솟값과 최댓값을 계산한 후, 해당 범위로 데이터를 변환한다.
  • 정규화 공식:

각 열의 값 x는 최솟값 min(x)과 최댓값 max(x)를 기준으로 [0, 1] 사이 값으로 변환된다.

 

적용 대상 열

 

• Open, High, Low, Close, Volume 열에 대해 정규화를 적용한다.
  
  
  • Open: 특정 날짜의 주식 시장 시작 가격.
  • High: 해당 날짜 중 가장 높은 가격.
  • Low: 해당 날짜 중 가장 낮은 가격.
  • Close: 주식 시장 마감 가격.
  • Volume: 주식 거래량.

 

왜 정규화가 필요한가?

 

1. 모델의 안정성 증가
   
   
   • 딥러닝 모델은 가중치(weight)를 학습하며 데이터를 처리한다.
   • 입력 값의 범위가 너무 크거나 작으면 학습 속도가 느려지거나 학습이 불안정해진다.
   • 정규화는 이러한 문제를 완화하여 안정적인 학습을 가능하게 한다.
2. 입력 데이터 간 균형 유지
   
   
   • 거래량(Volume)은 수백만 단위일 수 있지만, 가격(Open, High, 등)은 몇백 단위일 수 있다.
   • 이러한 스케일 차이는 특정 특징(feature)이 모델 학습에서 지나치게 중요하게 반영되도록 만들 수 있다.
   • 모든 특징을 동일한 범위로 맞추어 모델이 각 특징에 대해 균형 있게 학습하도록 만든다.
3. 수렴 속도 향상
   
   
   • 데이터를 정규화하면 모델 학습 과정에서 손실 함수의 수렴 속도가 빨라진다.
   • 스케일이 다른 입력 값을 처리하는 데 추가 계산이 필요 없기 때문이다.

 

추가적으로 고려할 점

 

현재 코드에서는 훈련 데이터와 테스트 데이터가 분리되기 전에 MinMaxScaler로 정규화를 수행하고 있다.

 

테스트 데이터의 분포를 모델이 미리 알게 되면, 실제 상황에서 예측 성능이 낮아질 수 있다.

 

왜 문제가 될 수 있는가?

 

• 데이터 누수(Data Leakage):
  
  
  • MinMaxScaler는 데이터를 정규화하기 위해 전체 데이터의 최소값과 최대값을 계산한다.
  • 테스트 데이터가 훈련 데이터와 함께 정규화되면, 테스트 데이터의 통계(최소값/최대값)가 훈련 과정에서 모델에 간접적으로 전달된다.
  • 이는 모델이 테스트 데이터에 대해 "미리 알고 있는 정보"를 바탕으로 학습하게 되어, 실제로 일반화 성능이 떨어질 가능성이 있다.

 

RNN에 사용할 시퀀스 데이터 생성

def seq_data(x, y, sequence_length):
    x_seq = []
    y_seq = []
    for i in range(len(x) - sequence_length):
        # x에서 sequence_length 길이만큼 슬라이싱하여 시퀀스를 생성
        x_seq.append(x[i : i + sequence_length].tolist())
        # y에서 시퀀스 다음 값을 정답으로 추가
        y_seq.append(y[i + sequence_length])
    # 생성된 시퀀스를 PyTorch 텐서로 변환하고 GPU로 이동
    return torch.FloatTensor(x_seq).to(device), torch.FloatTensor(y_seq).to(device).view([-1, 1])

 

주요 개념

 

1. 입력 데이터(x)와 정답 데이터(y):
   
   
   • x: RNN의 입력으로 사용할 데이터 배열이다. 예제에서는 Open, High, Low, Volume 데이터를 포함한다.
   • y: 예측할 목표 값이다. 예제에서는 Close 값을 사용한다.
2. 시퀀스 길이(sequence_length):
   
   
   • RNN은 과거 데이터를 기반으로 예측을 수행하므로, 이전 sequence_length 만큼의 데이터를 묶어서 하나의 시퀀스로 사용한다.
   • 예를 들어, sequence_length=5라면, 1일부터 5일까지의 데이터를 하나의 시퀀스로 사용해 6일째 데이터를 예측한다.

 

빈 리스트 초기화

x_seq = []
y_seq = []

• x_seq: 입력 시퀀스를 저장할 리스트이다.
• y_seq: 정답(타겟) 값을 저장할 리스트이다.

 

슬라이싱하여 시퀀스 생성

for i in range(len(x) - sequence_length):
    x_seq.append(x[i : i + sequence_length].tolist())
    y_seq.append(y[i + sequence_length])

• for i in range(len(x) - sequence_length):
  
  
  • 전체 데이터 길이에서 sequence_length를 뺀 만큼 반복한다. 이는 마지막 시퀀스가 y의 범위를 벗어나지 않도록 하기 위함이다.
• x[i : i + sequence_length]:
  
  
  • i부터 i + sequence_length까지 데이터를 슬라이싱하여 하나의 시퀀스를 생성한다.
• x_seq.append(...):
  
  
  • 슬라이싱된 시퀀스를 리스트에 추가한다.
• y_seq.append(...):
  
  
  • i + sequence_length번째의 y 값을 정답 데이터로 추가한다.

예를 들어:

• x = [1, 2, 3, 4, 5, 6, 7], y = [10, 20, 30, 40, 50, 60, 70], sequence_length=3일 때,
  
  
  • x_seq에는 [[1, 2, 3], [2, 3, 4], [3, 4, 5], [4, 5, 6]]가 저장된다.
  • y_seq에는 [40, 50, 60, 70]이 저장된다.

 

PyTorch 텐서로 변환

return torch.FloatTensor(x_seq).to(device), torch.FloatTensor(y_seq).to(device).view([-1, 1])

• 생성된 x_seq와 y_seq를 PyTorch의 FloatTensor로 변환한다.
• .to(device)를 사용해 GPU를 활용할 수 있도록 한다.
• y_seq는 .view([-1, 1])로 변환하여 2차원 텐서 형태로 만든다. 이는 학습 과정에서 출력 값과 손실 함수가 일치하도록 하기 위함이다.

 

왜 Python 리스트를 사용하는가?

 

NumPy 배열은 벡터화 연산(즉, 배열 전체에 대한 수학적 연산)에 최적화되어 있다.

 

하지만 이 코드에서는 배열 요소를 반복적으로 슬라이싱 하거나, tolist()를 호출하여 Python 리스트로 변환한다.

 

이 과정에서 다음과 같은 이유로 성능 저하가 발생한다:

 

• 불필요한 데이터 변환:
  
  
  • NumPy 배열을 tolist()로 변환하면 Python 리스트로 변환하는 추가적인 비용이 발생한다.
  • NumPy 배열은 메모리가 연속적이므로 부분 슬라이싱을 Python 리스트로 변환하면 효율성이 떨어진다.
• 반복문 내 연산:
  
  
  • NumPy 배열은 벡터화 연산이 강점인데, 이 코드에서는 반복문(for)을 통해 슬라이싱을 수행하므로 NumPy의 성능 이점이 사라진다.

 

반면 Python 리스트는 슬라이싱 및 리스트에 요소 추가가 간단하다.

 

또한 최종적으로 PyTorch 텐서로 변환하므로, Python 리스트의 속도 저하는 크게 문제 되지 않는다.

 

요약하자면 NumPy 배열의 장점(벡터화 연산)을 활용하지 않을 경우, Python 리스트가 더 적합하다.

 

데이터 분할

split = 200  # 학습 데이터와 테스트 데이터를 나누는 기준
sequence_length = 5  # 시퀀스의 길이를 설정

x_seq, y_seq = seq_data(X, y, sequence_length)  # 시퀀스 데이터를 생성
x_train_seq = x_seq[:split]  # 처음 200개의 데이터를 학습용 데이터로 사용
y_train_seq = y_seq[:split]  # 학습 데이터에 대한 정답값
x_test_seq = x_seq[split:]  # 200번째 이후의 데이터를 테스트용 데이터로 사용
y_test_seq = y_seq[split:]  # 테스트 데이터에 대한 정답값

 

split, sequence_length 변수

 

• split = 200은 데이터셋을 학습 데이터와 테스트 데이터로 나누는 기준이 된다.
  
  
  • 데이터의 처음 200개를 학습에 사용하고, 이후의 데이터를 테스트에 사용한다.
• 일반적으로 학습 데이터와 테스트 데이터는 7:3이나 8:2 비율로 나누지만, 이 코드는 고정된 개수를 사용한다.
• sequence_length = 5는 모델이 예측을 위해 참조하는 입력 데이터의 길이를 정의한다.
  
  
  • 즉, 과거 5일간의 데이터를 입력으로 사용해 다음 날의 종가(Close)를 예측한다.

 

seq_data 함수 호출

 

• seq_data(X, y, sequence_length)는 입력 데이터(X)와 출력 데이터(y)를 RNN이 처리할 수 있는 시퀀스 형태로 변환한다.
  
  
  • X는 주식 데이터를 기준으로 ["Open", "High", "Low", "Volume"] 열로 구성된다.
  • y는 예측 목표인 종가(Close) 데이터이다.
  • 함수가 반환하는 x_seq와 y_seq는 시퀀스 형태의 입력과 대응하는 출력이다.

 

학습 데이터와 테스트 데이터 분리

 

학습 데이터

x_train_seq = x_seq[:split]
y_train_seq = y_seq[:split]

• x_train_seq와 y_train_seq는 처음 200개의 데이터를 학습용으로 저장한다.
  
  
  • x_train_seq: 과거 5일간의 입력 데이터
  • y_train_seq: 학습용 입력 데이터에 대응하는 실제 출력 값

 

테스트 데이터

x_test_seq = x_seq[split:]
y_test_seq = y_seq[split:]

• x_test_seq와 y_test_seq는 200번째 이후 데이터를 테스트용으로 저장한다.
  
  
  • x_test_seq: 테스트용 입력 데이터
  • y_test_seq: 테스트용 입력 데이터에 대응하는 실제 출력 값

 

데이터셋과 데이터 로더

# 학습 및 테스트 데이터셋 생성
train = TensorDataset(x_train_seq, y_train_seq)  # 입력(x_train_seq)과 레이블(y_train_seq)을 묶은 데이터셋
test = TensorDataset(x_test_seq, y_test_seq)    # 테스트 데이터를 위한 데이터셋

# DataLoader로 데이터를 배치 단위로 로드
train_loader = DataLoader(
    dataset=train,       # 학습 데이터셋
    batch_size=20,       # 배치 크기: 한 번에 처리할 샘플 수
    shuffle=True         # 데이터를 랜덤으로 섞어서 모델에 공급
)
test_loader = DataLoader(
    dataset=test,        # 테스트 데이터셋
    batch_size=20,       # 배치 크기: 테스트 데이터는 학습과 동일
    shuffle=False        # 테스트 데이터는 순서를 유지하여 평가
)

 

TensorDataset

train = TensorDataset(x_train_seq, y_train_seq)
test = TensorDataset(x_test_seq, y_test_seq)

• TensorDataset은 PyTorch에서 제공하는 데이터 구조로, 입력 텐서(x_train_seq, x_test_seq)와 레이블 텐서(y_train_seq, y_test_seq)를 하나의 객체로 묶는다.
• 이 객체는 데이터셋처럼 작동하며, 각 샘플이 입력과 정답 쌍으로 구성된다.
• 예를 들어, 첫 번째 샘플은 (x_train_seq[0], y_train_seq[0]) 형태로 반환된다.
• 이유:
  
  
  • 데이터를 묶음으로 관리하면, 모델 학습 시 입력과 정답을 쉽게 제공할 수 있다.
  • 시퀀스 데이터와 목표 값이 서로 정렬된 상태를 유지할 수 있다.

 

DataLoader

train_loader = DataLoader(dataset=train, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(dataset=test, batch_size=batch_size, shuffle=False)

• DataLoader는 학습 데이터를 효율적으로 관리하고 모델에 공급하기 위한 도구이다.
• 주요 역할:
  
  1. 데이터를 배치(batch) 단위로 나눈다.
     
     
     • 딥러닝 모델은 데이터를 한 번에 전체 학습하지 않고, 일부 샘플(배치) 단위로 학습한다.
     • batch_size가 20이면 한 번의 학습 단계에서 20개의 샘플이 입력된다.
  2. 데이터 셔플(shuffle) 기능 제공:
     
     
     • shuffle=True는 데이터를 섞어서 제공한다. 이는 학습 데이터의 순서에 따른 편향을 줄인다.
     • 테스트 데이터는 성능 평가 시 일관성을 위해 셔플하지 않는다.
  3. 데이터를 자동으로 추출한다:
     
     
     • for data in train_loader:와 같이 반복문에서 사용할 수 있다.
     • 각 반복에서 입력 데이터와 레이블을 배치 단위로 반환한다.

 

DataLoader와 배치 학습의 이점

 

1. 메모리 효율성:
   
   
   • 전체 데이터를 한 번에 GPU 메모리에 올리지 않으므로, 메모리 부족 문제를 방지한다.
   • 배치 단위로 데이터를 처리해 메모리 사용량을 줄인다.
2. 모델 학습 안정성:
   
   
   • 배치 단위로 학습하면서 가중치 업데이트가 점진적으로 이루어진다.
   • 이는 손실 함수의 변동성을 줄이고, 모델 학습의 안정성을 높인다.
3. 속도 최적화:
   
   
   • 배치 처리는 병렬 계산이 가능하므로, 계산 속도가 빨라진다.
   • PyTorch의 DataLoader는 데이터 로드 속도를 최적화하기 위해 내부적으로 멀티스레드를 활용한다.

 

RNN 모델 정의

class VanillaRNN(nn.Module):
    def __init__(self, input_size, hidden_size, sequence_length, num_layers, device):
        super(VanillaRNN, self).__init__()
        self.rnn = nn.RNN(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Sequential(
            nn.Linear(hidden_size * sequence_length, 1), nn.Sigmoid()
        )
    
    def forward(self, x):
        h0 = torch.zeros(self.num_layers, x.size()[0], self.hidden_size).to(self.device)
        out, _ = self.rnn(x, h0)
        out = out.reshape(out.shape[0], -1)
        out = self.fc(out)
        return out

이 코드에서 정의된 VanillaRNN 클래스는 간단한 RNN 모델로, 주식 데이터를 입력받아 시계열 예측을 수행한다. 

 

모델은 RNN 레이어와 Fully Connected Layer(FC)로 구성되어 있으며, 

 

주요 구성 요소와 동작을 아래와 같이 분석할 수 있다.

 

RNN 모델 정의

class VanillaRNN(nn.Module):
    def __init__(self, input_size, hidden_size, sequence_length, num_layers, device):
        super(VanillaRNN, self).__init__()
        # RNN 레이어 정의
        self.rnn = nn.RNN(input_size, hidden_size, num_layers, batch_first=True)
        
        # Fully Connected Layer 정의
        self.fc = nn.Sequential(
            nn.Linear(hidden_size * sequence_length, 1),  # RNN 출력(feature)을 1차원으로 변환
            nn.Sigmoid()  # 결과를 [0, 1] 범위로 제한
        )

주요 매개변수:

• input_size:
  
  
  • RNN 레이어에 입력되는 각 데이터의 feature 개수를 나타낸다.
  • 이 코드에서는 주식 데이터를 나타내는 Open, High, Low, Volume으로, input_size=4가 된다.
• hidden_size:
  
  
  • RNN의 숨겨진 상태(hidden state)의 크기를 지정한다.
  • 모델이 학습 중 정보를 저장하거나 처리하는 용량을 나타낸다.
• sequence_length:
  
  
  • 하나의 입력 데이터에서 처리할 시간 단위의 길이를 나타낸다.
  • 이 코드에서는 sequence_length=5로 설정하여, 과거 5일의 데이터를 기반으로 예측한다.
• num_layers:
  
  
  • RNN 레이어의 층 수를 나타낸다.
  • 깊이를 추가하여 모델의 학습 능력을 확장한다.
• device:
  
  
  • 모델이 실행될 디바이스(CPU 또는 GPU)를 설정한다.

 

Forward 메서드

    def forward(self, x):
        # 초기 hidden state 설정
        h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(self.device)
        
        # RNN 레이어에 입력 데이터 x와 초기 hidden state를 전달
        out, _ = self.rnn(x, h0)
        
        # RNN 출력을 펼쳐서 Fully Connected Layer에 전달할 준비
        out = out.reshape(out.shape[0], -1)
        
        # Fully Connected Layer를 통해 최종 출력 계산
        out = self.fc(out)
        return out

초기 hidden state 생성

h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(self.device)

• RNN의 초기 hidden state를 0으로 초기화한다.
• num_layers는 RNN의 층 수, x.size(0)는 배치 크기, hidden_size는 숨겨진 상태 크기를 나타낸다.
• RNN은 이전 상태를 기반으로 연산을 수행하므로, 첫 번째 hidden state가 필요하다.

 

RNN 연산 수행

out, _ = self.rnn(x, h0)

• 입력 데이터 x와 초기 hidden state h0를 RNN 레이어에 전달한다.
• out은 각 시점(time step)에서 계산된 출력(feature map)이다.
• hidden state는 다음 시점에 전달되지만, 여기서는 사용하지 않는다.

 

출력 형상 변환

out = out.reshape(out.shape[0], -1)

• RNN의 출력을 펼쳐서 2차원 텐서로 변환한다.
  
  
  • RNN의 출력은 기본적으로 3차원 텐서인데, Fully Connected Layer는 2차원 텐서를 입력으로 받기 때문다.
• 배치 크기(batch_size)와 출력 feature를 하나의 차원으로 연결한다.
• 이를 통해 Fully Connected Layer에 입력으로 전달할 수 있다.

변환 과정을 좀 더 자세히 보면 다음과 같다.

 

out = out.reshape(out.shape[0], -1)는 다음을 수행한다:

• out.shape[0]: batch_size를 유지한다.
• -1: 나머지 차원을 자동으로 계산하여 1차원으로 펼친다.

구체적으로:

• 입력이 (20, 5, 8)이라면,
  
  
  • batch_size=20 (배치 크기)
  • sequence_length=5 (시간 단계 수)
  • hidden_size=8 (출력 차원 수)
• 이 코드를 실행하면 텐서의 형태가 (20, 40)으로 변환된다:
  
  
  • sequence_length * hidden_size = 5 * 8 = 40이 된다.

 

Fully Connected Layer 처리

out = self.fc(out)

• Fully Connected Layer는 입력 데이터를 압축하여 최종 예측 값을 계산한다.
• Sigmoid 활성화 함수는 출력 값을 [0, 1] 범위로 제한하여 예측 결과를 정규화한다.

 

모델 학습

criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=lr)

for epoch in trange(num_epochs):
    running_loss = 0.0  # 에포크 단위로 손실을 누적할 변수 초기화
    for data in train_loader:  # 미니배치 단위로 학습 데이터를 가져온다.
        seq, target = data  # 시퀀스 데이터와 타겟 데이터 분리
        out = model(seq)  # 모델을 통해 예측 값을 계산한다.
        loss = criterion(out, target)  # 예측 값과 실제 값의 손실을 계산한다.

        optimizer.zero_grad()  # 이전 단계에서 계산된 기울기(gradient)를 초기화한다.
        loss.backward()  # 손실 값에 따라 기울기를 계산한다.
        optimizer.step()  # 기울기를 이용해 모델의 파라미터를 업데이트한다.

        running_loss += loss.item()  # 현재 배치의 손실 값을 누적한다.
    loss_graph.append(running_loss / len(train_loader))  # 에포크별 평균 손실 값을 기록한다.

손실 함수와 최적화 알고리즘 정의

criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=lr)

• 손실 함수 정의:
  
  
  • criterion = nn.MSELoss()는 평균 제곱 오차(MSE, Mean Squared Error)를 손실 함수로 정의한다.
  • MSE는 예측 값과 실제 값의 차이를 제곱하여 평균을 계산한다.
  • MSE는 회귀 문제에서 일반적으로 사용되며, 값이 작을수록 모델의 예측이 실제 값에 가깝다는 것을 의미한다.
• 최적화 알고리즘 정의:
  
  
  • optimizer = optim.Adam(model.parameters(), lr=lr)는 Adam 옵티마이저를 사용하여 모델의 학습 가능 파라미터(model.parameters())를 업데이트한다.
  • Adam은 SGD(Stochastic Gradient Descent)의 변형으로, 학습 속도와 효율성을 높이는 데 유리하다.
  • lr은 학습률(learning rate)로, 한 번의 업데이트에서 얼마나 파라미터를 조정할지를 결정한다.

 

학습 단계

for epoch in trange(num_epochs):
    running_loss = 0.0  # 에포크 단위로 손실을 누적할 변수 초기화
    for data in train_loader:  # 미니배치 단위로 학습 데이터를 가져온다.
        seq, target = data  # 시퀀스 데이터와 타겟 데이터 분리
        out = model(seq)  # 모델을 통해 예측 값을 계산한다.
        loss = criterion(out, target)  # 예측 값과 실제 값의 손실을 계산한다.

        optimizer.zero_grad()  # 이전 단계에서 계산된 기울기(gradient)를 초기화한다.
        loss.backward()  # 손실 값에 따라 기울기를 계산한다.
        optimizer.step()  # 기울기를 이용해 모델의 파라미터를 업데이트한다.

        running_loss += loss.item()  # 현재 배치의 손실 값을 누적한다.
    loss_graph.append(running_loss / len(train_loader))  # 에포크별 평균 손실 값을 기록한다.

에포크 반복

 

• for epoch in trange(num_epochs):는 전체 학습 과정을 에포크(epoch) 단위로 반복한다.
• 에포크는 데이터셋 전체를 한 번 학습에 사용하는 주기를 의미한다.

 

배치 데이터 로드

 

• for data in train_loader:는 DataLoader를 통해 학습 데이터를 미니배치(batch) 단위로 가져온다.
• 배치 학습은 데이터를 나누어 처리함으로써 메모리 효율성을 높이고 학습 속도를 빠르게 한다.

 

예측 값 계산

 

• out = model(seq)는 입력 데이터(seq)를 모델에 통과시켜 예측 값을 계산한다.
• 이때 모델은 이전에 정의한 RNN 구조를 기반으로 입력 데이터의 시퀀스 정보를 활용해 출력 값을 생성한다.

 

손실 값 계산

 

• loss = criterion(out, target)는 예측 값(out)과 실제 값(target) 사이의 손실을 계산한다.
• 손실 값은 현재 모델의 예측이 얼마나 정확하지 않은지를 나타낸다.

 

기울기 초기화

 

• optimizer.zero_grad()는 이전 배치에서 계산된 기울기 값을 초기화한다.
• PyTorch에서는 기울기를 누적 방식으로 계산하므로 초기화하지 않으면 이전 배치의 기울기가 남아 오류를 발생시킨다.

 

역전파(Backward Propagation)

 

• loss.backward()는 손실 값에 따라 기울기를 계산한다.
• 이 과정에서 모델의 각 파라미터에 대해 손실이 어떻게 변하는지를 계산한다.

 

파라미터 업데이트

 

• optimizer.step()는 계산된 기울기를 이용해 모델의 파라미터를 업데이트한다.
• 이때 Adam 옵티마이저는 학습률, 모멘텀(momentum), 적응적 학습률 조정을 결합해 효율적으로 파라미터를 조정한다.

 

손실 값 누적

 

• running_loss += loss.item()는 현재 배치의 손실 값을 누적한다.
• loss.item()은 텐서를 숫자로 변환하여 누적이 가능하도록 한다.

 

평균 손실 값 기록

 

• loss_graph.append(running_loss / len(train_loader))는 한 에포크의 평균 손실 값을 기록한다.
• 이를 통해 학습 진행 상황을 시각화하거나 분석할 수 있다.

 

모델 평가 및 예측

def plotting(train_loader, test_loader, actual):
    with torch.no_grad():  # 평가 단계에서는 gradient 계산을 비활성화하여 메모리 사용을 줄인다.
        train_pred, test_pred = [], []  # 학습 데이터와 테스트 데이터에 대한 예측 결과를 저장할 리스트를 초기화한다.
        
        # 학습 데이터 예측
        for data in train_loader:  # 학습 데이터를 배치 단위로 가져온다.
            seq, target = data  # 배치에서 입력 시퀀스(seq)와 정답(target)을 분리한다.
            out = model(seq)  # 모델에 시퀀스를 입력하여 예측 값을 얻는다.
            train_pred += out.cpu().numpy().tolist()  # GPU 텐서를 CPU로 이동한 후, 리스트에 추가한다.
        
        # 테스트 데이터 예측
        for data in test_loader:  # 테스트 데이터를 배치 단위로 가져온다.
            seq, target = data  # 배치에서 입력 시퀀스(seq)와 정답(target)을 분리한다.
            out = model(seq)  # 모델에 시퀀스를 입력하여 예측 값을 얻는다.
            test_pred += out.cpu().numpy().tolist()  # GPU 텐서를 CPU로 이동한 후, 리스트에 추가한다.

    # 학습 데이터와 테스트 데이터를 모두 합친 예측 결과
    total = train_pred + test_pred  # 학습 예측 값과 테스트 예측 값을 하나의 리스트로 결합한다.

    # 결과 시각화
    plt.figure(figsize=(16, 8))  # 그래프 크기를 설정한다.
    plt.plot(np.ones(100) * len(train_pred), np.linspace(0, 1, 100), "--", linewidth=0.6)  
    # 학습 데이터와 테스트 데이터의 경계를 나타내는 세로선을 그린다.
    
    plt.plot(actual, "--")  # 실제 값을 점선으로 나타낸다.
    plt.plot(total, "b", linewidth=0.6)  # 예측 값을 파란색 선으로 나타낸다.
    plt.legend(["train boundary", "actual", "prediction"])  # 그래프의 범례를 설정한다.
    plt.show()  # 그래프를 화면에 출력한다.

평가 함수 정의

def plotting(train_loader, test_loader, actual):
    with torch.no_grad():

• 평가 함수:
  
  
  • train_loader와 test_loader를 입력받아 모델의 예측 값을 계산한다.
  • actual은 실제 값(ground truth)으로, 예측 결과와 비교하기 위해 사용한다.
• torch.no_grad():
  
  
  • 평가 과정에서는 모델의 학습이 이루어지지 않으므로, 불필요한 그래디언트 계산을 비활성화한다.
  • 메모리 사용을 줄이고 계산 속도를 높인다.

 

학습 데이터 예측

        train_pred, test_pred = [], []
        for data in train_loader:
            seq, target = data
            out = model(seq)
            train_pred += out.cpu().numpy().tolist()

• train_pred 초기화:
  
  
  • 학습 데이터의 예측 값을 저장할 리스트를 생성한다.
• 반복문을 통한 예측:
  
  
  • train_loader에서 배치 데이터를 하나씩 가져온다.
  • seq는 입력 데이터, target은 정답 데이터이다.
• 모델 예측 수행:
  
  
  • model(seq)를 통해 입력 데이터를 기반으로 예측 값을 계산한다.
• 결과 저장:
  
  
  • 모델의 예측 결과(out)를 CPU로 이동하고, numpy 배열로 변환하여 리스트에 추가한다.

 

테스트 데이터 예측

        for data in test_loader:
            seq, target = data
            out = model(seq)
            test_pred += out.cpu().numpy().tolist()

• test_pred 초기화:
  
  
  • 테스트 데이터의 예측 값을 저장할 리스트를 생성한다.
• 반복문을 통한 예측:
  
  
  • test_loader에서 배치 데이터를 하나씩 가져온다.
• 모델 예측 수행:
  
  
  • 학습 데이터와 동일하게 model(seq)로 예측 값을 계산한다.
• 결과 저장:
  
  
  • 예측 결과를 test_pred 리스트에 추가한다.

 

결과 합치기

    total = train_pred + test_pred

• 예측 값 결합:
  
  
  • 학습 데이터(train_pred)와 테스트 데이터(test_pred)의 예측 값을 하나의 리스트로 합친다.
  • 합쳐진 결과는 전체 데이터에 대한 예측 값이다.

 

시각화

    plt.plot(actual, "--")
    plt.plot(total, "b", linewidth=0.6)
    plt.legend(["actual", "prediction"])
    plt.show()

• 실제 값 플롯:
  
  
  • actual 데이터를 점선("--")으로 그린다. 이는 정답 데이터이다.
• 예측 값 플롯:
  
  
  • 예측 결과(total)를 파란색 실선으로 그린다.
  • linewidth=0.6은 선의 두께를 얇게 설정하여 깔끔하게 표시한다.
• 범례 추가:
  
  
  • "actual"과 "prediction"의 범례를 표시하여 그래프를 해석하기 쉽게 만든다.
• 그래프 출력:
  
  
  • plt.show()를 통해 학습 및 테스트 데이터의 예측 성능을 시각적으로 확인한다.

 

요약

 

1. 학습 데이터 예측:
   
   
   • train_loader를 반복하며 모델의 예측 값을 계산하고 저장한다.
2. 테스트 데이터 예측:
   
   
   • test_loader를 반복하며 모델의 예측 값을 계산하고 저장한다.
3. 결과 결합:
   
   
   • 학습 데이터와 테스트 데이터의 예측 값을 하나로 합친다.
4. 시각화:
   
   
   • 실제 값과 예측 값을 한 그래프에 그려 모델의 성능을 직관적으로 확인한다.

 

정리

 

이렇게 해서 얼렁뚱땅 네 번째 딥러닝 코드를 뜯어보았다.

 

기본적인 인공 신경망에서 시작해 회귀문제, CNN, RNN까지 왔는데

 

솔직히 수학적인 원리를 제대로 파악하지 못하고 쓰는데 먼저 익숙해지는 느낌이 들어 꺼림칙하긴 하다.

 

그래도 뭐, 하다 보면 더 깊이 공부할 수 있는 날도 오겠지.

 

끝!