소스파일은 github.com/galaxywiz/StockCrawler_py 에서 확인 가능합니다.
머신 러닝(Machine Leaning), 줄여서 ML에 대해서 기술하려고 하면, 사실 책 한권 더 내야 할 정도로 내용이 많습니다.
머신 러닝은 역사가 꽤 깊은데 최근에 주목받는 이유는 이제서야 머신 러닝을 개인 PC에서도 할 수 있을 만큼 PC속도나 구조가 개편되었고, 최근 여러 소프트웨어에서 이를 활용하다 보니 많이 체감이 되기 때문입니다.
단순히 알파고란 애가 나와서 이세돌9단 이겨서 그 이후 뭔가 나온 게 아닌 기술이죠.
예를 들어 구글 포토란 프로그램(앱) 이 있는데, 이 애는 사진이 클라우드에 무료로 올라가서 저장해주는 것도 (사진 사이즈 제한이 있지만) 멋진데, 사진 속 인물은 구별해서 인물별로 따로 그룹을 묶을 수 있다던 지, 할로윈 옷이라 검색하면 할로윈 옷을 입었던 사진만 나온다던 지가 구현되어 있습니다.
이런 것은 기존 프로그램으로는 불가능하였습니다. 왜냐면 사진은 우리가 볼 때 사진이란 형태를 인식하지만 실제 저장 파일 내부는 0과 1의 숫자 조합으로 되어있는 데이터 덩어리 이거든요
그럼 어떻게 이런 사람이 하는 것 마냥 구분하게 프로그래밍 하냐?
말로는 간단합니다. 인간의 뇌를 모방해서 프로그래밍 하면 되는 거죠.
그리고 정말 우수한 천재들과 단체(구글)등에서 이를 일반인(?)들이 코딩하기 편하도록 라이브러리화 해주었습니다.
우리는 이 라이브러리를 써서 특정 분야의 일을 사람이 하는 것 마냥 코딩해주면 됩니다.
이쪽에 관련하여 더 자세히 알고 싶으시면 https://www.youtube.com/watch?v=p_RdUAOaKgM
에 있는 구글에서 내놓은 머신 러닝 강좌를 보시면 좋습니다.
일단 구글에서 머신 러닝을 위해 텐서플로우란 라이브러리를 내놓은 건 알겠고, 이제 나 대신 컴퓨터가 대신 주식 패턴을 찾다가 다음날 오를 거 같으면 내게 알려주면 대박이지 않을까 한 생각을 가지실 수 있을 거 같은데요.
그거에 관련해서 아래와 같은 코드를 작성했습니다.
stockPredic.py
# 텐서플로어를 사용해서 주식 데이터를 예측해 보자
# 과거 데이터를 기반으로 패턴을 머신러닝으로 찾아내서
# 이걸 가지고 다음날 데이터를 예측.
# 물론 미래일은 불확실이므로 정말 맞는건 아니고
# 특별한 이슈가 없으면, 아마 60,70%로 맞을지도 정도
import tensorflow as tf
import numpy as np
import pandas as pd
import datetime
import gc
import os
from stockData import StockData
class StockPredic:
def __init__(self, stockData):
# 하이퍼파라미터
self.seqLength_ = 28 # 1개 시퀀스의 길이(시계열데이터 입력 개수)
self.epochNum_ = 50 # 에폭 횟수(학습용전체데이터를 몇 회 반복해서 학습할 것인가 입력)
self.tenserDir_ = "tensorSave/"
self.stockData_ = stockData
futureConsidered = ["start", "high", "low", "close"]
self.priceTable_ = stockData.chartData_[futureConsidered]
self.inColumnCnt_ = len(futureConsidered)
self.outputCnt_ = 1 # 결과데이터의 컬럼 개수
self.priceTable_ = self.priceTable_.values[1:].astype(np.float)
#----------------------------------------------------------#
def __modelName(self):
name = "model_%s" % (self.stockData_.code_)
return name
# 너무 작거나 너무 큰 값이 학습을 방해하는 것을 방지하고자 정규화한다
# x가 양수라는 가정하에 최소값과 최대값을 이용하여 0~1사이의 값으로 변환
# Min-Max scaling
def __minMaxScaling(self, x):
npX = np.asarray(x)
return (npX - npX.min()) / (npX.max() - npX.min() + 1e-7) # 1e-7은 0으로 나누는 오류 예방차원
# 정규화된 값을 원래의 값으로 되돌린다
# 정규화하기 이전의 org_x값과 되돌리고 싶은 x를 입력하면 역정규화된 값을 리턴한다
def __inverseMinMaxScaling(self, orgX, x):
orgNpX = np.asarray(orgX)
npX = np.asarray(x)
return (npX * (orgNpX.max() - orgNpX.min() + 1e-7)) + orgNpX.min()
#----------------------------------------------------------#
def predic(self, refresh = False):
try:
price = self.priceTable_[:, :-1]
normPrice = self.__minMaxScaling(price)
volume = self.priceTable_[:, -1:]
normVol = self.__minMaxScaling(volume)
x = np.concatenate((normPrice, normVol), axis=1)
y = x[:, [3]] ####### close 종가가 타겟 #######
dataX = [] # 입력으로 사용될 Sequence Data
dataY = [] # 출력(타켓)으로 사용
for i in range(0, len(y) - self.seqLength_):
_x = x[i : i + self.seqLength_]
_y = y[i + self.seqLength_] # 다음 나타날 주가(정답)
dataX.append(_x) # dataX 리스트에 추가
dataY.append(_y) # dataY 리스트에 추가
# 학습용/테스트용 데이터 생성
# 전체 70%를 학습용 데이터로 사용
trainSize = int(len(dataY) * 0.7)
# 데이터를 잘라 학습용 데이터 생성
trainX = np.array(dataX[0:trainSize])
trainY = np.array(dataY[0:trainSize])
trainX = np.reshape(trainX, (trainX.shape[0], trainX.shape[1], self.inColumnCnt_))
# 데이터를 잘라 테스트용 데이터 생성
testX = np.array(dataX[trainSize:len(dataX)])
testY = np.array(dataY[trainSize:len(dataY)])
path = self.tenserDir_
if os.path.exists(path) == False:
os.makedirs(path)
savePath = "%s%s.h5" % (path, self.__modelName())
with tf.device("/cpu:0"):
if os.path.isfile(savePath) == True:
model = tf.keras.models.load_model(savePath)
model.summary()
print("* %s 모델 로드 [%s] " % (self.stockData_.name_, savePath))
if refresh:
model.fit(trainX, trainY, validation_data=(testX, testY), batch_size=10, epochs=1)
model.save(savePath)
return 0
else:
model = tf.keras.models.Sequential()
model.add(tf.keras.layers.LSTM(256, return_sequences=True, input_shape=(self.seqLength_, self.inColumnCnt_))) # (timestep, feature)
model.add(tf.keras.layers.LSTM(256, return_sequences=False))
model.add(tf.keras.layers.Dense(self.outputCnt_, activation='linear'))
model.compile(loss='mse', optimizer=tf.keras.optimizers.RMSprop(clipvalue=1.0))
model.summary()
model.fit(trainX, trainY, validation_data=(testX, testY), batch_size=10, epochs=self.epochNum_)
model.save(savePath)
# sequence length만큼의 가장 최근 데이터를 슬라이싱한다
recentData = np.array([x[(len(x) - self.seqLength_): ]])
# 내일 종가를 예측해본다
testPredict = model.predict(recentData)
testPredict = self.__inverseMinMaxScaling(price, testPredict) # 금액데이터 역정규화한다
recentData = self.__inverseMinMaxScaling(price, recentData)
tf.keras.backend.clear_session()
del recentData
del model
return testPredict[0]
except:
print("* %s 모델 예측 에러" % (self.stockData_.name_))
return 050라인의 predic 함수만 사용하는 형태입니다.
이미 22라인에서 클래스 생성하면서 시가, 고가, 저가, 종가 데이터만 사용해서 24라인처럼 priceTable을 만들고, 이걸 이용해서 53 라인에서 정규화 하는데, 왜냐면 주식의 가격은 컴퓨터가 보기엔 어지러운 숫자들이라 컴퓨터가 알기 쉽도록 이를 0과 1사이의 숫자로 변환해 주는 겁니다.
이후 74라인처럼 전체 데이터 70%를 학습용으로 쓰고 나머지 최근 30%를 해답용으로 분리하는 작업을 합니다.
이렇게 만들어진 데이터를 텐서플로우 + 케라스 인터페이스를 사용해서 98라인처럼 모델을 만듭니다.
99라인 이후는 신경망 블록을 쌓는 단계이고 이렇게 만들어진 인공두뇌 모델(이하 모델)을 105라인에서 훈련시켜 줍니다. 그리고 저장하는데, 이는 훈련시키는 시간 비용이 많아서 추후 사용시를 위함 입니다. (93라인 보시면 모델이 있으면 로딩해서 바로 처리하죠)
이후 111라인에서 현제 데이터 넣어 예측하고, 이를 결과값으로 리턴 해줍니다.
즉 그러니까 이 코드는 10년치 주식 데이터가 있다면, 2010~2017년까지 데이터를 훈련용으로 입력하고, 컴퓨터가 자체적으로 패턴이 있는지 이것 저것 시도해 보고
2017~2020년 주식 데이터로 맞춰보면서 패턴을 검증하는 프로그램입니다.
몇몇 앱에서 이런 비슷한 유형을 찾는 게 같은데,
아마 내부적으로 저런 프로그램을 돌려서 예측 값을 주는 걸로 보입니다.
'재테크 > 주식 알람 시스템 만들기' 카테고리의 다른 글
| 11. 조립 제작 (0) | 2020.11.08 |
|---|---|
| 10. Bot (0) | 2020.11.08 |
| 8-3. 단기 변동성 돌파 전략 (0) | 2020.11.08 |
| 8-2. MACD 전략 (0) | 2020.11.08 |
| 8-1. 골든 크로스 전략 (0) | 2020.11.08 |
