LSTM(+GRU)을 이용한 삼성전자(+NAVER) 주가 예측하기

220922 작성

<본 블로그는 coding-yoon 님의 블로그를 참고해서 공부하며 작성하였습니다 :-) >

https://coding-yoon.tistory.com/131

[Pytorch] LSTM을 이용한 삼성전자 주가 예측하기

 

 

 

 

 

 

 

1️⃣ library load

• pandas_datareader
  • Yahoo Finance에서 증시 자료를 받아올 수 있
  • 웹 상의 데이터를 DataFrame 객체로 만드는 기능을 제공

import numpy as np
import pandas as pd
import pandas_datareader.data as pdr
import matplotlib.pyplot as plt

import datetime

import torch
import torch.nn as nn
from torch.autograd import Variable 

import torch.optim as optim
from torch.utils.data import Dataset, DataLoader

 

 

 

 

 

 

 

2️⃣ 삼성 전자 주식 불러오기

• 삼성전자의 종목코드는 005930

start = (2000, 1, 1)  # 2020년 01년 01월 
start = datetime.datetime(*start)  
end = datetime.date.today()  # 현재 

# yahoo 에서 삼성 전자 불러오기 
df = pdr.DataReader('005930.KS', 'yahoo', start, end)
df.Close.plot(grid=True)

2022.09.22 기준으로 요즘 주식 장 파란불~이다 ㅠㅠㅠㅠ내림추세ㄴ

 

 

 

3️⃣ 모델 성능 학습하기위해 데이터 분류하기

• open 시가
• high 고가
• low 저가
• close 종가
• volume 거래량 (필요 없어서 DROP)
• Adj Close 주식의 분할, 배당, 배분 등을 고려해 조정한 종가 (예측하고자 하는 TARGET)

X = df.drop(columns='Volume')
y = df.iloc[:, 5:6]

print(X)
print(y)

 

 

4️⃣ 학습이 잘되기 위해 데이터 정규화 및 split

• StandardScaler : 각 특징의 평균을 0, 분산을 1이 되도록 변경
• MinMaxScaler : 최대/최소값이 각각 1, 0이 되도록 변경

from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.model_selection import train_test_split

mm = MinMaxScaler()
ss = StandardScaler()

X_ss = ss.fit_transform(X)
y_mm = mm.fit_transform(y) 

X_train, X_test, y_train, y_test = train_test_split(X_ss, y_mm, test_size=0.2, shuffle=True, random_state=34)

print("Training Shape", X_train.shape, y_train.shape)
print("Testing Shape", X_test.shape, y_test.shape)

 

 

• 학습할 수 있는 형태로 변환하기 위해 Torch로 변환
  • Torch의 Variable
    • data : a의 tensor 형태의 데이터가 담김
    • grad : data가 거쳐온 layer에 대한 미분값이 축적
    • grad_fn : 미분값을 계산한 함수에 대한 정보

X_train_tensors = Variable(torch.Tensor(X_train))
X_test_tensors = Variable(torch.Tensor(X_test))

y_train_tensors = Variable(torch.Tensor(y_train))
y_test_tensors = Variable(torch.Tensor(y_test))

X_train_tensors_final = torch.reshape(X_train_tensors,   (X_train_tensors.shape[0], 1, X_train_tensors.shape[1]))
X_test_tensors_final = torch.reshape(X_test_tensors,  (X_test_tensors.shape[0], 1, X_test_tensors.shape[1])) 

print("Training Shape", X_train_tensors_final.shape, y_train_tensors.shape)
print("Testing Shape", X_test_tensors_final.shape, y_test_tensors.shape)

 

5️⃣ LSTM 네트워크

• RNN (default)
  • RNN의 입력 : [sequence, batch_size, input_size] 
  • nn.RNN : 기본 인자 값으로 input_size, hidden_size (nn.RNN을 걸치면 나오는 Output), num_layer (층)
    • output
      • output[-1] : [sequence, batch_size, hidden_size]  에서 Sequence의 가장 끝
      • hidden[-1] : [num_layer, batch_size, hidden_size]  에서 num_layer의 가장 끝

 

