[RNN] 시퀀스 데이터와 순환 신경망(RNN) 학습 정리

1. 시퀀스 데이터 (Sequence Data)

시퀀스 데이터(Sequence Data)는 시간적 혹은 순서적 관계가 중요한 데이터를 말합니다. 일반적인 데이터가 순서와 무관하게 독립적으로 다뤄지는 것과 달리, 시퀀스 데이터는 각 요소가 특정 순서에 따라 나열되어 있고 앞뒤 맥락이 의미를 가집니다.

대표적인 예로는 자연어 문장(단어의 순서 중요), 시계열 데이터(주가, 날씨 변화), 음성 신호(시간에 따른 파형), 영상 프레임(연속된 이미지) 등이 있습니다. 따라서 시퀀스 데이터는 단순한 값들의 모음이 아니라, 순서와 맥락이 포함된 정보 구조로 이해하는 것이 핵심입니다.

---

2. RNN (Recurrent Neural Network)

RNN(Recurrent Neural Network, 순환 신경망)은 시계열 데이터나 연속적인 데이터를 다룰 때 사용되는 인공 신경망으로, 일반적인 신경망(CNN, MLP 등)이 입력을 한 번 처리하고 끝나는 것과 달리, 과거의 정보를 기억하며 다음 계산에 반영하는 특징이 있습니다.

일반적인 신경망은 현재 입력만 보고 예측하지만, 시계열 데이터나 자연어처럼 이전 정보가 중요한 경우에는 적절하지 않기 때문에 RNN이 필요합니다. RNN은 기존 신경망과 달리 자신의 출력을 다시 입력으로 사용하여 과거 정보를 기억하는 구조를 가지며, 이를 통해 시계열 데이터의 패턴을 학습할 수 있습니다.

그러나 일반적인 RNN은 장기 의존성(Long-Term Dependency) 문제로 인해 학습이 어려울 수 있으며, 이를 해결하기 위해 장단기 메모리(Long Short-Term Memory, LSTM)나 게이트 순환 유닛(Gated Recurrent Unit, GRU)과 같은 변형 모델이 개발되었습니다.

은닉층에 있는 RNN의 처리 단위를 셀(cell)이라고 부르며, 셀의 출력을 은닉 상태(hidden state)라고 합니다. RNN은 시점(time step)에 따라서 입력을 받는데 현재 시점의 hidden state인 $h_t$ 연산을 위해 직전 시점의 hidden state인 $h_{t-1}$을 입력받습니다. 이것이 RNN이 과거의 정보를 기억할 수 있는 방법입니다.

하이퍼볼릭탄젠트(tanh) 함수는 시그모이드 함수와 달리 -1 ~ 1의 범위 값으로 값을 반환하는 함수입니다. 반환값의 범위가 시그모이드 함수보다 크므로 일반적으로 은닉층에서는 시그모이드 함수보다 더 잘 동작합니다.

---

3. 실습 1: RNN 모델로 이미지 분류하기 (MNIST)

이미지 데이터인 MNIST를 시퀀스 데이터로 해석하여(28행을 순차적인 입력으로 간주) RNN 모델로 분류하는 실습 코드입니다.

import torch
import torch.nn as nn
import torchvision.transforms as transforms
import torchvision.datasets as dset

# 1. 데이터셋 로드
train_dataset = dset.MNIST(root='.', train=True, transform=transforms.ToTensor(), download=True)
test_dataset = dset.MNIST(root='.', train=False, transform=transforms.ToTensor())

batch_size = 100
train_loader = torch.utils.data.DataLoader(dataset=train_dataset, batch_size=batch_size, shuffle=True)
test_loader = torch.utils.data.DataLoader(dataset=test_dataset, batch_size=batch_size, shuffle=False)

# 2. RNN 모델 정의
class RNNModel(nn.Module):
    def __init__(self, input_dim, hidden_dim, layer_dim, output_dim):
        super(RNNModel, self).__init__()
        self.hidden_dim = hidden_dim
        self.layer_dim = layer_dim
        # batch_first: 배치 차원을 맨 앞에 두겠다는 의미 (batch_size, seq_len, input_size)
        self.rnn = nn.RNN(input_dim, hidden_dim, layer_dim, batch_first=True)
        self.fc = nn.Linear(hidden_dim, output_dim)

    def forward(self, x):
        # h0 초기화: (num_layers, batch, hidden_size)
        h0 = torch.zeros(self.layer_dim, x.size(0), self.hidden_dim).to(x.device)
        out, hn = self.rnn(x, h0)
        # 마지막 시점(t)의 출력만 사용하여 최종 결과 예측
        out = self.fc(out[:, -1, :])
        return out

# 3. 하이퍼파라미터 및 모델 설정
input_dim = 28   # 28행을 순서대로 입력
hidden_dim = 100 
layer_dim = 1    
output_dim = 10  
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

model = RNNModel(input_dim, hidden_dim, layer_dim, output_dim).to(device)
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

