戦略テスト(Vectorized)
データ収集
これまでの位置づけでは,単純に1000個の3点棒でバックグラウンドテストを行ったが,量は3日未満であったため,戦略検証における信頼性は低かった.基本編で受信した3カプセル化データのAPIを利用して,3年以上の3カプセル化データでDBを構築する.
データベースにデータを入れる
from binance.spot import Spot
import datetime
import mplfinance as mpf
import pandas as pd
# MYSql 디비의 테이블에 데이터 넣기
# 디비 및 테이블 생성에 대해서는 해당 블로그에서 다루지 않음. 구글링 참조
import time
import pymysql
con = pymysql.connect(host='jikding.net', user='lazydok', password='13cjswo79', db='fin', charset='utf8')
# cur = con.cursor(pymysql.cursors.DictCursor)
cur = con.cursor()
client = Spot()
print(client.time())
client = Spot(key='', secret='') # 본인의 바이낸스 KEY 입력
def insert_chart(data, symbol, interval):
for i in range(len(data)):
data[i][0] = pd.to_datetime(data[i][0], unit='ms').strftime('%Y-%m-%d %H:%M:%S')
data[i][6] = pd.to_datetime(data[i][6], unit='ms').strftime('%Y-%m-%d %H:%M:%S')
# print(data[i])
# raise Exception
sql = """
REPLACE INTO CHART_{}_{}
VALUES (
%s,%s,%s,%s,%s,
%s,%s,%s,%s,%s,
%s,%s
)
""".format(symbol, interval.upper())
cur.executemany(sql, data)
con.commit()
UTC_PLUS_9 = 9 * 60 * 60 * 1000
start_time = int(time.mktime(datetime.datetime.strptime('2018-01-01 00:00:00', '%Y-%m-%d %H:%M:%S').timetuple()) * 1000)
start_time += UTC_PLUS_9
stop_time = int(time.mktime(datetime.datetime.strptime('2021-01-01 00:00:00', '%Y-%m-%d %H:%M:%S').timetuple()) * 1000)
stop_time += UTC_PLUS_9
end_time = 0
while end_time < stop_time:
print(pd.to_datetime(start_time, unit='ms'))
end_time = start_time + 1000 * 60 * 3 * 1000 - 1
data = client.klines(symbol='BTCUSDT', interval='3m', startTime=start_time, endTime=end_time, limit=1000)
insert_chart(data, 'BTCUSDT', '3m')
start_time = end_time
time.sleep(60/1200)データの読み込み
sql = "SELECT * FROM CHART_BTCUSDT_3M WHERE DATETIME BETWEEN '20210101' AND '20220312'" # 21년 ~ 22년 3월 까지 데이터
cur.execute(sql)
data = cur.fetchall()
df = pd.DataFrame(data, columns=
[
'datetime',
'open',
'high',
'low',
'close',
'volume',
'closeTime',
'QuoteAssetVolume',
'NumTrades',
'TakerBuyBaseAssetVolume',
'TakerBuyQuoteAssetVolume' ,
'Ignore'
]
)
# df['datetime'] = pd.to_datetime(df['datetime'], unit='ms')
df.set_index('datetime', inplace=True)
df = df[['open', 'high', 'low', 'close', 'volume']]
df
208471件のデータを戻しました.
Vectorized Back Testingとは?
イベント駆動のメリットとデメリット
For文で以上の20万個のデータが発生したかどうかを一つ一つ検証し計算する方法が最も現実的で、論理的にも最も簡単です.実生活に戦略を入れるときもイベント駆動で行うのが当たり前の方法です.しかし,戦略をテストするためには数回の試みや調整を行わないことが重要である.
前の位置決めで背もたれテストを行います.
これはMACD信号を簡単に使う戦略です.
acc = {
'CASH': 1,
'F_BTC': {
'LONG': {'QTY': 0, 'MARGIN': 0, 'PRC': 0, 'DATETIME': None},
'SHORT': {'QTY': 0, 'MARGIN': 0, 'PRC': 0, 'DATETIME': None}
}
}
eval_amt_hist = []
td_hist = []
pre_prc = 0
s_t = time.time()
for i, (date, row) in zip(range(len(df)), df.iterrows()):
long = acc['F_BTC']['LONG']
short = acc['F_BTC']['SHORT']
if i == 0: # 최초
pre_macd_diff = 0
long['DATETIME'] = date
short['DATETIME'] = date
print(acc)
elif i + 1 == len(df): # 마지막
''
else: # 나머지
pre_macd_diff = df.iloc[i-1]['MACD_DIFF']
now_macd_diff = row['MACD_DIFF']
prc = df.iloc[i+1]['close'] # 매수, 매도시 가격은 이벤트 발생 다음 봉의 시가
if pre_macd_diff < 0 and now_macd_diff > 0:
# SHORT 청산
if short['QTY'] > 0:
acc['CASH'] += (short['PRC'] - prc) * short['QTY'] + short['MARGIN']
rt = short['PRC']/prc - 1
term = (date - short['DATETIME']).total_seconds()
td_hist.append({'RETURN': rt, 'TERM': term})
acc['F_BTC']['SHORT'] = {'QTY': 0, 'MARGIN': 0, 'PRC': 0, 'DATETIME': None}
short = acc['F_BTC']['SHORT']
# LONG 진입
cash = acc['CASH']
qty = cash / prc
long['QTY'] = qty
long['PRC'] = prc
long['MARGIN'] = cash
long['DATETIME'] = date
acc['CASH'] -= cash
elif pre_macd_diff > 0 and now_macd_diff < 0:
# LONG 청산
if long['QTY'] > 0:
acc['CASH'] += (prc - long['PRC']) * long['QTY'] + long['MARGIN']
rt = prc/long['PRC'] - 1
term = (date - long['DATETIME']).total_seconds()
td_hist.append({'RETURN': rt, 'TERM': term})
acc['F_BTC']['LONG'] = {'QTY': 0, 'MARGIN': 0, 'PRC': 0, 'DATETIME': None}
long = acc['F_BTC']['LONG']
#SHORT 진입
cash = acc['CASH']
qty = cash / prc
short['QTY'] = qty
short['PRC'] = prc
short['MARGIN'] = cash
short['DATETIME'] = date
acc['CASH'] -= cash
eval_amt = acc['CASH']
eval_amt += (row['close'] - long['PRC']) * long['QTY'] + long['MARGIN']
eval_amt += (short['PRC'] - row['close']) * short['QTY'] + short['MARGIN']
eval_amt_hist.append(eval_amt)
pre_macd_diff = row['MACD_DIFF']
print('종료, 실행 시간: {}'.format(time.time() - s_t))
df['return'] = eval_amt_hist
bt = df[['return']].copy()
bt['BM'] = df['close']/df['close'].iloc[0]
bt.plot(figsize=(20,10))
まず、戦略収益率が悪い.少し修正して、再テストするとき、
実行時間は42.81秒を超えた.少し戦略的に数値を変えて、もう一度回すと、また1分近くかかります.1つのプロジェクトを2年にしてもこれだけの時間がかかり、同時に10以上のプロジェクトのポートフォリオバックグラウンドテストを行うには、想像もできない時間がかかります.コンピュータにとって,数十万以上のデータを1つずつFor文演算することは非常に非効率である.
ベクトル化演算とは?
PythonではPandasを使って楽にバックグラウンド演算ができます.例えば、5、6、7、6、5は、この5つの数字に対して1を加算する演算を行う.
イベント駆動の場合、For文の周りに5回の演算を行い、数字ごとに1を加算します.ただし,バックグラウンド演算では5つの数に1を1回加算する内部演算を行う.この演算が数十万回または数百万回を超えると、後置演算と繰り返し文演算の速度に大きな違いがあります.
参考までに、上記の同じデータを用いて同じポリシーテストを行うと、0.5秒程度のバックグラウンド計算が必要となる.すなわち、上記の場合、速度は100倍程度速い.
バックグラウンド計算の欠点
10、11、12、11、10の5つの数について、簡単な演算であれば、文を繰り返す必要がなく、コンピュータは並列演算で簡単に答えを出すことができます.しかし,前後関係による演算は後置演算では実現しにくい.例:
11を超えると買収額が11以下に下がると、投げ売りの戦略がある.バックグラウンド演算で、対応する戦略で、いつ買収、販売を行いますか.また,低条件で行われた取引がどれだけの収益率を生み出したかを計算するには,すぐに一括で計算することはできない.
Vectorized BackTestの実装
Log Return
重複文を使用しないため、特定の条件で買収・売買によって収益率を計算し、Logを使用することができます.
rtr trtをt時間のLog Return、ptp tptをt時間帯の価格、t-1からt時間帯のReturnを
rt=log(ptpt−1)=log(pt)−log(pt−1)r_t = log(\frac{p_t}{p_{t-1}}) = log(p_t) - log(p_{t-1})rt=log(pt−1pt)=log(pt)−log(pt−1)
使えます.
t=1,2,3,すなわち,3つの期間において,ログは以下のように返される.
r3+r2+r1r_3 + r_2 + r_1r3+r2+r1
=(log(p3)−log(p2))+(log(p2)−log(p1))+(log(p1)−log(p0))= (log(p_3) - log(p_2)) + (log(p_2) - log(p_1)) + (log(p_1) - log(p_0))=(log(p3)−log(p2))+(log(p2)−log(p1))+(log(p1)−log(p0))
=log(p3)−log(p0)= log(p_3) - log(p_0)=log(p3)−log(p0)
=log(p3p0)= log(\frac{p_3}{p_0})=log(p0p3)
上記の式により,tの0から3までの収益率は簡単なログ戻りをexpするだけでよい.すなわち,tからTまでの収益率は,tからTまでのログバック数の和のみを必要とし,対応する値にexpを取出せばよい.
戦略テスト
まず後置演算でログバックと位置を求める.
# Log Retrun
rs = df['close'].apply(np.log).diff(1)
# 포지션 구하기(MACD가 Signal을 상향 돌파시 롱, 하향 돌파시 숏
pos = df['MACD_DIFF'].apply(np.sign) # +1 if long, -1 if short
# MACD OSC 및 롱 숏 포지션 표현하기
fig, ax = plt.subplots(2,1)
df['MACD_DIFF'].iloc[-1000:].plot(ax=ax[0], title='MACD OSC')
pos.iloc[-1000:].plot(ax=ax[1], title='Position', figsize=(20, 10))
テスト収益率図をログバックと位置で描画します.
my_rs = pos.shift(1)*rs # 다음봉 종가 가격으로 샀다고 가정
bt = pd.DataFrame()
bt['return'] = my_rs.cumsum().apply(np.exp)
bt['BM'] = df['close']/df['close'].iloc[0]
bt.plot(figsize=(20,10))
テストの概要
mdd = ((bt['return'] - bt['return'].cummax()) / bt['return'].cummax()).min()
bm_mdd = ((bt['BM'] - bt['BM'].cummax()) / bt['BM'].cummax()).min()
years = (bt.index[-1] - bt.index[0]).total_seconds()/60/60/24/365
cagr = (bt['return'][-1])**(1/years) - 1
bm_cagr = (bt['BM'][-1] / bt['BM'][0])**(1/years) - 1
sharpe_ratio = (bt['return'][-1] - 1)/bt['return'].std()
test = pd.DataFrame()
test['close_diff_log'] = rs
test['position_chaged'] = pos.shift(1).diff(1).abs()
test['position_chaged'][test['position_chaged'] == 0] = np.nan
test['grp'] = test.index.astype(int)
test['grp'] = test['position_chaged'] * test['grp']
test['grp'] = test['grp'].fillna(method='ffill')
is_win = np.exp(test.groupby('grp')['close_diff_log'].sum()).apply(lambda v: 1 if v > 1 else 0)
num_tds = len(is_win)
win_rate = is_win.sum() / num_tds
win_rate
summary = {
'MDD': round(mdd * 100, 2),
'CAGR': round(cagr * 100, 2),
'ALPHA': round((cagr-bm_cagr) * 100, 2),
'SHARPE_RATIO': round(sharpe_ratio, 2),
'-CAGR/MDD': round(cagr/-mdd*100,2),
'WIN_RATE': round(win_rate*100, 2),
'TOT_TRADE_CNT': num_tds
}
s_df = pd.DataFrame.from_dict(summary, orient='index').rename(columns={0:'Summary'})
s_df
このテストは、イベント駆動方式のように複雑なポリシーをテストすることができず、収益率自体も正確ではありません.しかし,アイデア生成とポリシー調整により,後で局方式で複数回作成したポリシーを必要とし,イベント駆動のバックグラウンドテストにより詳細な検証を行う.
漏電距離の適用
# 벡터연산 수수료 다시 구하기.
rs = df['close'].apply(np.log).diff(1)
pos = df['MACD_DIFF'].apply(np.sign) # +1 if long, -1 if short
tc = 0.003 # assume transaction cost of 0.3%
tc_rs = pos.diff(1).abs() * tc
my_rs1 = pos.shift(1)*rs # don't include costs
my_rs2 = pos.shift(1)*rs - tc_rs
bt = pd.DataFrame()
bt['return'] = my_rs2.cumsum().apply(np.exp)
bt['BM'] = df['close']/df['close'].iloc[0]
my_rs1.cumsum().apply(np.exp).plot(figsize=(20, 10))
bt['return'].plot()
bt['BM'].plot()
plt.legend(['without transaction costs', 'with transaction costs', 'Benchmark'])
Reference
この問題について(戦略テスト(Vectorized)), 我々は、より多くの情報をここで見つけました https://velog.io/@lazydok/전략-백테스팅Vectorizedテキストは自由に共有またはコピーできます。ただし、このドキュメントのURLは参考URLとして残しておいてください。
Collection and Share based on the CC Protocol