소스파일은 github.com/galaxywiz/StockCrawler_py 에서 확인 가능합니다.

 

이제 봇까지 만들었으니 이 봇을 실행시켜 주는 시작점을 만들어 주도록 합시다.

Main 은 말 그대로 시작점이니, 봇을 셋팅 해서 만들어 준 다음 이 봇을 어떻게 굴리는지에 대해 보시면 좋을 거 같습니다.

 

main.py

### 라이브러리 설치
# python -m pip install --upgrade pip

### 필요 라이브러리들
# pip install pandas pandas-datareader dataframe
# pip install psutil requests schedule telepot yfinance beautifulsoup4 plotly  

### 딥러닝에 필요한 것들
# pip install tensorflow keras matplotlib astroid==2.2.5 pylint==2.3.1

### 윈도우에서 talib 설치시 아래 링크에서 파이썬 버젼에 맞는 (cpXX-amd64)로 다운 받아서 해줘야 한다.
# https://www.lfd.uci.edu/~gohlke/pythonlibs/#ta-lib
# pip install .\TA_Lib-0.4.19-cp38-cp38-win_amd64.whl

# 리눅스에선 그냥 
# pip install ta-lib
# 안될경우 리눅스에서 설치법
# $ wget http://prdownloads.sourceforge.net/ta-lib/ta-lib-0.4.0-src.tar.gz
# $ tar -xvf ta-lib-0.4.0-src.tar.gz 
# $ cd ta-lib
# $ ./configure --prefix=/usr
# $ make

# 이 프로그램 설계
# 1. 웹에서 인기 있는 종목 수집 (거래량 순)
# 2. 웹에서 데이터 수집 (한국 주식 / 미국 주식)
# 3. 전략 테스팅 매수 / 매도 시스널 포착
# 4. 3에서 포착된 주식 차트 그리고 텔레그램으로 전송

import os
import time
import signal
import sys

import botConfig
from bot import Bot

def test(bot):
    bot.getStocksList()
    bot.checkStrategy()

def signalHandler(sig, frame):
    print('You pressed Ctrl+C!')
    sys.exit(0)
#-----------------------------------------
# 메인 함수 시작
if __name__ == '__main__':
    signal.signal(signal.SIGINT, signalHandler)
    
    usaBot = Bot(botConfig.USABotConfig())
    koreaBot = Bot(botConfig.KoreaBotConfig())

    test(usaBot)
    test(koreaBot)

    botList = []
    botList.append(usaBot)
    botList.append(koreaBot)

    while(True):
        now = time.localtime()
        if now.tm_wday < 6:  ## 일요일은 안함
            for bot in botList:
                bot.do()

        current = "%04d-%02d-%02d %02d:%02d:%02d" % (now.tm_year, now.tm_mon, now.tm_mday, now.tm_hour, now.tm_min, now.tm_sec)
        print("지금시간 : [%s]" % current)
        time.sleep(60)

    

28라인까지는 이 프로그램 돌리기 위한 셋팅이나 아이디어에 대해서 적어 놨습니다.

실제 함수는 47라인에서 시작합니다.

 

48라인 보시면 시그널SIGINT 이런 거 받아오는데 저게 뭐냐 면 이 프로그램이 돌다가 Ctrl + C 입력이 들어오면 어떻게 처리할 것인지를 정의한 코드 입니다. 42라인 함수를 실행하는데 보면 exit로 프로그램을 강제로 꺼주게 하죠.

50라인에서 봇을 세팅해서 만들어 주고, 56 list안에 봇들을 넣어줍니다.

 

이후 while(true)로 무한 루프를 돌면서 1분마다 한 번씩 bot 들을 실행시켜 줍니다.

실행하면 아래 화면처럼 나올 겁니다.

 

'재테크 > 주식 알람 시스템 만들기' 카테고리의 다른 글

10. Bot  (0) 2020.11.08
9. 머신 러닝에 대해서  (0) 2020.11.08
8-3. 단기 변동성 돌파 전략  (0) 2020.11.08
8-2. MACD 전략  (0) 2020.11.08
8-1. 골든 크로스 전략  (0) 2020.11.08

소스파일은 github.com/galaxywiz/StockCrawler_py 에서 확인 가능합니다.

 

그동안 코드 읽으시느라 고생 많으셨습니다.
이제 지금까지 만들어 놓은 파츠를 조립하여 움직이는 몸통을 만들어 봅시다.

