CNN을 활용한 주가 방향 예측

이번 포스팅에서는 시계열 데이터에 Convolutional neural network을 적용하여 주가 방향을 예측해보도록 하겠습니다. Convolutional neural network은 주로 이미지 분석에 많이 활용되지만, 간간히 주가를 예측하기 위해도 활용되는 것 같습니다. 

Research paper들을 참조하여 모델을 구성한 것은 아니고, CNN을 활용해보는 정도에 의의를 두고자 합니다. 향후 시간을 내서 paper들을 읽고 체계적으로 구성해보는 것도 좋을 것 같습니다.

이번에는 삼성전자 대신, SKT주가를 활용하고자 합니다. 좀 더 긴 기간의 가격 정보를 활용하고자 하는데, 삼성전자의 경우 액면분할이 이루어진지 꽤 최근이라고 볼 수 있기 때문에 긴 기간의 주가를 활용하기는 부적절하다고 판단했습니다.

먼저 필요한 Module 및 데이터를 불러오겠습니다. 데이터는 앞선 포스팅(https://direction-f.tistory.com/6)에서 언급한 크롤링을 이용해서 수집하였고, 저는 별도로 클래스를 구성해 놓아 import 하여 활용하였습니다.

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
from py_stock_data import read_data_from_naver ## 별도로 구성한 Module


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

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["017670"]

dictionary로 데이터를 받아오는 이유는 향후에 많은 회사들의 주가 데이터를 한번에 가져올 수 있게 구성한 것입니다. 추가적으로, data를 최근 날짜의 주가가 맨 뒤로 가도록 순서를 바꿔주고 "종가"데이터만 활용할 것이기 때문에 "종가" 데이터만 별도 DataFrame으로 저장했습니다. 또한 Scaling을 통해 Neural net 성능을 높이고자 했습니다.

#날짜순 정렬
data=data.set_index("날짜")
data=data.sort_index(ascending=True)

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

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

그 다음으로, 이전 포스팅에서 한 것과 같이 return 칼럼을 추가하고 Sequence 데이터를 구성하였습니다. 이번에는 30일치 가격 정보를 예측하는데 활용하고자 합니다.

## return
def make_return(data):
    return_list =[0]
    
    for i in range(len(data)-1):
        if (data.iloc[i+1]["Close"]/data.iloc[i]["Close"])-1 >=0:
            return_list.append(1)

        else :
            return_list.append(0)
        
    return return_list 

data_skt["return"]=make_return(data)

## sequence data
def make_data(data, window_size=30):
    feature_list =[]    
    label_list=[]    
    
    if "return" in data.columns:
    
        for i in range(len(data)-window_size):
            feature_list.append(np.array(data.iloc[i:i+window_size]["Close"]))            
            label_list.append(np.array(data.iloc[i+window_size]["return"]))

        data_X = np.array(feature_list)        
        data_Y = np.array(label_list)
        
        return data_X, data_Y 
    
    else : 
        
        for i in range(len(data)-5):
            feature_list.append(np.array(data.iloc[i:i+window_size]["Close"]))
            
        data_X = np.array(feature_list)
        
    
        return data_X
    
skt_X, skt_Y = make_data(data_skt)

## 날짜별, Train/Test 데이터 셋 구분
train_data, train_label = skt_X[:-300], skt_Y[:-300]
test_data, test_label = skt_X[-300:], skt_Y[-300:]

CNN은 Pytorch를 활용하여 구현하겠습니다. 예전에는 Tensorflow와 Keras를 종종 활용하였는데, 최근에 Pytorch를 접하고 numpy와 비슷한 연산에 매료되어 Pytorch를 일부러라도 더 활용하려고 하고 있습니다. 

따라서, 데이터를 Pytorch에서 학습할 수 있도록 tensor로 변경해주고 입력에 요구되는 Shape으로 변경해줍니다.

train_data = torch.FloatTensor(train_data)
train_data = train_data.view(train_data.shape[0], 1, train_data.shape[1]) ## 배치수, 채널, Row

train_label = torch.FloatTensor(train_label)
train_label = train_label.view(train_label.shape[0],1)

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

test_data = torch.FloatTensor(test_data)
test_data = test_data.view(test_data.shape[0],1,test_data.shape[1])

test_label = torch.FloatTensor(test_label)
test_label = test_label.view(test_label.shape[0],1)

다음으로 Network를 구성해보겠습니다. Sequence 정보는 이미지와 같이 2D가 아니라 1D 이기 때문에 Filter와 Pooling Layer도 1D로 구성했습니다. 마지막에 Fully Connected Layer을 3개 연결해줍니다.

class CNN(nn.Module):
    def __init__(self):
        super(CNN, self).__init__()
        
        self.conv1 = nn.Conv1d(1, 5, 3) ## 1*30-> 5*28
        self.relu1 = nn.ReLU()
        self.max1d1 = nn.MaxPool1d(2, stride =2) ## 5*28 -> 5*14
        self.conv2 = nn.Conv1d(5,10,3) ## 5*14->10*12
        self.relu2 = nn.ReLU()
        self.max1d2 = nn.MaxPool1d(2, stride = 2) ##10*12->10*6
        
        self.fc1 = nn.Linear(60, 30)
        self.fc2 = nn.Linear(30, 15)
        self.fc3 = nn.Linear(15, 1)     
        
        
    def forward(self,x):
        x = self.conv1(x)
        x = self.relu1(x)
        x = self.max1d1(x)
        x = self.conv2(x)
        x = self.relu2(x)
        x = self.max1d2(x)
        x= x.view(-1, 60)
        
        x= self.fc1(x)
        x= self.fc2(x)
        x= self.fc3(x)
       
        return x

이제, 구성한 Network를 학습시켜보겠습니다.

model = CNN()
loss_fn = nn.BCEWithLogitsLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)


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

마지막으로 정확도를 추정해보겠습니다.

pred_=torch.round(torch.sigmoid(model(test_data)))
test_pred_=pred_.detach().numpy()

print("Accuracy:",metrics.accuracy_score(test_label, test_pred_))

정확도는 약 44% 나왔습니다.  Parameter나 Network 구성을 정말 임의로 했기 때문에 좋은 결과를 기대하기는 어려울 것 같습니다.

그리고 요즘 주가 방향을 예측하는 모형을 만들면서(아무렇게나 만드는 모형이지만...) 연구자료에서 결과부분만 조금씩 살펴보는데 60%넘는 예측력을 가지는 모델을 찾기 쉽지 않은 것 같습니다. 

저도 57~8%정도의 정확도를 달성하는 모형을 만들 수 있도록 여러가지 모형을 적용해보는 것과 동시에 앞선 선구자들이 적용했던 것도 참고하도록 해야겠습니다.

 

참조

https://pytorch.org/docs/master/generated/torch.nn.Conv1d.html

https://pytorch.org/docs/master/generated/torch.nn.MaxPool1d.html

https://jaeyung1001.tistory.com/45

https://towardsdatascience.com/pytorch-tabular-binary-classification-a0368da5bb89