2020/5/22
この記事は全４回中の第３回です。他の回についてはこちら（ #１ / #２ / #４ ）。 bitFlyerを利用したBitcoinの自動取引について、Pythonと裁定取引についてはある程度わかるけど他はわからないという人のために概要を簡潔に書いていきます。 第３回の今回は実際に動かすコードを書いていきます。 なお、この連載記事は昔自分が行なっていた自動取引をまとめ直すという記事のため、今回示すコードと当時自分が使っていたコードとで実装が異なる部分がかなりあります。 （簡単な動作確認はしたものの）今回示したコードを使って長期間運用した実績はないのであくまでサンプルとして見ていただけると幸いです。
実際に書いていく前にコード全体の構成を考えます。 今回はmmbotやアービトラージのようなbotであることを生かした特殊な手法では無く、単に裁量取引の厳密なルールに基づく自動化というコンセプトでコードを書いていきます。 実際に取引をする部分は終了条件を満たすまで無限ループさせて、そのループの中でentry,exitを繰り返すような構成にし、簡単のため建玉中は新たなentryはしないようにします。また、注文は全て成行で行います。 entry,exitの判断は、今回は現在時刻までのOHLCデータを引数とする関数として実装します。 また、bitFlyerAPIを利用して取引をする関数、テクニカル指標を計算する関数はクラスにまとめることで整理できます。 今回はロジックとテクニカル指標計算の関数をまとめたファイルを classes.py とし、実際に動かすコードを bot.py として作っていきます。
#1で作った関数をクラスとしてまとめていくことで取引所関連の操作の関数を整理することができます。 ただし、それをすでにやってくれたものがpybitflyerなので今回はありがたく使わせてもらうことにします。
Python
import pybitflyer # ***には自分のAPI keyとAPI secretを入れる api = pybitflyer.API(api_key="***",api_secret="***")
準備はこれだけです。bitFlyerに関するAPIは基本的にこの api を用います。
テクニカル指標をクラス Indicators に定義していきます。 基本的には#2で実装したものをクラス用に書き直したものですが、RSIに関してはtalibと比べて自分で書いたコードが遅かったのでtalibがインストールされてる場合にはそちらを使うようにしています。 注意点として、これらのテクニカル指標は入力したOHLCデータの全ての時刻に対して計算するので、 len(ohlc) が大きいとその分計算も遅くなります。 よって ind = Indicators(ohlc) とする際に渡す ohlc は長すぎないようにします。
[コードを表示]
Python:classes.py（前半）
import numpy as np
import pandas as pd
# ta-libの読み込み
try:
import talib
talibflag = True
except ModuleNotFoundError:
print("talibがinstallされていないため一部指標の計算が遅くなります")
talibflag = False
# 指標
class Indicators:
def __init__(self,ohlc):
"""
ohlc : np.array([[timestamp,o,h,l,c],...])
"""
self.ohlc = ohlc # OHLC配列(5,*)
self.closeprice = ohlc[:,4] # 終値配列(*)
def ATR(self,n):
"""
return : ATRのndarray(length=len(ohlc))
"""
if talibflag:
return talib.ATR(self.ohlc[:,2],self.ohlc[:,3],self.closeprice,timeperiod=n)
else:
tr = []
atr = []
for i in range(1,len(self.ohlc)):
tr.append(max(self.ohlc[-i][2],self.ohlc[-i-1][4]) - min(self.ohlc[-i][3],self.ohlc[-i-1][4]))
tr.reverse()
atr = pd.Series(tr).ewm(span=n).mean().values
return atr
# EMA
def EMA(self,n):
"""
return : EMAのndarray(length=len(ohlc))
"""
alpha = 2/(n+1)
x0 = self.closeprice[:n].mean()
ema = [x0]
for i in range(len(self.closeprice)-n):
ema.append(ema[-1]*(1-alpha)+self.closeprice[n+i]*alpha)
nanlist = np.zeros(n-1)
nanlist[:] = np.nan
return np.concatenate([nanlist,np.array(ema)])
# SMA
def SMA(self,n):
"""
return : SMAのndarray(length=len(ohlc))
"""
return pd.Series(self.closeprice).rolling(n).mean().values
# MACD(n1:shortEMA,n2:longEMA,n3:signal)
def MACD(self,n1,n2,n3):
"""
return : MACDのndarray,signalのndarray
"""
macd = self.EMA(n1) - self.EMA(n2)
macd[:n2] = np.nan
signal = pd.Series(macd).rolling(n3).mean().values
return macd,signal
def RSI(self,n):
"""
return : RSIのndarray
"""
if talibflag:
rsi = talib.RSI(self.closeprice,timeperiod=n)
return rsi
else:
RSI_period = n
diff = pd.Series(self.closeprice).diff(1)
positive = diff.clip(lower=0).ewm(alpha=1.0/RSI_period).mean()
negative = diff.clip(upper=0).ewm(alpha=1.0/RSI_period).mean()
rsi = 100 - 100/(1-positive/negative)
return rsi.values
def RCI(self,n):
"""
return: RCIのndarray
"""
rci = []
for j in range(len(self.closeprice) - (n-1)):
table = np.zeros([2,n])
# closeprice[-n:0]になるのを回避
if j == len(self.closeprice)-n:
table[0] = self.closeprice[-n:]
else:
table[0] = self.closeprice[-len(self.closeprice)+j:-len(self.closeprice)+n+j]
table[1] = np.arange(n,0,-1)
sortedtabel = table[:,np.argsort(table[0])]
index = np.arange(n,0,-1)
d = 0
for i in range(n):
d += (index[i]-sortedtabel[1][i])**2
rci.append((1-6*d/(n*(n*n-1)))*100)
nanlist = np.zeros(n-1)
nanlist[:] = np.nan
return np.concatenate([nanlist,np.array(rci)])
def BB(self,n,sigma=2):
base = pd.Series(self.closeprice).rolling(n).mean().values
sig = pd.Series(self.closeprice).rolling(n).std(ddof=1).values
upper_band = base + sigma*sig
lower_band = base - sigma*sig
return upper_band, lower_band
ロジックも同様にクラスで書いていきます。 try_entry(ohlc) ：ohlcを渡すと、売り買いどちらかにエントリーするか何もしないかを判断する関数 sell_exit(ohlc) ：（売りポジションを持っている状態で）ohlcを渡すと、損益を確定させるかどうかを判断する関数 buy_exit(ohlc) ：（買いポジションを持っている状態で）ohlcを渡すと、損益を確定させるかどうかを判断する関数 の３つを Logic_test というクラス内に定義します。 ただし、テストロジックとして「1分足のEMA25とEMA10を使って、ゴールデンクロスで買いエントリー、デッドクロスで売りエントリーして、5000の値幅が動いたら利確/損切り」という非常に単純なものを設定しています。
[コードを表示]
Python:classes.py(後半）
class Logic_test:
"""
EMA25とEMA10のゴールデンクロスで買い、デッドクロスで売り
"""
def __init__(self):
self.state = {'buy':False, 'sell':False}
self.exit = {'settle':False, 'result':None}
self.width = {'base':-1, 'p_width':-1, 'l_width':-1}
def try_entry(self,ohlc):
"""
signalがTrueになる条件（売買のエントリー条件）と
利確損切りのwidthを定義する。
"""
ind = Indicators(ohlc)
self.state['buy'] = self.state['sell'] = False
self.width['p_width'] = self.width['l_width'] = -1
"""
ロジック部分
"""
EMA25 = ind.EMA(25)
EMA10 = ind.EMA(10)
if EMA25[-2] > EMA10[-2] and EMA25[-1] < EMA10[-1]:
self.state['buy'] = True
if EMA25[-2] < EMA10[-2] and EMA25[-1] > EMA10[-1]:
self.state['sell'] = True
# 一定値の値幅で利確損切りをする場合以下を設定する
self.width['p_width'] = self.width['l_width'] = 1000
self.width['base'] = ohlc[-1][4]
return self.state,self.width
def sell_exit(self,ohlc):
"""
ショートポジションの際のexit条件を記載する。
条件を満たすならself.exitのsettleにTrueを、resultにその際の（予想される）損益を入れる。
"""
ind = Indicators(ohlc)
self.exit = {'settle':False, 'result':None}
# 値幅exitの場合
if self.width['p_width'] != -1:
# 利確
if self.width['base'] - ohlc[-1][4] > self.width['p_width']:
self.exit['settle'] = True
self.exit['result'] = self.width['p_width']
# 損切り
if ohlc[-1][4] - self.width['base'] > self.width['l_width']:
self.exit['settle'] = True
self.exit['result'] = -self.width['l_width']
"""
# ドテンの場合
else:
signal,_ = self.try_entry(ohlc)
if signal['buy']:
self.exit['settle'] = True
self.exit['result'] = self.width['base'] - ohlc[-1][4]
# その他のexitの場合
if hogehoge:
self.exit['settle'] = True
self.exit['result'] = self.width['base'] - ohlc[-1][4]
"""
return self.exit
def buy_exit(self,ohlc):
"""
ロングポジションの際のexit条件を記載する。
条件を満たすならself.exitのsettleにTrueを、resultにその際の（予想される）損益を入れる。
"""
ind = Indicators(ohlc)
self.exit = {'settle':False, 'result':None}
# 値幅exitの場合
if self.width['p_width'] != -1:
# 利確
if ohlc[-1][4] - self.width['base'] > self.width['p_width']:
self.exit['settle'] = True
self.exit['result'] = self.width['p_width']
# 損切り
if self.width['base'] - ohlc[-1][4] > self.width['l_width']:
self.exit['settle'] = True
self.exit['result'] = -self.width['l_width']
"""
# ドテンの場合
signal,_ = self.try_entry(ohlc)
if signal['sell']:
self.exit['settle'] = True
self.exit['result'] = ohlc[-1][4] - self.width['base']
# その他のexitの場合
if hogehoge:
self.exit['settle'] = True
self.exit['result'] = ohlc[-1][4] - self.width['base']
"""
return self.exit
辞書型のクラス変数３つはそれぞれ以下のような役割をしています。 self.state = {'buy':False, 'sell':False} ：ポジションを持っているかの状態を格納 self.exit = {'settle':False, 'result':None} ：決済注文をするかどうかの判断と、そのときの期待損益を格納 self.width = {'base':-1, 'p_width':-1, 'l_width':-1} ：ポジションを持ったときの値段と、利確損切りの値幅を格納 entryやexitの条件を書き換えることで様々なロジックを実現できます。
さて、実際に取引する部分を書いていきます。
はじめに説明した通り、無限ループの中で「ポジションなし」「買いポジション」「売りポジション」の３状態を推移します。
[コードを表示]
Python:bot.py
# coding: utf-8
import csv,json
import numpy as np
import pandas as pd
import datetime
from pprint import pprint
import matplotlib.pyplot as plt
import time
import requests
# ログ設定
import logging
from logging import getLogger,FileHandler,Formatter
logger = getLogger("bot")
logger.setLevel(logging.INFO)
fh = FileHandler('bot.log')
fh.setLevel(logging.INFO)
format = Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(format)
logger.addHandler(fh)
import pybitflyer
api = pybitflyer.API(api_key="***",api_secret="***")
import classes
# 注文拒否対策をした注文
def sendchildorder(side,size):
cnt = 0
while cnt < 20:
response = api.sendchildorder(product_code="FX_BTC_JPY",child_order_type="MARKET",side=side,size=size)
try:
if bool(response['child_order_acceptance_id']):
break
except:
pass
cnt += 1
time.sleep(2)
# CryptowatchからOHLCデータ取得
def getohlc(periods):
response = requests.get("https://api.cryptowat.ch/markets/bitflyer/btcfxjpy/ohlc",params = {"periods": periods})
ohlc = np.array(response.json()['result'][str(periods)])[-100:,:5] # 長すぎると計算時間が増えるので最新100件に整形
return ohlc
if __name__ == "__main__":
logger.info("【botを起動します】")
print("【botを起動します】")
# ロジックの設定
# ------------------------------
logic = classes.Logic_test()
periods = 60
size = 0.01
# ------------------------------
# bot
flag = {'check':True, 'sell_position':False, 'buy_position':False}
result = [0]
try:
while True:
# 建玉未保有時
while flag['check']:
# ohlc取得
ohlc = getohlc(periods)
# entryの判断
entry,_ = logic.try_entry(ohlc)
if entry['buy']:
logger.info("買い注文をします 現在価格："+str(ohlc[-1][4]))
print("買い注文をします 現在価格："+str(ohlc[-1][4]))
# 注文
sendchildorder("BUY",size)
# flagの更新
flag['check'] = False
flag['buy_position'] = True
time.sleep(periods)
break
if entry['sell']:
logger.info("売り注文をします 現在価格："+str(ohlc[-1][4]))
print("売り注文をします 現在価格："+str(ohlc[-1][4]))
# 注文
sendchildorder("SELL",size)
# flagの更新
flag['check'] = False
flag['sell_position'] = True
time.sleep(periods)
break
time.sleep(periods//5)
# 買いポジション保有時
while flag['buy_position']:
# ohlc取得
ohlc = getohlc(periods)
# exitの判断
exit = logic.buy_exit(ohlc)
if exit['settle']:
logger.info("売り決済をします 損益："+str(exit['result']))
print("売り決済をします 損益："+str(exit['result']))
result.append(exit['result'])
# 決済注文
sendchildorder("SELL",size)
# flagの更新
flag['check'] = True
flag['buy_position'] = False
break
time.sleep(periods//5)
# 売りポジション保有時
while flag['sell_position']:
# ohlc取得
ohlc = getohlc(periods)
# exitの判断
exit = logic.sell_exit(ohlc)
if exit['settle']:
logger.info("買い決済をします 損益："+str(exit['result']))
print("買い決済をします 損益："+str(exit['result']))
result.append(exit['result'])
#決済注文
sendchildorder("BUY",size)
# flagの更新
flag['check'] = True
flag['sell_position'] = False
break
time.sleep(periods//5)
"""
# 終了条件（あれば）
if hogehoge:
break
"""
except KeyboardInterrupt:
logger.info("【botを終了します】\n")
print("【botを終了します】\n")
# 描画
plt.plot(range(len(result)),result.cumsum())
plt.title("Result")
plt.xlabel("Number of entries made")
plt.ylabel(f"Profit (×{size} JPY)")
plt.savefig("result.png")
loggerを利用しているので少し複雑に見えるかもしれませんが、構造自体は単純です。 もしわかりづらければlogger関連の部分を全て消しても本質的には問題ありません。 また、実際に注文をする部分については、bitFlyerの注文拒否対策として新たに関数を定義し、
Python
def sendchildorder(side,size):
cnt = 0
while cnt < 20:
response = api.sendchildorder(product_code="FX_BTC_JPY",child_order_type="MARKET",side=side,size=size)
try:
if bool(response['child_order_acceptance_id']):
break
except:
pass
cnt += 1
time.sleep(2)
という書き方をしています。 これは response['child_order_acceptance_id'] にIDがちゃんと入っていることが確認できるまで注文をトライし、20回連続で失敗したら諦めてループを抜けるという処理です。 bitFlyerはサーバーが重くなると注文の遅延や拒否が頻繁に起こるのでbotがその度エラーで止まってしまわないようにいろいろと工夫が必要です。 例外処理やメール通知などを実装して、対応できるようにしておかないと、痛い目を見ることがあるので注意してください。 実際自分が取引しているときにはそのほかにも二重注文や注文消失など様々な問題が起こったので、これだけの対策では不十分かもしれませんし、注文拒否対策についてももっと良い書き方があるかもしれません。
ローカルの環境で動かすとパソコンの電源が落ちたり通信が途切れるたびに止まってしまうので、AWSなどのクラウドサーバー上で動かすことになると思います。
今回示したコードだと、classes.pyとbot.pyを同じ階層に作って、自分のAPIをbot.py内の *** に入れ、bot.pyを実行することになります。
うまく実装できていればbot.logという名前のログファイルとコンソールに「売り注文をします 現在価格：992048.0」のようなログが記録されていきます。
さて、今回はここまでにします。
今回示したコードはシンプルなbotの構成アイデアが伝わるように書いたつもりです。
そのため、実際に動かす際に直面しうる様々な例外的な状況への対応は足りていないかもしれません。
また、今回仮設定しているロジックも非常に単純なものなので到底利益を生み出せるものではないと考えられます。
では、利益を生み出せるような戦略はどのように探していけば良いのか？ということで、最終回の次回はbotのバックテストと応用的な自動取引の概観について書いていきたいと思います。
トラブルを防ぐために最後にいくつか注意書きです。
一つ目。
本連載のコードをコピペして改変せずにそのまま使うことは絶対にしないでください。また、もしそのような利用をしたことにより使用者が不利益を被っても執筆者である私は一切の責任を負いません。
二つ目。
万が一、本連載のコードに何か誤りがあって、それにより使用者が不利益を被っても、執筆者である私は一切の責任を負いません。
三つ目。
本連載のコードを無断で転載、再配布することは禁止します。
以上、よろしくお願いします。
要はコードを参考にするのも、自動取引を行うのも、全て自己責任でお願いしますということです。
今回のまとめ
参考
特になし