우선 파트가 2부분으로 되어 있습니다.
봇의 환경설정을 해주는 botConfig.py 라는 것과 이 botConfig라는 걸 이용해서 실제 돌아가는 bot.py 입니다.

먼저 환경 설정에 대해서 생각해 봅시다.

 

botConfig.py

import dataframe
from datetime import datetime
from datetime import timedelta
import time
import util

from stockCrawler import USAStockCrawler, KoreaStockCrawler
from sqliteStockDB import DayPriceDB, DayPriceFloatDB
from stockData import StockData, BuyState
from tradeStrategy import MaTradeStrategy, LarryRTradeStrategy, MACDTradeStrategy

# 봇 설정
class BotConfig:
    def crawlingTime(self):
        pass

#---------------------------------------------------------#
class KoreaBotConfig(BotConfig):
    def __init__(self):
        self.telegramToken_ = "1080369141:AAFfXa9y70x-wqR2nJBKCVMNLmNFpm8kwA0"
        self.telegramId_ = "108036914" 
        
        self.isFileLoad_ = False
        #self.listFileName_ = "Kr_watchList.txt"
        self.crawler_ = KoreaStockCrawler()
        self.dayPriceDB_ = DayPriceDB("KoreaStockData.db", "day_price")
        self.chartDir_ = "chart_Korea/"
        self.baseWebSite_ = "http://finance.daum.net/quotes/A%s"
        self.strategy_ = MACDTradeStrategy()
        self.isStock_ = True
        self.limitSize_ = 250
        
    def crawlingTime(self):
        now = time.localtime()
        startHour = 16
        startMin = 30
        if now.tm_wday < 5:
            if now.tm_hour == startHour and now.tm_min >= startMin: 
                return True
        return False

#---------------------------------------------------------#
class USABotConfig(BotConfig):
    def __init__(self):
        self.telegramToken_ = "1080369141:AAFfXa9y70x-wqR2nJBKCVMNLmNFpm8kwA0"
        self.telegramId_ = "108036914" 

        self.isFileLoad_ = False
        #self.listFileName_ = "USA_watchList.txt"
        self.crawler_ = USAStockCrawler()
        self.dayPriceDB_ = DayPriceFloatDB("USAStockData.db","day_price")
        self.chartDir_ = "chart_USA/"
        self.baseWebSite_ = "https://finance.yahoo.com/quote/%s"
        self.strategy_ = MACDTradeStrategy()
        self.isStock_ = True
        self.limitSize_ = 200

    def crawlingTime(self):
        now = time.localtime()
        startHour = 7
        startMin = 0
        if 0 < now.tm_wday and now.tm_wday < 6:
            if now.tm_hour == startHour and now.tm_min >= startMin: 
                return True
        return False

간략하게 한국 주식시장과, 미국 주식 시장 감시하는 설정입니다.
모두 생성자에서 변수 설정하고 이것은 어떤 크롤러 쓰고, 어떤 유를 쓰며, 어떤 전략을 채택하고, 텔레그램에서 url을 열 떄 사용하는 웹사이트는 무엇이며 등

나중에 일본 / 홍콩 / 상하이 같은 거 만드실 때 적절히 수정만 하면 다른 나라 주식 감시에도 큰 도움이 되실 겁니다.

 

다음은 봇에 대한 소스입니다.

기본적으로 __ 로 내부에서 사용하는 함수와, 외부에서 호출 가능한 함수를 구분하면서 코드를 읽으시면 한결 보기 편하실 겁니다.

 

bot.py

# 주식 데이터 수집해서 매매 신호 찾고 처리하는
# 주식 봇을 기술
from enum import Enum
import pandas as pd 
from pandas import Series, DataFrame
import numpy as np
import dataframe
from datetime import datetime
from datetime import timedelta
import time
import os
import shutil
import glob
import locale
import util

import botConfig
from stockCrawler import USAStockCrawler, KoreaStockCrawler
from sqliteStockDB import DayPriceDB, DayPriceFloatDB
from stockData import StockData, BuyState
from telegram import TelegramBot
from stockPredic import StockPredic
from printChart import PrintChart
from telegram import TelegramBot
from tradeStrategy import MaTradeStrategy, LarryRTradeStrategy, MACDTradeStrategy