https://brunch.co.kr/@linecard/324

 

• GPU 없어서 CPU

device = torch.device('cpu')

class LSTM1(nn.Module):
  def __init__(self, num_classes, input_size, hidden_size, num_layers, seq_length):
    super(LSTM1, self).__init__()
    self.num_classes = num_classes #number of classes
    self.num_layers = num_layers #number of layers
    self.input_size = input_size #input size
    self.hidden_size = hidden_size #hidden state
    self.seq_length = seq_length #sequence length
 
    self.lstm = nn.LSTM(input_size=input_size, hidden_size=hidden_size,
                      num_layers=num_layers, batch_first=True) #lstm
    self.fc_1 =  nn.Linear(hidden_size, 128) #fully connected 1
    self.fc = nn.Linear(128, num_classes) #fully connected last layer

    self.relu = nn.ReLU() 

  def forward(self,x):
    h_0 = Variable(torch.zeros(self.num_layers, x.size(0), self.hidden_size)).to(device) #hidden state
    c_0 = Variable(torch.zeros(self.num_layers, x.size(0), self.hidden_size)).to(device) #internal state   
    # Propagate input through LSTM

    output, (hn, cn) = self.lstm(x, (h_0, c_0)) #lstm with input, hidden, and internal state
   
    hn = hn.view(-1, self.hidden_size) #reshaping the data for Dense layer next
    out = self.relu(hn)
    out = self.fc_1(out) #first Dense
    out = self.relu(out) #relu
    out = self.fc(out) #Final Output
   
    return out

 

• 네트워크 파라미터 구성

num_epochs = 30000 #1000 epochs
learning_rate = 0.00001 #0.001 lr

input_size = 5 #number of features
hidden_size = 2 #number of features in hidden state
num_layers = 1 #number of stacked lstm layers

num_classes = 1 #number of output classes

lstm1 = LSTM1(num_classes, input_size, hidden_size, num_layers, X_train_tensors_final.shape[1]).to(device)

loss_function = torch.nn.MSELoss()    # mean-squared error for regression
optimizer = torch.optim.Adam(lstm1.parameters(), lr=learning_rate)  # adam optimizer

 

6️⃣ 학습

for epoch in range(num_epochs):
  outputs = lstm1.forward(X_train_tensors_final.to(device)) #       forward pass
  optimizer.zero_grad()     # caluclate the gradient, manually setting to 0
 
  # obtain the loss function
  loss = loss_function(outputs, y_train_tensors.to(device))

  loss.backward() #calculates the loss of the loss function
 
  optimizer.step() #improve from loss, i.e backprop

  if epoch % 100 == 0:
    print("Epoch: %d, loss: %1.5f" % (epoch, loss.item()))

ㅋㅋ.. 0.00001

 

num_epochs = 10000 #1000 epochs  으로 하면

 

 

 

7️⃣ 예측

• 위 처럼 변환~

df_X_ss = ss.transform(df.drop(columns='Volume'))
df_y_mm = mm.transform(df.iloc[:, 5:6])

df_X_ss = Variable(torch.Tensor(df_X_ss)) #converting to Tensors
df_y_mm = Variable(torch.Tensor(df_y_mm))

#reshaping the dataset
df_X_ss = torch.reshape(df_X_ss, (df_X_ss.shape[0], 1, df_X_ss.shape[1]))

• 빨간 선 이후가 예측한 것...

train_predict = lstm1(df_X_ss.to(device))#forward pass
data_predict = train_predict.data.detach().cpu().numpy() #numpy conversion
dataY_plot = df_y_mm.data.numpy()

data_predict = mm.inverse_transform(data_predict) #reverse transformation
dataY_plot = mm.inverse_transform(dataY_plot)
plt.figure(figsize=(10,6)) #plotting
plt.axvline(x=4563, c='r', linestyle='--') #size of the training set

