재테크/주식 알람 시스템 만들기
10. Bot
중년 플머 김씨
2020. 11. 8. 00:22
소스파일은 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라인처럼 텔레그램으로 이 데이터를 전송합니다.