class Bot:
    REFRESH_DAY = 1

    def __init__(self, botConfig):
        self.stockPool_ = {}
        self.config_ = botConfig
        self.telegram_ = TelegramBot(token = botConfig.telegramToken_, id = botConfig.telegramId_)

        locale.setlocale(locale.LC_ALL, '')
        now = datetime.now() - timedelta(days=1)  
        self.lastCrawlingTime_ = now

    def __process(self):
        self.getStocksList()
        self.checkStrategy()
        
        now = time.localtime()
        current = "%04d-%02d-%02d %02d:%02d:%02d" % (now.tm_year, now.tm_mon, now.tm_mday, now.tm_hour, now.tm_min, now.tm_sec)
        print("[%s] 갱신 완료" % current)

    def __doScheduler(self):
        now = datetime.now()
        print("[%s] run" % self.config_.__class__.__name__)

        if self.config_.crawlingTime():
            elpe = now - self.lastCrawlingTime_
            if elpe.total_seconds() < (60*60*24 - 600):
                return
            
            self.lastCrawlingTime_ = datetime.now()
            self.__process()

    #----------------------------------------------------------#
    def sendMessage(self, log):
        TelegramBot.sendMessage(self, log)
  
    #----------------------------------------------------------#
    # db 에 데이터 저장 하고 로딩!
    def __getStockInfoFromWeb2DB(self, name, code):
        loadDays = 10
        # DB에 데이터가 없으면 테이블을 만듬
        sel = self.config_.dayPriceDB_.getTable(code)
        if sel == 0:
            return None
        elif sel == 1:  # 신규 생성 했으면
            loadDays = 365*5  #5 년치

        # 크롤러에게 code 넘기고 넷 데이터 긁어오기
        df = self.config_.crawler_.getStockData(code, loadDays)
        if df is None:
            print("! 주식 [%s] 의 크롤링 실패" % (name))
            return None

        # 데이터 저장      
        self.config_.dayPriceDB_.save(code, df)
        print("====== 주식 일봉 데이터 [%s] 저장 완료 =====" % (name))

    def __loadFromDB(self, code):
        ret, df = self.config_.dayPriceDB_.load(code)
        if ret == False:
            return False, None

        return True, df

    ## db 에서 데이터 봐서 있으면 말고 없으면 로딩
    def __loadStockData(self, name, code, marketCapRanking):  
        now = datetime.now()
        ret, df = self.__loadFromDB(code)
            
        if ret == False:
            self.__getStockInfoFromWeb2DB(name, code)
            ret, df = self.__loadFromDB(code)
            if ret == False:
                return None
        else:
            dateStr = df.iloc[-1]['candleTime']
            candleDate = datetime.strptime(dateStr, "%Y-%m-%d")
            elpe = (now - candleDate).days
            if elpe > self.REFRESH_DAY:
                self.__getStockInfoFromWeb2DB(name, code)
                ret, df = self.__loadFromDB(code)
                if ret == False:
                    return None

        #30일전 데이터가 있는지 체크
        if len(df) < 35:
            return None

        prevDateStr = df.iloc[-15]['candleTime']
        candleDate = datetime.strptime(prevDateStr, "%Y-%m-%d")
        elpe = (now - candleDate).days
        if elpe > 30:
            print("%s 데이터 로딩 실패" % name)
            return None

        sd = StockData(code, name, df)
        self.stockPool_[name] = sd
        sd.calcIndicator()
        sd.marketCapRanking_ = marketCapRanking

        print("*%s, %s load 완료" % (name, code))

    def getStocksList(self, limit = -1):
        self.stockPool_.clear()
        isFileLoad = self.config_.isFileLoad_
        if isFileLoad:
            fileName = self.config_.listFileName_
            stockDf = self.config_.crawler_.getStocksListFromFile(fileName)
        else: 
            tableLimit = self.config_.limitSize_
            stockDf = self.config_.crawler_.getStocksList(tableLimit)    # 웹에 있는 종목을 긁어온다.
            
        # 주식의 일자데이터 크롤링 / db 에서 갖고 오기
        for idxi, rowCode in stockDf.iterrows():
            name = rowCode['name']
            code = rowCode['code']
            marketCapRanking = rowCode['ranking']
            if type(name) != str:
                continue
            self.__loadStockData(name, code, marketCapRanking)
            if limit > 0:
                if idxi > limit:
                    break
    #----------------------------------------------------------#
    def __checkNowTime(self, sd):
        now = datetime.now()
        nowCandle = sd.candle0()
        dateStr = nowCandle["candleTime"]
        candleDate = datetime.strptime(dateStr, "%Y-%m-%d")
        elpe = (now - candleDate).days
        
        temp = self.REFRESH_DAY
        if now.weekday() == 6:
            temp += 2

        if elpe <= temp:
            return True
        return False

    def __doPredic(self, sd):
        vm = StockPredic(sd)
        predicPrice = vm.predic()
        sd.predicPrice_ = predicPrice[0]
        del vm

    def __drawChart(self, sd):
        # 시그널 차트화를 위한 
        chartMaker = PrintChart() 
        dir = self.config_.chartDir_
        chartPath = chartMaker.saveFigure(dir, sd)
        del chartMaker

        return chartPath

    def checkStrategy(self):
        now = datetime.now()
        time = now.strftime("%Y-%m-%d")
                
        for name, sd in self.stockPool_.items():
            if self.__checkNowTime(sd) == False:
                continue

            nowCandle = sd.candle0()
            nowPrice = nowCandle["close"]

            strategy = self.config_.strategy_
            strategy.setStockData(sd)
            action = BuyState.STAY
            timeIdx = len(sd.chartData_) - 1
            
            # 고전 전략 (EMA 골든 크로스로 판단)
            if strategy.buy(timeIdx):
                self.__doPredic(sd)
                action = BuyState.BUY
                sd.teleLog_ = "[%s][%s] 시총 순위[%d]\n" % (time, sd.name_, sd.marketCapRanking_)
                sd.teleLog_ +=" * [%s] long(매수) 신호\n" % (strategy.__class__.__name__)

            elif strategy.sell(timeIdx):
                self.__doPredic(sd)
                action = BuyState.SELL
                sd.teleLog_ = "[%s][%s] 시총 순위[%d]\n" % (time, sd.name_, sd.marketCapRanking_)
                sd.teleLog_ +=" * [%s] short(매도) 신호\n" % (strategy.__class__.__name__)
             
            sd.strategyAction_ = action

            if sd.strategyAction_ != BuyState.STAY:
                # if self.config_.isStock_:
                #     if sd.strategyAction_ == BuyState.BUY:
                #         if nowPrice > sd.predicPrice_:
                #             continue
                
                # sd.teleLog_ +=" * 금일 종가[%f] -> 예측[%f]\n" % (nowPrice, sd.predicPrice_)
                webSite = self.config_.baseWebSite_ % (sd.code_)
                sd.teleLog_ += webSite

                # 시그널 차트화를 위한 
                chartData = sd.chartData_
                chartData["BuySignal"] = strategy.buyList()
                chartData["SellSignal"] = strategy.sellList()
                chartPath = self.__drawChart(sd)
                if chartPath != None:
                    self.telegram_.sendPhoto(chartPath, sd.teleLog_) 

        self.config_.dayPreDB_.saveStockData(self.stockPool_)

    #----------------------------------------------------------#
    def do(self):
        self.__doScheduler()