# 4. 학습 및 평가
num_epochs = 20
for epoch in range(num_epochs):
    model.train()
    for images, labels in train_loader:
        images = images.to(device).view(-1, 28, 28) # (batch, seq_len, input_dim)
        labels = labels.to(device)

        optimizer.zero_grad()
        outputs = model(images)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

    model.eval()
    correct, total = 0, 0
    for images, labels in test_loader:
        images = images.to(device).view(-1, 28, 28)
        labels = labels.to(device)
        outputs = model(images)
        _, predicted = torch.max(outputs.data, 1)
        total += labels.size(0)
        correct += (predicted == labels).sum()

    accuracy = 100 * correct / total
    print(f'Epoch: {epoch}. Loss: {loss.item():.4f}. Accuracy: {accuracy:.2f}%')

---

4. 실습 2: RNN을 이용한 KOSPI 주가 예측

과거의 주가 데이터를 기반으로 다음 날의 종가를 예측하는 시계열 회귀 모델 실습입니다.

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

# 1. 환경 설정 및 데이터 로드
seed = 2025
np.random.seed(seed)
torch.manual_seed(seed)

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
df = pd.read_csv('/content/drive/MyDrive/랭체인 AI 영상객체탐지분석 플랫폼 구축/11. 멀티모달/data/kospi.csv')

# 전처리: 날짜 정렬 및 결측치 처리(ffill, bfill)
if 'Date' in df.columns:
    df['Date'] = pd.to_datetime(df['Date'], errors='coerce')
    df = df.sort_values('Date').reset_index(drop=True)
df = df.ffill().bfill()

FEATS = ['Open', 'High', 'Low', 'Close']
split_idx = int(len(df) * 0.5)
train_df, test_df = df.iloc[:split_idx].copy(), df.iloc[split_idx:].copy()

# 데이터 스케일링
scaler_x, scaler_y = MinMaxScaler(), MinMaxScaler()
X_train = scaler_x.fit_transform(train_df[FEATS].values)
X_test = scaler_x.transform(test_df[FEATS].values)
y_train = scaler_y.fit_transform(train_df[['Close']].values)
y_test = scaler_y.transform(test_df[['Close']].values)

# 시퀀스 데이터 생성 함수
def make_seq(X, y, L: int):
    Xs, ys = [], []
    for i in range(len(X) - L):
        Xs.append(X[i: i+L])
        ys.append(y[i+L])
    return torch.tensor(np.array(Xs), dtype=torch.float32), \
           torch.tensor(np.array(ys), dtype=torch.float32).view(-1, 1)

sequence_length = 5
X_train_seq, y_train_seq = make_seq(X_train, y_train, sequence_length)
X_test_seq, y_test_seq = make_seq(X_test, y_test, sequence_length)

train_loader = DataLoader(TensorDataset(X_train_seq, y_train_seq), batch_size=16, shuffle=True)
test_loader = DataLoader(TensorDataset(X_test_seq, y_test_seq), batch_size=16, shuffle=False)

# 2. RNN 회귀 모델 정의
class RNNRegressor(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers):
        super().__init__()
        self.rnn = nn.RNN(input_size=input_size, hidden_size=hidden_size,
                          num_layers=num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, 1)

    def forward(self, x):
        h0 = torch.zeros(self.rnn.num_layers, x.size(0), self.rnn.hidden_size, device=x.device)
        out, _ = self.rnn(x, h0)
        yhat = self.fc(out[:, -1, :]) # 마지막 시점의 hidden state 사용
        return yhat

model = RNNRegressor(X_train_seq.size(2), 32, 2).to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 3. 모델 학습 (Gradient Clipping 적용)
epochs = 200
for epoch in range(1, epochs+1):
    model.train()
    running = 0.0
    for xb, yb in train_loader:
        xb, yb = xb.to(device), yb.to(device)
        optimizer.zero_grad()
        pred = model(xb)
        loss = criterion(pred, yb)
        loss.backward()
        # 기울기 폭주 방지를 위한 clip_grad_norm_
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()
        running += loss.item()
    if epoch % 20 == 0 or epoch == 1:
        print(f"[{epoch:03d}/{epochs}] train MSE: {running/len(train_loader):.6f}")

# 4. 예측 및 시각화
model.eval()
pred_list = []
with torch.no_grad():
    for xb, _ in test_loader:
        pred = model(xb.to(device))
        pred_list.append(pred.cpu().numpy())

pred_unscaled = scaler_y.inverse_transform(np.vstack(pred_list))
y_actual_test = df['Close'].values[split_idx + sequence_length:]

plt.figure(figsize=(20, 10))
plt.plot(df['Close'].values, color='gray', alpha=0.3, label='full series')
plt.plot(np.arange(split_idx + sequence_length, len(df)), y_actual_test, 'b', label='actual')
plt.plot(np.arange(split_idx + sequence_length, len(df)), pred_unscaled, 'r', label='prediction')
plt.legend()
plt.title('KOSPI Close Price Prediction')
plt.show()

---

오늘의 요약:

• RNN은 은닉 상태(Hidden State)를 통해 과거 정보를 현재로 전달하는 구조를 가진다.
• 이미지 데이터도 순차적으로 입력하면 RNN으로 분류가 가능하다.
• 시계열 예측 시에는 데이터 스케일링과 시퀀스 생성 과정이 필수적이다.
• 기울기 폭주를 막기 위해 Gradient Clipping 기법을 활용할 수 있다.