
仮想通貨bot 勉強記録⑱
~jsonファイルのデータでバックテストをする~
◆前回までのあらすじ
1回実行するだけで大量にローソク足データを入手できるコードを作成しました。最大でどれだけ取得できるか分かりませんが、1分足データを1か月分(43,200本)は取得できました。
◆今回やること
・作成したjsonファイルをもとにバックテストを行う
前回過去データを大量に取得できたので、そのデータをもとにバックテストをやってみます。
作成したコードはこちら
from datetime import datetime
from rich import print as pp
import pybybit
import json
"""------------------------"""
apis = [
'プライベートキー',
'シークレットキー'
]
bybit = pybybit.API(*apis, testnet=True)
"""------------------------"""
file = open("./ここにファイル名を書き込む","r",encoding="utf-8")
response = json.load(file)
#ローソク足情報を返す関数
def get_price(i):
data = response[i]
#返り値取得
return { "close_time" : data['open_time'],
"open_price" : float(data['open']),
"high_price" : float(data['high']),
"low_price" : float(data['low']),
"close_price" : float(data['close'])}
#画面出力
def print_price(data):
pp( " 時間: " + datetime.fromtimestamp(data['close_time']).strftime('%Y/%m/%d %H:%M')
+ " 始値: " + str(data['open_price'])
+ " 終値: " + str(data['close_price']))
#ローソク足の条件判別
def check_candle(data,side):
try:
#実体割合の算出
realbody_rate = abs(data["close_price"] - data["open_price"]) / (data["high_price"]-data["low_price"])
except ZeroDivisionError:
increase_rate = 0
try:
#実体の大きさの算出
increase_rate = (data["close_price"] / data["open_price"]) - 1
except ZeroDivisionError:
increase_rate = 0
if side == "buy":
if data["close_price"] < data["open_price"] : return False #ローソク足が赤だったらFalse
elif increase_rate < 0.0001 : return False #実体の大きさが現在価格の0.01%未満ならFalse
elif realbody_rate < 0.1 : return False #実体の割合がローソク足の10%未満ならFalse
else : return True #上記すべて条件が当てはまらなければTrue
if side == "sell":
if data["close_price"] > data["open_price"] : return False #ローソク足が緑だったらFalse
elif increase_rate > -0.0001 : return False #実体の大きさが現在価格の0.01%未満ならFalse
elif realbody_rate < 0.1 : return False #実体の割合がローソク足の10%未満ならFalse
else : return True #上記すべて条件が当てはまらなければTrue
#ローソク足の連続上昇の判別
def check_ascend( data,last_data ):
#今回の始値が前回の始値を上回っている且つ今回の終値が前回の終値を上回っていればTure
if data["open_price"] > last_data["open_price"] and data["close_price"] > last_data["close_price"]:
return True
else: return False
#ローソク足の連続下降の判別
def check_descend( data,last_data ):
#今回の始値が前回の始値を下回っている且つ今回の終値が前回の終値を下回っていればTure
if data["open_price"] < last_data["open_price"] and data["close_price"] < last_data["close_price"]:
return True
else:
return False
#買いサイン・注文を出す関数
def buy_signal( data,last_data,flag ):
if flag["buy_signal"] == 0 and check_candle( data,"buy" ):#陽線1本目
flag["buy_signal"] = 1
elif flag["buy_signal"] == 1 and check_candle( data,"buy" ) and check_ascend( data,last_data ):#陽線2本目
flag["buy_signal"] = 2
elif flag["buy_signal"] == 2 and check_candle( data,"buy" ) and check_ascend( data,last_data ):#陽線3本目
pp("3本連続で陽線 なので" + str(data["close_price"]) + "で買い指値")
flag["buy_signal"] = 3
"""指値買い注文コード"""
flag["order"]["exist"] = True
flag["order"]["side"] = "BUY"
#陽線が途切れたらシグナルリセット
else:
flag["buy_signal"] = 0
return flag
#売りサイン・注文を出す関数
def sell_signal( data,last_data,flag ):
if flag["sell_signal"] == 0 and check_candle( data,"sell" ):#陰線1本目
flag["sell_signal"] = 1
elif flag["sell_signal"] == 1 and check_candle( data,"sell" ) and check_descend( data,last_data ):#陰線2本目
flag["sell_signal"] = 2
elif flag["sell_signal"] == 2 and check_candle( data,"sell" ) and check_descend( data,last_data ):#陰線3本目
pp("3本連続で陰線 なので" + str(data["close_price"]) + "で売り指値")
"""指値売り注文コード"""
flag["order"]["exist"] = True
flag["order"]["side"] = "SELL"
#陰線が途切れたらシグナルリセット
else:
flag["sell_signal"] = 0
return flag
#決済用の関数
def close_position( data,last_data,flag ):
if flag["position"]["side"] == "BUY":
if data["close_price"] < last_data["close_price"]:
pp("前回の終値を下回ったので" + str(data["close_price"]) + "あたりで成行で決済します")
"""成行売り注文コード"""
flag["position"]["exist"] = False
if flag["position"]["side"] == "SELL":
if data["close_price"] > last_data["close_price"]:
pp("前回の終値を上回ったので" + str(data["close_price"]) + "あたりで成行で決済します")
"""成行買い注文コード"""
flag["position"]["exist"] = False
return flag
#注文状況確認用の関数
def check_order( flag ):
"""注文状況確認コード"""
flag["order"]["exist"] = False
flag["order"]["count"] = 0
flag["position"]["exist"] = True
flag["position"]["side"] = flag["order"]["side"]
return flag
#メイン関数
def main():
last_data = get_price(0)
print_price( last_data )
flag = {
"buy_signal":0,
"sell_signal":0,
"order":{
"exist" : False,
"side" : "",
"count" : 0},
"position":{
"exist" : False,
"side" : ""}
}
i = 0
while i< 60*24*3: #テストを行うローソク足の本数
if flag["order"]["exist"]:
flag = check_order( flag )
data = get_price(i)
print_price( data )
if flag["position"]["exist"]:
flag = close_position( data,last_data,flag )
else:
flag = buy_signal( data,last_data,flag )
flag = sell_signal( data,last_data,flag )
last_data["close_time"] = data["close_time"]
last_data["open_price"] = data["open_price"]
last_data["close_price"] = data["close_price"]
i+=1
main()
◆解説
基本的には勉強記録⑯で作成したテストコードと同じです。
なので変化点だけを説明します。
・変化点①:import json
from datetime import datetime
from rich import print as pp
import pybybit
import json
保存したjsonファイルを開くためのモジュール(json)をインポートしておきます。
・変化点②:ファイルを読み込む部分
file = open("./ここにファイル名を書き込む","r",encoding="utf-8")
response = json.load(file)
ファイルを読み込む部分です。open("ファイルの場所","読み取り形式","文字形式")の順で入力します。
・変化点③:def get_price(i)
def get_price(i):
data = response[i]
以前は関数名がget_price(min,i)でしたが、m(interval)はファイルに保存されているデータしか使えないので、指定しません。
また、中身もdata = response['result'][i]でしたが、保存してあるデータは最初からリスト形式になっているので、['result']は消しておきます。
・変化点④:def main()
last_data = get_price(0)
関数の引数に合わせて、(0)のみにしています。
while i< 60*24*3: #テストを行うローソク足の本数
ここが目玉ポイントです!今までは最大200までしか設定できなかったけど、好きなだけテストを行えます。
今回は3日分の1分足データでテストを行うので、60(分)*24(時間)*10(日)回で指定しました。
以上です!
◆実行方法
早速実行したいところですが、少し準備が必要なんですたい。
・Pydocフォルダ内に、新しくフォルダを作る
Pydocフォルダ内に、backtestというフォルダを作ります。
作ったら、その中に勉強記録⑰で作成したコード(get-backdata.py)と、今回作成したコード(backtest.py)を入れ、get-backdata.pyを実行します。
こんな感じになるはず。
・Anaconda3でbacktestディレクトリに移動する
Anaconda3を起動したら、
cd backtest
と入力します。
これでAnaconda3がbacktest.pyを実行できるようになり、準備完了です。
◆実行結果
ずら~~~~っとデータが流れて、3日分のテストが完了しました。
3日分だと24秒くらいで実行終了します。
無事に保存したjsonファイルを使ってバックテストができるようになったので、今回はここまで。