제일 마지막, 234 라인의 do 라는 함수가 메인입니다.
이건 단순히 doScheduler 라는 걸 호출 하는데, 이렇게 만든 이유는 밖에서 보기엔 XXX_bot.do 라고 하면 봇에게 뭐 시키라고 하는 것 처럼 보이기 때문에 코드 읽으면서 의미 파악이 쉽기 때문에 저렇게 작성 했습니다.

 

내부에선 스케쥴을 돌게 하는데, 스케쥴이 정의된 48라인 보시면, 아까 botConfig crawlingTime 함수를 호출해서, 지금 이걸 실행해도 되는 시간인지 체크합니다.


물론 하루에 한 번씩 돌아야 하니 돌았을 때 실행되는 시간을 저장해서 연달아 돌지 않도록 장치를 해줍니다.

스케쥴 끝자락58라인에 보면 __process 라고 실제 진행에 대한 함수를 호출합니다.


그럼 40라인 정의를 보면, 주식 데이터 갖고 오고, 주식 전략을 체크해 보고 있죠.

그럼 주식을 갖고 오는 130라인 getStockList를 살펴봅시다.

아까 설정한 botConfig에 의해서 파일리스트로 종목을 얻을지, 웹에서 긁어올지 선택을 합니다. 그렇게 해서 나온 리스트는 147라인의 __loadStockData로 일봉 데이터를 긁어오게 됩니다.

 

