RNN을 이용한 주가 예측

시계열 자료를 분석하기 위해서 딥러닝이 최근에는 활발히 적용되고 있습니다. 시계열 정보를 다루는 딥러닝 중에가서 가장 기본적인 RNN을 활용해 주가를 예측(Fitting)해보도록 하겠습니다. 

지난번에 했던 단순/다항 회귀분석이나 이번에 작성할 RNN 모형 모두 정확한 예측보다는, 적용될 수 있는 예측방법들이 이러한 것들이 있다는 것을 정리하고자 합니다. 따라서 각 모형에 대한 자세한 이론이나 설명보다는 간단한 모델들을 직접 구현하면서 정리하는데 주안점을 두고 있습니다.

먼저 필요한 Module을 Import하고 크롤링을 통해 수집했던 주가 데이터를 불러옵니다

import pandas as pd
import numpy as np
import torch
import torch.nn as nn
import os
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler
from math import sqrt
from sklearn.metrics import mean_squared_error

os.chdir(".../new_stock")
sam = pd.read_csv("samsung.csv")
sam = sam.sort_index(ascending=False)
sam=sam.reset_index()
sam=sam.drop(["index"], axis=1)

RNN을 추정하는데 데이터를 Scaling 하는 것이 효율이 좋다고 알려져 있습니다. 따라서 기본적인 Min-Max Scaler을 적용하여 0~1까지의 값을 가질 수 있게 데이터를 변환하였습니다.

scaler = MinMaxScaler()
sam["Close"]=scaler.fit_transform(sam["종가"].values.reshape(-1,1))
data=sam["Close"]

 그 다음 RNN에 데이터를 집어넣기 위해서는 Sequence와 Input size를 생각하여 데이터를 구성해야 합니다. 저는 Sequence를 10으로, Input size는 1로 두고 RNN 모형의 Input을 넣기 위한 데이터를 만들었습니다. 그 후에 임의의 길이로 train data와 test data로 구분하였습니다. 

def make_dataset(data, window_size = 10):
    feature_list =[]
    label_list =[]
    for i in range(len(data)-window_size):
        feature_list.append(np.array(data.iloc[i:i+window_size]))
        label_list.append(np.array(data.iloc[i+window_size]))
        
    return np.array(feature_list), np.array(label_list)


data_X, data_Y = make_dataset(data)
train_data, train_label = data_X[:-100], data_Y[:-100]
test_data, test_label = data_X[-100:], data_Y[-100:]

Input data의 경우 Shape을 3차원으로 변경해야 RNN 모형의 Input으로 넣을 수 있습니다. 따라서 아래와 같이 Data processing 작업을 추가하여 Input data의 경우 (Batch size, Sequence, Input size)로 구성하였습니다.

## Train
train_data = torch.FloatTensor(train_data)
train_label = torch.FloatTensor(train_label)
train_data=train_data.view(train_data.shape[0],10,-1)
train_label=train_label.view(train_data.shape[0],1)

train_rnn = torch.utils.data.TensorDataset(train_data, train_label)
train_loader =torch.utils.data.DataLoader(dataset = train_rnn, batch_size =train_data.shape[0], shuffle = False)

## Test
test_data = torch.FloatTensor(test_data)
test_label = torch.FloatTensor(test_label)
test_data=test_data.view(test_data.shape[0],10,-1)

test_rnn = torch.utils.data.TensorDataset(test_data, test_label)
test_loader =torch.utils.data.DataLoader(dataset = test_rnn, batch_size =train_data.shape[0], shuffle = False)

이제 간단한 RNN과 학습 Loop을 구성하도록 하겠습니다.

class RNN(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim, layers):
        super(RNN, self).__init__()
        self.rnn = nn.RNN(input_dim, hidden_dim, num_layers=layers, batch_first = True)
        self.fc = nn.Linear(hidden_dim, output_dim, bias = True)
        
    def forward(self,x):
        x, _status = self.rnn(x)
        x = self.fc(x[:,-1])
        return x
    
model = RNN(input_dim=1,hidden_dim=30,output_dim=1,layers=2)
loss_fn = nn.MSELoss(reduction ="mean")
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

for epoch in range(500):
    for i,(train_, label_) in enumerate(train_loader):
        # print(train_)
        # print(label_)
        optimizer.zero_grad()
        y_pred = model(train_)
        loss = loss_fn(label_, y_pred)
        loss.backward()
        optimizer.step()
    if epoch% 50 ==0:
        print(epoch ,loss.item())

이제 모델을 학습한 결과를 시각화 해보도록 하겠습니다.

plt.plot(y_pred.detach().numpy(), label ="prediction")
plt.plot(train_label, label="Real")
plt.legend()
plt.show()

y_test_pred = model(test_data)
plt.plot(y_test_pred.detach().numpy(), label ="prediction")
plt.plot(test_label, label="Real")
plt.legend()
plt.show()

시각화 결과를 보면 단순/다항 회귀모형보다 훨씬 예측값이 실제값을 잘 따라간다는 것을 확인 할 수 있습니다.

이제, Scaled된 데이터가 아니라 실제 데이터와도 Fitting을 해보겠습니다. 이번에는 train데이터와 test데이터를 하나의 Figure에 나타내도록 하겠습니다.

y_train_pred = model(train_data)
y_train_pred = scaler.inverse_transform(y_train_pred.detach().numpy())

y_test_pred = model(test_data)
y_test_pred = scaler.inverse_transform(y_test_pred.detach().numpy())

trainPredictPlot = np.empty_like(sam.iloc[10:]["Close"])
trainPredictPlot[:]=np.nan
trainPredictPlot[0:len(y_train_pred)] = y_train_pred[:,0]

testPredictPlot = np.empty_like(sam.iloc[10:]["Close"])
testPredictPlot[:] = np.nan
testPredictPlot[len(y_train_pred):len(sam.iloc[10:]["Close"])+1] = y_test_pred[:,0]


plt.figure(figsize=(15,8))
plt.plot(scaler.inverse_transform(sam.iloc[10:]["Close"].values.reshape(-1,1)))
plt.plot(trainPredictPlot)
plt.plot(testPredictPlot)
plt.show()

확실히 단순/다항 회귀모형보다는 변동성을 더 잘 반영한다는 것을 알 수 있습니다. 다만 Test를 예측할 때 실제 데이터를 가지고 예측을 했기 때문에 예측력이 뛰어난 모델이라고 단언하기 어렵습니다. 하지만  RNN을 활용하면 추세를 잘 표현할 수 있다는 것을 알 수 있습니다.

Reference

https://eda-ai-lab.tistory.com/429

https://www.kaggle.com/taronzakaryan/stock-prediction-lstm-using-pytorch

https://teddylee777.github.io/tensorflow/LSTM%EC%9C%BC%EB%A1%9C-%EC%98%88%EC%B8%A1%ED%95%B4%EB%B3%B4%EB%8A%94-%EC%82%BC%EC%84%B1%EC%A0%84%EC%9E%90-%EC%A3%BC%EA%B0%80