plt.plot(dataY_plot, label='Actuall Data') #actual plot
plt.plot(data_predict, label='Predicted Data') #predicted plot
plt.title('Time-Series Prediction')
plt.legend()
plt.show()

 

 

 

 

+ GRU Model

class GRU(nn.Module) :
    def __init__(self, num_classes, input_size, hidden_size, num_layers, seq_length) :
        super(GRU, self).__init__()
        self.num_classes = num_classes
        self.num_layers = num_layers
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.seq_length = seq_length
        
        self.gru = nn.GRU(input_size=input_size,hidden_size=hidden_size,
                         num_layers=num_layers,batch_first=True)
        self.fc_1 = nn.Linear(hidden_size, 128)
        self.fc = nn.Linear(128, num_classes)
        self.relu = nn.ReLU()
        
    def forward(self, x) :
        h_0 = Variable(torch.zeros(self.num_layers, x.size(0), self.hidden_size))
        output, (hn) = self.gru(x, (h_0))
        hn = hn.view(-1, self.hidden_size)
        out = self.relu(hn)
        out = self.fc_1(out)
        out = self.relu(out)
        out = self.fc(out)
        return out

num_epochs = 1000
learning_rate = 0.0001

input_size=5
hidden_size=2
num_layers=1

num_classes=1
model=GRU(num_classes,input_size,hidden_size,num_layers,X_train_tensors_final.shape[1]).to(device)

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

for epoch in range(num_epochs) :
    outputs = model.forward(X_train_tensors_final)
    optimizer.zero_grad()
    loss = criterion(outputs, y_train_tensors)
    loss.backward()
    
    optimizer.step()
    if epoch % 100 == 0 :
        print(f'Epoch : {epoch}, loss : {loss.item():1.5f}')

df_X_ss = ss.transform(df.drop(columns='Volume'))
df_y_mm = mm.transform(df.iloc[:, 5:6])

df_X_ss = Variable(torch.Tensor(df_X_ss)) #converting to Tensors
df_y_mm = Variable(torch.Tensor(df_y_mm))

#reshaping the dataset
df_X_ss = torch.reshape(df_X_ss, (df_X_ss.shape[0], 1, df_X_ss.shape[1]))

train_predict = model(df_X_ss.to(device)) #forward pass
data_predict = train_predict.data.detach().cpu().numpy() #numpy conversion
dataY_plot = df_y_mm.data.numpy()

# Undo the scaling of X according to feature_range
data_predict = mm.inverse_transform(data_predict) #reverse transformation
dataY_plot = mm.inverse_transform(dataY_plot)
plt.figure(figsize=(10,6)) #plotting
plt.axvline(x=4563, c='r', linestyle='--') #size of the training set

plt.plot(dataY_plot, label='Actuall Data') #actual plot
plt.plot(data_predict, label='Predicted Data') #predicted plot
plt.title('Time-Series Prediction')
plt.legend()
plt.show()

 

 

흠............

epoch가 클수록 loss 값이 0에 가까워져서

actual 과 predict 그래프가 매우 유사하다

 

왜 클수록 완전 유사하게 잘 예측되는거지.........???

일단 내가 epoch 작게 해서 이상한 그래프 예측값이 나왔지만.......

이렇게 완전 유사...........

 

왜그럴까요?

궁긓ㅁ해요

답좀...

 

 

 

 

---

• NAVER 주가 ^^
  • 원래 이렇게 예측을 잘한다고요?
  • layer도 작은데..;;?

start = (2000, 1, 1)  # 2020년 01년 01월 
start = datetime.datetime(*start)  
end = datetime.date.today()  # 현재 

# yahoo 에서 NAVER 불러오기 
df = pdr.DataReader('035420.KS', 'yahoo', start, end)
df.Close.plot(grid=True)

왼쪽은 LSTM, 오른쪽은 GRU