__loadStockData는 보시면 살짝 복잡해 보이는 if문 코드들이 앞에 있는데, 이게 처음에 db에서 데이터를 로딩해서 없으면 데이터 크게 긁어 오는 것이고, 있다면, 최근 날짜를 조회해서 지금 시간과 1일이 지났는지 봐서 최대한 웹에서 긁어오지 않도록 하고 있습니다. (느리니까요)

다 로딩되면 124라인처럼 bot내부 pool에 등록을 시켜주고, 기술지표를 계산해 줍니다.

 

여기까지가 getStockList 내용이고요. 그 다음에 실행되는 것은 checkStrategy 함수를 살펴보죠

182라인인데, 각 주식 리스트에 대해서 botConfig에서 지정한 tradeStrategy.py에서 기술된 전략을 기반으로 매수 / 매도를 판단합니다.

 

추가로 매수 매도 판단되면 다음날 주식 종가를 예측하기 위해 __doPredic 이란 함수를 사용하고, 이것은 위의 stockPredic.py에서 작성한 머신 러닝 예측을 통해 값을 받아옵니다.

 

그리고 매수, 매도를 판단 받았다면 214라인주석 처리를 했는데, 이는 머신 러닝 값을 믿는다면, 사용하는 것이고, 전 아직은 아닌 거 같아 주석 처리를 해 놓았습니다.

 

이후 224라인에서 이 주식의 추세 판단을 위해 시각화, 즉 차트를 그려주고, 차트 이미지가 제대로 생성되었으면 229라인처럼 텔레그램으로 이 데이터를 전송합니다.

'재테크 > 주식 알람 시스템 만들기' 카테고리의 다른 글

11. 조립 제작  (0) 2020.11.08
9. 머신 러닝에 대해서  (0) 2020.11.08
8-3. 단기 변동성 돌파 전략  (0) 2020.11.08
8-2. MACD 전략  (0) 2020.11.08
8-1. 골든 크로스 전략  (0) 2020.11.08

소스파일은 github.com/galaxywiz/StockCrawler_py 에서 확인 가능합니다.

 

머신 러닝(Machine Leaning), 줄여서 ML에 대해서 기술하려고 하면, 사실 책 한권 더 내야 할 정도로 내용이 많습니다.

머신 러닝은 역사가 꽤 깊은데 최근에 주목받는 이유는 이제서야 머신 러닝을 개인 PC에서도 할 수 있을 만큼 PC속도나 구조가 개편되었고, 최근 여러 소프트웨어에서 이를 활용하다 보니 많이 체감이 되기 때문입니다.

단순히 알파고란 애가 나와서 이세돌9단 이겨서 그 이후 뭔가 나온 게 아닌 기술이죠.

 

예를 들어 구글 포토란 프로그램() 이 있는데, 이 애는 사진이 클라우드에 무료로 올라가서 저장해주는 것도 (사진 사이즈 제한이 있지만) 멋진데, 사진 속 인물은 구별해서 인물별로 따로 그룹을 묶을 수 있다던 지, 할로윈 옷이라 검색하면 할로윈 옷을 입었던 사진만 나온다던 지가 구현되어 있습니다.

 

이런 것은 기존 프로그램으로는 불가능하였습니다. 왜냐면 사진은 우리가 볼 때 사진이란 형태를 인식하지만 실제 저장 파일 내부는 01의 숫자 조합으로 되어있는 데이터 덩어리 이거든요

사진 파일

 

사진 데이터 내부 모습

 

그럼 어떻게 이런 사람이 하는 것 마냥 구분하게 프로그래밍 하냐?

말로는 간단합니다. 인간의 뇌를 모방해서 프로그래밍 하면 되는 거죠.

그리고 정말 우수한 천재들과 단체(구글)등에서 이를 일반인(?)들이 코딩하기 편하도록 라이브러리화 해주었습니다.

 

우리는 이 라이브러리를 써서 특정 분야의 일을 사람이 하는 것 마냥 코딩해주면 됩니다.

이쪽에 관련하여 더 자세히 알고 싶으시면 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 0

50라인의 predic 함수만 사용하는 형태입니다.

