<6. 지표 혼합 수익률 백테스팅 프로그래밍 - (3) >
import pandas as pd
import logging as log
import StockDB as db
import asyncio as asy
import sys
import Utility
from mpldatacursor import datacursor
from matplotlib.dates import DateFormatter
import matplotlib.pyplot as plt
from functools import reduce
log.basicConfig(stream=sys.stdout, level=log.DEBUG)
async def db_init(loop, during, type,divide):
log.info("initializing Database")
global dB
try:
dB= db.StockDB( during , type,divide)
await dB.init_pool(loop)
except Exception as e:
log.warning("Database init Error : {}".format(repr(e)))
async def load_kospi_kosdaq_profit():
log.info("Loading {} kospi/kosdaq profit")
global dB
global start
global end
try:
result = await dB.req_kospi_kosdaq_data(start, end)
criteria = result[0].iloc[0][0]
kospi_profit = pd.DataFrame((result[0]/criteria).Close.astype(float).rename("kospi buy/hold"))
criteria = result[1].iloc[0][0]
kosdaq_profit = pd.DataFrame((result[1] / criteria).Close.astype(float).rename("kosdaq buy/hold"))
except Exception as e:
log.warning("Loading kospi/kosdaq profit Error : {}".format(repr(e)))
return pd.merge(kospi_profit, kosdaq_profit, how='outer', left_index=True, right_index=True)
async def show_chart(data,title):
log.info("Preparing chart")
global start
global end
try:
index_profit = await load_kospi_kosdaq_profit()
final_data = pd.merge(data, index_profit, how='outer', left_index=True, right_index=True)
final_data = final_data.fillna(method='ffill').fillna(method='bfill')
axes = final_data.plot(grid=True, title=title)
lines = axes.get_lines()
fmt = DateFormatter('%Y-%m-%d')
datacursor(lines,
formatter=lambda **kwargs: 'Return : {y:.4f}'.format(**kwargs) + '\ndate: ' + fmt(kwargs.get('x')))
plt.show()
except Exception as e:
log.warning("Preparing chart Error : {}".format(repr(e)))
raise
async def load_backtesting_data(start, end, quarter):
log.info("Loading backtesting data")
global market
global dB
try:
if market == Utility.ALL:
try:
data = await dB.req_backtesting_data(Utility.KOSPI, start, end, quarter)
except ResourceWarning as re:
data = pd.DataFrame()
try:
data = data.append(await dB.req_backtesting_data(Utility.KOSDAQ, start, end, quarter))
except ResourceWarning as re:
if data.empty is True:
return
else:
data = await dB.req_backtesting_data(market, start, end, quarter)
data['Date'] = data['Date'].astype('datetime64[ns]')
profit_by_date = pd.DataFrame(data.groupby(data['Date'])['Profit'].mean())
except ResourceWarning as re:
log.warning("No Data")
return
except Exception as e:
log.warning("Loading backtesting data Error : {}".format(repr(e)))
raise
return profit_by_date
def sorted_data(datas):
dic = {}
sorted_data = []
for i, data in enumerate(datas):
if data is None:
continue
dic[data.index[0]] = i
dic = sorted(dic.items())
for i in dic:
sorted_data.append(datas[i[1]])
return sorted_data
async def backtesting(type):
log.info("Starting backtesting")
global start
global end
try:
date_settings = Utility.get_date_ranges(start, end)
quarter_settings = Utility.get_quarter_ranges(start, end)
futures = [asy.ensure_future(load_backtesting_data(date_setting[0], date_setting[1], quarter_settings[i]))
for i, date_setting in enumerate(date_settings)]
results = await asy.gather(*futures)
sort_results = sorted_data(results)
profit = reduce(lambda df1, df2: df1.append(df2), sort_results)
cul_profit = (profit + 1 - 0.004).cumprod()
await show_chart(cul_profit,type)
except Exception as e:
log.info("Backtesting Error : {}".format(repr(e)))
raise
async def main(loop):
global market
global start
global end
market = Utility.KOSDAQ
during = "6"
divide = 20
start="2008-01-01"
end="2018-01-01"
type = Utility.TMV_PBR
await db_init(loop, during, type,divide)
await backtesting(type)
loop = asy.get_event_loop()
loop.run_until_complete(main(loop))
큰 부분부터 하나씩 세세하게 설명하겠다. 먼저 main() 함수를 보자. market은 코스닥/코스피 종목을 대상으로 할 지 아니면 전체 시장을 대상으로 할지 선택한다. during은 앞에서 많이 언급했다. Scope 값을 결정하는 변수이다. divide는 지표 기준으로 몇 등분할지 결정하는 변수이다. 20을 대입하면 지표 기준 상위 또는 하위 5% 주식을 가져온다. start와 end는 백테스팅 기간이다. type은 어떤 지표들을 가지고 백테스팅할 지 결정하는 변수이다. type에 들어갈 변수들은 이전에 설명했던 Utility.py 에 있다. 아래 코드는 시가 총액과 PBR 이 하위 5%인 주식들의 2008년 1월 1일부터 2018년 1월 1일까지의 수익율 구해 백테스팅한다.
async def main(loop):
global market
global start
global end
market = Utility.ALL # Utility.KOSPI, Utility.KOSDAQ
during = "6"
divide = 20
start="2008-01-01"
end="2018-01-01"
type = Utility.TMV_PBR
await db_init(loop, during, type,divide)
await backtesting(type)
loop = asy.get_event_loop()
loop.run_until_complete(main(loop))
db_init() 메서드는 코드가 쉽기에 생략한다. backtesting() 메서드를 보자. date_settings와 quarter_settings 를 구한다. 구하는 메서드는 이전 포스트에서 설명했다. load_backtesting_data()는 인자로 받은 기간과 분기 정보를 토대로 매수/매도 수익율을 비동기적으로 반환한다. 비동기적으로 실행되기 때문에 각 load_backtesting_data()의 실행 순서와 리턴 순서가 달라진다. 따라서 리턴된 데이터들의 시간적 순서를 맞춰야 한다. 누적 수익율을 구하기 위해서는 시간적 연속성이 필요하다. 2010년 1월부터 2012년 1월 누적 수익율을 구한다고 가정하자. 그러면 results는 [2011년부터 2012까지 수익율, 2010년부터 2011년까지의 수익율]이 결과로 된다. 이들 순서는 무작위로 된다. 그러면 무엇이 시간적으로 앞에 있는 지 확인할 수 없다. 따라서 정렬을 해주기 위해 sorted_data 메서드를 이용한다. 시간적으로 정렬됬다면 reduce와 append를 이용해 각 연도 수익율 데이터를 합치고 cumprod()를 이용해 누적곱을 구한다. 0.004는 슬리피지 값이다. 마지막으로 show_chart()를 호출해 차트를 보여준다.
async def backtesting(type):
log.info("Starting backtesting")
global start
global end
try:
date_settings = Utility.get_date_ranges(start, end)
quarter_settings = Utility.get_quarter_ranges(start, end)
futures = [asy.ensure_future(load_backtesting_data(date_setting[0], date_setting[1], quarter_settings[i]))
for i, date_setting in enumerate(date_settings)]
results = await asy.gather(*futures)
sort_results = sorted_data(results)
profit = reduce(lambda df1, df2: df1.append(df2), sort_results)
cul_profit = (profit + 1 - 0.004).cumprod()
await show_chart(cul_profit,type)
except Exception as e:
log.info("Backtesting Error : {}".format(repr(e)))
raise
load_backtesting_data() 메서드는 위에서 말한 "각 지표 순위별 수익률 백테스팅 프로그래밍"에서 나오는 부분이기에 생략한다.
sorted_data는 DB에서 불러온 수익율 데이터들을 시간적으로 정렬한다. 정렬하는 데 각 수익율 데이터에 있는 일자 데이터를 이용한다. 각 수익율 데이터는 데이터 프레임으로 'Date' 인덱스와 "Profit" 칼럼으로 구성되 있다. 이들은 연도별로 구분되기에 겹치는 날짜가 없다. 따라서 Date 인덱스의 첫 번째 값을 기준으로 정렬한다. dic[data.index[0]] = i 는 각 수익율 데이터 프레임의 날짜 데이터를 키로하고 enumerate()로 반환된 i 값을 값으로 하는 딕셔내리를 만든다. 이 i는 i의 키 값을 가지는 datas의 데이터 프레임 인덱스를 가리킨다. 이 딕셔내리를 키 값을 기준으로 정렬한다면 딕셔내리는 날짜 순으로 배열될 것이다. 날짜 순으로 정렬된 딕셔내리의 값은 해당 날짜를 가진 datas의 위치 인덱스이기에 위치 인덱스를 토대로 정렬된 순서대로 새로운 리스트에 데이터를 추가하면 된다. 다음은 이에 대한 로직이다. sorted(dic.items())를 호출하면 키 값을 기준으로 정렬된다. 다만 이 때 데이터 형이 딕셔내리에서 리스트 형태로 바뀐다. 그 다음 for문을 통해 정렬된 순서대로 접근해 datas[i[1]]을 순서대로 리스트에 넣으면 된다. i[1]이 datas의 위치 인덱스가 된다.
def sorted_data(datas):
dic = {}
sorted_data = []
for i, data in enumerate(datas):
if data is None:
continue
dic[data.index[0]] = i
dic = sorted(dic.items())
for i in dic:
sorted_data.append(datas[i[1]])
return sorted_data
show_chart() 메서드와 load_kospi_kosdaq_profit() 메서드는 "각 지표 순위별 수익률 백테스팅 프로그래밍"에서 나오는 부분이기에 생략한다.
아래 사진은 해당 프로그램의 결과 화면이다. 위 전체 소스에 나와있는 설정으로 백테스팅했다. 최대 41배다. 하지만 2018년도 까지 반토막 이상이 났다. 물론 10년간 11배긴 하지만.....
여기까지 해서 시가 총액/PER/PBR 기준 변동성 돌파 전략 분석 프로그래밍 시리즈의 프로그래밍 부분을 마쳤다. 다음 포스트에서는 지금까지 구현한 프로그램들을 토대로 시가 총액/PER/PBR 기준 변동성 돌파 전략이 유용한지 분석해볼 예정이다.
참고로 필자는 컴퓨터 공학과를 재학 중인 대학생입니다. 따라서 코드가 완벽할 수 없습니다. 알고리즘이나 코드가 비효율적이거나 오류가 있다면 댓글 달아주세요..
