LSTM을 활용한 주가 예측

RNN의 경우 예측해야 하는 지점과 이용하는 정보의 시작지점의 차이가 크면, 역전파시 그래디언트의 소실이 커져 효과적으로 학습이 되지 않을 수 있습니다.

이러한 한계를 극복하기 위해서 나온 방법이 LSTM으로 적절하게 과거정보를 잊고 현재정보를 기억하여 그래디언트의 소실을 막으면서 장기간의 데이터를 학습 할 수 있습니다.

따라서, 이번 포스트에서는 LSTM을 활용하여 주가를 한번 예측해보록 하겠습니다. 이번에는 SK이노베이션의 주가정보를 활용하여 예측을 해보겠습니다.

먼저 이전 포스트에서 다뤘던 내용과 같이, 필요한 모듈 및 데이터를 입력하고, 지정한 Seqeunce로 학습 및 예측에 활용할 데이터를 만듭니다.

import pandas as pd
import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
import os
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler
from sklearn import metrics
import datetime
import requests
from bs4 import BeautifulSoup
os.chdir(".../new_stock")
from py_stock_data import read_data_from_naver

## data input
str_datefrom = datetime.datetime.strftime(datetime.datetime(year=2010, month=1, day=1), '%Y.%m.%d')
interest_list = ["096770"] 

stock_data ={}

for i, code in enumerate(interest_list):
    print(i,":", code)
    new = read_data_from_naver(code)
    stock_data[code] = new.read_data(str_datefrom)
   

data = stock_data["096770"]

## data sorting and reset index
data=data.sort_values(by=["날짜"], ascending = True).reset_index()
data=data.drop(["index"], axis=1)

## scaling
data_ski = data[["종가"]]
data_ski.rename(columns={"종가": "Close"}, inplace = True)

scaler = MinMaxScaler()
data_ski["Close"]= scaler.fit_transform(data_ski["Close"].values.reshape(-1,1))

## sequence data
def make_dataset(data, window_size = 20):
    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_ski)

data_X.shape
data_Y.shape

train_data, train_label = data_X[:-300], data_Y[:-300]
test_data, test_label = data_X[-300:], data_Y[-300:]


## tensor set
X_train = torch.from_numpy(train_data).float()
y_train = torch.from_numpy(train_label).float()

X_test = torch.from_numpy(test_data).float()
y_test = torch.from_numpy(test_label).float()

이제 학습하기 위한 LSTM 클래스를 만들어보겠습니다. 여기서는 매 Batch마다 Hidden state와 Cell State를 초기화해주는 Stateless를 적용하겠습니다. Batch마다 Hidden State와 Cell State를 초기화해줘도 되고 안해줘도 됩니다. 다만 매번 Batch시에 전에 생성됐던 State를 활용하여 학습하는 것이 이번 미션에서는 적절치 않다고 개인적으로 판단하여 Stateless로 파라미터를 학습하겠습니다.

class LSTM_(nn.Module):
    
    def __init__(self, input_dim, hidden_dim, num_layers, output_dim):
        super(LSTM_, self).__init__()
        
        self.hidden_dim = hidden_dim
        self.num_layers = num_layers
        self.input_dim = input_dim
        
        
        self.lstm =nn.LSTM(input_dim, hidden_dim, num_layers, batch_first =True)
        
        self.fc = nn.Linear(hidden_dim, output_dim)
        
    
    def forward(self, x):
        h0, c0 = self.init_hidden(x)
        
        out,(hn, cn)=self.lstm(x, (h0,c0))
        
        
        out = self.fc(out[:,-1,:])
        
        return out
    
    def init_hidden(self,x):
        self.h0 =torch.zeros(self.num_layers, x.size(0), self.hidden_dim)
        self.c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_dim)
        
        return h0, c0

이제 Train Loop을 구성해보겠습니다.

model = LSTM_(input_dim=1,hidden_dim=30,output_dim=1,num_layers=2)
loss_fn = torch.nn.MSELoss(reduction="sum")
optimiser = torch.optim.Adam(model.parameters(), lr=0.01)

hist = np.zeros(200)    
    
for t in range(200):    
    
    # Forward pass
    
    y_train_pred = model(X_train)

    loss = loss_fn(y_train_pred, y_train)
    if t % 10 == 0 and t !=0:
        print("Epoch ", t, "MSE: ", loss.item())
    hist[t] = loss.item()

    # Zero out gradient, else they will accumulate between epochs
    optimiser.zero_grad()

    # Backward pass
    loss.backward()

    # Update parameters
    optimiser.step()

이제 Train 데이터와 Test 데이터를 활용하여 그래프를 Fitting 해보겠습니다.

## Train Fitting
plt.plot(y_train_pred.detach().numpy(), label="Preds")
plt.plot(y_train.detach().numpy(), label="Real")
plt.legend()
plt.show()

## Test Fitting
y_test_pred = model(X_test)
plt.plot(y_test_pred.detach().numpy(), label="Preds")
plt.plot(y_test.detach().numpy(), label="Real")
plt.legend()
plt.show()

## Test Fitting-inverser scaling 
y_test_pred = scaler.inverse_transform(y_test_pred.detach().numpy())
y_test = scaler.inverse_transform(y_test.detach().numpy())
plt.plot(y_test_pred,label="Preds")
plt.plot(y_test, label = "Real")
plt.legend()
plt.show()

Train Fitting

Test Fitting

Test Fitting -Inverse Scaling

이제 추가적으로, Fitting이 아닌 예측을 해보도록 하겠습니다. 구성한 LSTM 모델은 20일치의 데이터를 활용하여 1일치의 데이터를 예측하게 됩니다.

따라서 처음에는 20일치 데이터를 활용하여 1일치를 예측하고 그 다음에는 예측한 값을 우리 데이터인 것처럼 포함하여 예측을 수행하도록 하겠습니다.

test_seq=X_test[:1] ## X_test에 있는 데이터중 첫번째것을 가지고옮
preds =[]

for _ in range(len(X_test)):
    
    # model.init_hidden(test_seq)
    y_test_pred = model(test_seq)
    
    pred = y_test_pred.item()   
    preds.append(pred)   
    new_seq = test_seq.numpy()    
    new_seq = np.append(new_seq, pred)    
    new_seq = new_seq[1:] ## index가 0인 것은 제거하여 예측값을 포함하여 20일치 데이터 구성
    test_seq = torch.from_numpy(new_seq).view(1,20,1).float()   
    
plt.plot(preds, label="Preds")
plt.plot(y_test.detach().numpy(), label="Data")
plt.legend()
plt.show()    

Prediction vs Real

그래프를 보면, 제대로 예측하고 있다고 보기는 어려울 것 같습니다. 역시 주가를 예측하는 것은 단순한 모델로는 쉽지 않은 것 같습니다.

참조:

https://www.curiousily.com/posts/time-series-forecasting-with-lstm-for-daily-coronavirus-cases/

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

https://www.kaggle.com/purplejester/a-simple-lstm-based-time-series-classifier