이미 22라인에서 클래스 생성하면서 시가, 고가, 저가, 종가 데이터만 사용해서 24라인처럼 priceTable을 만들고, 이걸 이용해서 53 라인에서 정규화 하는데, 왜냐면 주식의 가격은 컴퓨터가 보기엔 어지러운 숫자들이라 컴퓨터가 알기 쉽도록 이를 01사이의 숫자로 변환해 주는 겁니다.

 

이후 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

소스파일은 github.com/galaxywiz/StockCrawler_py 에서 확인 가능합니다.

 

단기 변동성 돌파는 레리 윌리엄 (Larry R Williams) 라는 기술 투자의 대가로 유명 하신 분이 만드신 매매 방법인데, 일일 단위로 보고 있다가, 오늘 시작가가 전날 range (고가-저가) * 노이즈 이상을 돌파 시, 매수 신호라고 판단하여 진입. 다음날 시가에 정리하는 매매 방법입니다.

 

보통 노이즈 값은 0.7, 0.8 값을 사용하는 편이며,

몇몇 방법들 보니 20, 30일치 분봉들의 평균을 내서 노이즈를 계산하는 등 여러 방법으로 조절 할 수 있습니다

 

[2020-10-31][LG화학]

[LarryRTradeStrategy] short(매도) 신호 잡혔을 때 차트입니다

 

#//---------------------------------------------------//
class LarryRTradeStrategy(TradeStrategy):
    RANGE = 20

    def noice(self, timeIdx):
        return 0.7

    def __buySignal(self, timeIdx):
        chartData = self.stockData_.chartData_
        noice = self.noice(timeIdx)
        candle = chartData.iloc[timeIdx]
        startPrice = candle["start"]
        closePrice = candle["close"]
        stand = startPrice + ((candle["high"] - candle["low"]) * noice)

        if closePrice > stand:
            return True
        return False

    def __sellSignal(self, timeIdx):
        chartData = self.stockData_.chartData_
        noice = self.noice(timeIdx)
        candle = chartData.iloc[timeIdx]
        startPrice = candle["start"]
        closePrice = candle["close"]
        stand = startPrice - ((candle["high"] - candle["low"]) * noice)

        if closePrice < stand:
            return True
        return False

    def buy(self, timeIdx):
        chartData = self.stockData_.chartData_
        if len(chartData) <= self.RANGE:
            return False

        if self.stockData_.position_ == BuyState.STAY:
            if self.__buySignal(timeIdx):
                return True
        return False

    def sell(self, timeIdx):
        chartData = self.stockData_.chartData_
        if len(chartData) <= self.RANGE:
            return False

        if self.stockData_.position_ == BuyState.STAY:
            if self.__sellSignal(timeIdx):
                return True
        return False

    def buyList(self):
        signalList = []
        for i in range(0, self.RANGE):
            signalList.append(np.nan)
        
        chartData = self.stockData_.chartData_

        for i in range(self.RANGE, len(chartData)):
            nowCandle = chartData.iloc[i]
            if self.__buySignal(i):
                signalList.append(nowCandle["close"])
            else:
                signalList.append(np.nan)

        return signalList

    def sellList(self):
        signalList = []
        for i in range(0, self.RANGE):
            signalList.append(np.nan)
        
        chartData = self.stockData_.chartData_

        flag = False
        for i in range(self.RANGE, len(chartData)):
            nowCandle = chartData.iloc[i]
            if flag:
                signalList.append(nowCandle["close"])
                flag = False
                continue

            if self.__buySignal(i):
                flag = True
            
            signalList.append(np.nan)

        return signalList     

지금까지 매매 전략을 코드로 녹여내는 방법에 대해서 기술해 보았습니다.

여기에 양념 토핑 하 듯이, 몇몇 규칙을 더 추가해서 나만의 전략을 만들어 보시다 보면 좋을 결과가 나올 것입니다.

(예를 들어 전날 양봉이 뜨기 전까지 매수를 대기한다든가…)

'재테크 > 주식 알람 시스템 만들기' 카테고리의 다른 글

10. Bot  (0) 2020.11.08
9. 머신 러닝에 대해서  (0) 2020.11.08
8-2. MACD 전략  (0) 2020.11.08
8-1. 골든 크로스 전략  (0) 2020.11.08
8. 매매 전략에 대해서  (0) 2020.11.08

소스파일은 github.com/galaxywiz/StockCrawler_py 에서 확인 가능합니다.

 

MACD(Moving Average Convergence & Divergence) 약자이며, 이동 평균선들 간의 수렴 확산을 보기 좋게 나타낸 지표입니다.

각각의 이동평균선은 확산(Divergence)와 수렴(Convergence)하는 성질을 이용한 매매 지표로, 시간이 흐름에 따라 이동평균선들의 수렴, 확산하는 시기를 포착하여 매수 / 매도 포지션을 취하는 방식입니다.

보통 EMA 9, 12, 26일 이동평균선을 사용합니다.


MACD
선은 EMA12– EMA26일 값이고, 시그널선은 EMA9 선을 사용하며, 오실레이트 값은 MACD선과 시그널선의 차로 계산해서 그래프상 보기 좋은 용도로 사용합니다.

 

일반적으로 MACD가 시그널을 상향 돌파하면 골든 크로스로 판단하여 매수하고,
반대로 하양돌파 하면, 데드 크로스로 판단하여 매도하는 전략입니다.

 

아래 차트는 이 전략을 사용해서 뽑아낸 차트입니다. 실제 hts 앱의 주식 차트 지표와 비교했을 때 같은 결과임을 알 수 있습니다.

[2020-10-30][롯데케미칼]
[MACDStrategy] long(
매수) 신호가 잡혔을 때 차트입니다.

#//---------------------------------------------------//
class MACDTradeStrategy(TradeStrategy):
    TERM = 1
    def __buySignal(self, timeIdx):
        if timeIdx < self.TERM:
            return False
            
        chartData = self.stockData_.chartData_
        prevCandle = chartData.iloc[timeIdx - 1]
        macd = prevCandle["MACD"]
        signal = prevCandle["MACDSignal"]
        if macd > signal:
            nowCandle = chartData.iloc[timeIdx]
            macd = nowCandle["MACD"]
            signal = nowCandle["MACDSignal"]

            if macd < signal:
                return True

        return False

    def __sellSignal(self, timeIdx):
        if timeIdx < 1:
            return False
            
        chartData = self.stockData_.chartData_
        prevCandle = chartData.iloc[timeIdx - 1]
        macd = prevCandle["MACD"]
        signal = prevCandle["MACDSignal"]
        if macd < signal:
            nowCandle = chartData.iloc[timeIdx]
            macd = nowCandle["MACD"]
            signal = nowCandle["MACDSignal"]

            if macd > signal:
                return True

        return False

    def buy(self, timeIdx):
        chartData = self.stockData_.chartData_
        if len(chartData) <= self.TERM:
            return False

        if self.stockData_.position_ == BuyState.STAY:
            if self.__buySignal(timeIdx):
                return True
        return False

    def sell(self, timeIdx):
        chartData = self.stockData_.chartData_
        if len(chartData) <= self.TERM:
            return False

        if self.stockData_.position_ == BuyState.STAY:
            if self.__sellSignal(timeIdx):
                return True
        return False

    def buyList(self):
        signalList = []
        for i in range(0, self.TERM):
            signalList.append(np.nan)
        
        chartData = self.stockData_.chartData_

        for i in range(self.TERM, len(chartData)):
            nowCandle = chartData.iloc[i]
            if self.__buySignal(i):
                signalList.append(nowCandle["close"])
            else:
                signalList.append(np.nan)

        return signalList

    def sellList(self):
        signalList = []
        for i in range(0, self.TERM):
            signalList.append(np.nan)
        
        chartData = self.stockData_.chartData_

        for i in range(self.TERM, len(chartData)):
            nowCandle = chartData.iloc[i]
            if self.__sellSignal(i):
                signalList.append(nowCandle["close"])
            else:
                signalList.append(np.nan)

        return signalList

 

 

코드를 읽어 보시면 이전의 골든 크로스 코드랑 많이 비슷하게 느껴 지실 겁니다. 골든 크로스는 2개의 EMA값을 크로스 하냐 마냐로 판단하고, MACD역시 MACD랑 시그널의 크로스를 하냐 마냐로 판단하기 때문에 코드가 유사하게 작성됩니다.

'재테크 > 주식 알람 시스템 만들기' 카테고리의 다른 글

9. 머신 러닝에 대해서  (0) 2020.11.08
8-3. 단기 변동성 돌파 전략  (0) 2020.11.08
8-1. 골든 크로스 전략  (0) 2020.11.08
8. 매매 전략에 대해서  (0) 2020.11.08
7. 유틸 함수들  (0) 2020.11.08

+ Recent posts