仮想通貨bot　勉強記録⑳

もう第20回になってしもうた。。。

～ドンチャンチャネルのロジックを作ってみる～

◆前回までのあらすじ

テストコードで損益を計算したら、勝率の低いロジックなのに損益がプラスになってしまいました。

なんでやねん。

※一応解決？しました。

原因はおそらく手数料がちゃんと計算できていなかったことだと思います。

#====================トレードパフォーマンス確認====================
def records(flag,data):

  entry_price = flag["position"]["price"]
  exit_price = round(float(data["close"]))
  trade_cost  = round(lot/float(data["close"])* slippage)

上記のtrade_costの計算及び表示方法をミスってました。
正しくは↓

trade_cost  = lot * slippage

これだけです。exit価格(float(data["close"]))を掛け忘れ&roundで四捨五入することで手数料が0$になってました。

これにより、総損益はプラスですが手数料を加味すると損益はマイナスになりました。（損益がマイナスになって喜ぶの複雑なんだぜ。。。）

◆今回やること

・新しいロジックでテストコードをつくる

これまで赤三兵・黒三兵のロジックでテストを行ってきましたが、実践に移る前に他のロジックでもテストをしてみます。
つまり、もうちょっと勉強しておきます。

今回も我らが教科書のこちらの記事を参考にします。

BTCFXのドンチャンチャネルブレイクアウトの勝率をバックテストで検証する | 文系でもわかる！BitcoinのBOT自動売買トレードの始め方

作成コードはこちら！

from datetime import datetime
from rich import print as pp
import pybybit
import time
import json

#====================API設定====================
apis = [
'プライベートキー',
'シークレットキー'
]
bybit = pybybit.API(*apis, testnet=True)
#===============================================

#====================バックテスト用の初期設定値====================
lot        = 100                                  # 1トレードのロット
slippage   = 0.00075                              # 手数料やスリッページ（0.075％初期値）
chart_min  = 5                                    # 5分足
test_start = 0                                    # 何番目のデータからテストするか
test_end   = 60*24*1/chart_min                    # 何番目のデータまでテストするか
path       = "./1614524400-1617212340-price.json" # 読み込むファイルのパス
term       = 20                                   # 過去n期間
wait       = 0                                    # 待機時間


#====================APIから価格データ取得====================
def get_price_from_API(chart_min):

   #取得開始時刻の設定
   get_start = {
       "year"   : int(2021), #年
       "month"  : int(3),    #月
       "day"    : int(30),    #日
       "hour"   : int(00),   #時
       "minute" : int(00)    #分
             }

   get_start = int(datetime(**get_start).timestamp())

   #取得終了時刻の設定
   get_end = {
       "year"   : int(2021), #年
       "month"  : int(4),    #月
       "day"    : int(1),    #日
       "hour"   : int(00),   #時
       "minute" : int(00)    #分
           }

   get_end = int(datetime(**get_end).timestamp())

   #データを格納する変数
   price = []

   #tがnowより小さければ実行
   while get_start < get_end:

      #pybybitでローソク足取得
      k = bybit.rest.inverse.public_kline_list(
             symbol = "BTCUSD",
             interval= chart_min,
             from_ = get_start
             ).json()

      #klinesに取得したデータを入れる
      price += k["result"]
      #200本x足の長さ分だけタイムスタンプを進める
      get_start += 200*60*chart_min

   return price

#====================ファイルから価格データを読み込む====================
def get_price_from_file(path):
   file = open(path,'r',encoding='utf-8')
   price = json.load(file)

   return price
#====================画面出力関====================
def print_price(data):
   pp( " 時間： " + datetime.fromtimestamp(data['open_time']).strftime('%Y/%m/%d %H:%M')
     + " 始値： " + str(data['open'])
     + " 終値： " + str(data['close']))


#====================ロジック判定====================
def donchian( data,last_data ):

   highest = max(i["high"] for i in last_data)
   if data["high"] > highest:
       return {"side":"BUY","price":highest}

   lowest = min(i["low"] for i in last_data)
   if data["low"] < lowest:
       return {"side":"SELL","price":lowest}

   return {"side" : None , "price":0}


#====================買い・売り注文====================
def entry_signal( data,last_data,flag ):

   signal = donchian( data,last_data )

   if signal["side"] == "BUY":
       pp("過去{0}足の最高値{1}円を、直近の高値が{2}円でブレイク".format(term,signal["price"],data["high"]))
       pp(str(data["close"]) + "円で買い指値")

       # ここに買い注文のコードを入れる

       flag["order"]["exist"] = True
       flag["order"]["side"] = "BUY"
       flag["order"]["price"] = data["close"]

   if signal["side"] == "SELL":
       pp("過去{0}足の最安値{1}円を、直近の安値が{2}円でブレイク".format(term,signal["price"],data["low_price"]))
       pp(str(data["close"]) + "円で売り指値")

       # ここに売り注文のコードを入れる

       flag["order"]["exist"] = True
       flag["order"]["side"] = "SELL"
       flag["order"]["price"] = data["close"]

   return flag


#====================注文状況確認====================
def check_order( flag ):

   """ここに注文状況確認コード"""

   flag["order"]["exist"] = False
   flag["position"]["exist"] = True
   flag["position"]["side"] = flag["order"]["side"]
   flag["position"]["price"] = flag["order"]["price"]

   return flag


#====================成行決済&ドテン注文====================
def close_position( data,last_data,flag ):

   flag["position"]["count"] += 1

   signal = donchian( data,last_data )

   if flag["position"]["side"] == "BUY":
       if signal["side"] == "SELL":
           pp("過去{0}足の最安値{1}円を、直近の安値が{2}円でブレイク".format(term,signal["price"],data["low"]))
           pp("成行注文を出してポジションを決済します")

           # 決済の成行注文コードを入れる

           flag["position"]["exist"] = False
           flag["position"]["count"] = 0

           pp("さらに" + str(data["close"]) + "円で売りの指値注文を入れてドテンします")

           # ここに売り注文のコードを入れる

           flag["order"]["exist"] = True
           flag["order"]["side"] = "SELL"

   if flag["position"]["side"] == "SELL":
       if signal["side"] == "BUY":
           pp("過去{0}足の最高値{1}円を、直近の高値が{2}円でブレイク".format(term,signal["price"],data["high"]))
           pp("成行注文を出してポジションを決済します")

           # 決済の成行注文コードを入れる

           flag["position"]["exist"] = False
           flag["position"]["count"] = 0

           pp("さらに" + str(data["close"]) + "円で買いの指値注文を入れてドテンします")

           # ここに買い注文のコードを入れる

           flag["order"]["exist"] = True
           flag["order"]["side"] = "BUY"

   return flag


#====================メイン====================
def main():
   price = get_price_from_API(chart_min)
   last_data = []

   flag = {
       "order":{
           "exist" : False,
           "side"  : "",
           "count" : 0
       },
       "position":{
           "exist" : False,
           "side"  : "",
           "count" : 0
       }
   }

   i = 0
   while i < test_end:

       # ドンチャンの判定に使う過去term足分の安値・高値データを準備する
       if len(last_data) < term:
           last_data.append(price[i])
           print_price(price[i])
           time.sleep(wait)
           i += 1

           #if文の条件を満たすまでループ処理
           continue

       data = price[i]
       print_price(data)

       if flag["order"]["exist"]:
           flag = check_order( flag )
       elif flag["position"]["exist"]:
           flag = close_position( data,last_data,flag )
       else:
           flag = entry_signal( data,last_data,flag )


       # 過去データをterm個ピッタリに保つために先頭を削除
       del last_data[0]

       last_data.append( data )
       i += 1
       time.sleep(wait)

main()

今回は損益計算まではやらないですぞ。

◆解説

・初期値設定

#====================バックテスト用の初期設定値====================
lot        = 100                                  # 1トレードのロット
slippage   = 0.00075                              # 手数料やスリッページ（0.075％初期値）
chart_min  = 5                                    # 5分足
test_start = 0                                    # 何番目のデータからテストするか
test_end   = 60*24*1/chart_min                    # 何番目のデータまでテストするか
path       = "./1614524400-1617212340-price.json" # 読み込むファイルのパス
term       = 20                                   # 過去n期間
wait       = 0                                    # 待機時間

内容はコメントアウトで書いてますが、ロットやスリッパゲは今回は使わないです。スリッパゲ！

・def get_price_from_API(chart_min)

勉強記録⑰で作ったコードをいじって、関数化したものです。
入手データをjsonファイルに保存せず、そのまま使います。

・def get_price_from_file(path)

#====================ファイルから価格データを読み込む====================
def get_price_from_file(path):
   file = open(path,'r',encoding='utf-8')
   price = json.load(file)

   return price

役割が上の関数と被ってますが、API・jsonファイルのどちらからでもローソク足情報を入手できるようにしてあります。
メイン関数でget_priceのどちらかの関数を指定します。

・def donchian( data,last_data )

#====================ロジック判定====================
def donchian( data,last_data ):

   highest = max(i["high"] for i in last_data)
   if data["high"] > highest:
       return {"side":"BUY","price":highest}

   lowest = min(i["low"] for i in last_data)
   if data["low"] < lowest:
       return {"side":"SELL","price":lowest}

   return {"side" : None , "price":0}

ドンチャンブレイクのロジックを実行する関数です。
高値ブレイク時のロジックのみ説明します。

highest = max(i["high"] for i in last_data)
if data["high"] > highest:
    　return {"side":"BUY","price":highest}

highest = max(i["high"] for i in last_data)
変数：highestに、last_dataの["high"]の中から最も大きい値を入れます。

if data["high"] > highest:
　return {"side":"BUY","price":highest}
data["high"] が highest を上回ったら、"side":"BUY","price":highestを返します。

返り値は注文を出す関数で使用します。

・def entry_signal( data,last_data,flag )

#====================買い・売り注文====================
def entry_signal( data,last_data,flag ):

   signal = donchian( data,last_data )

   if signal["side"] == "BUY":
       pp("過去{0}足の最高値{1}円を、直近の高値が{2}円でブレイク".format(term,signal["price"],data["high"]))
       pp(str(data["close"]) + "円で買い指値")

       # ここに買い注文のコードを入れる

       flag["order"]["exist"] = True
       flag["order"]["side"] = "BUY"
       flag["order"]["price"] = data["close"]

   if signal["side"] == "SELL":
       pp("過去{0}足の最安値{1}円を、直近の安値が{2}円でブレイク".format(term,signal["price"],data["low_price"]))
       pp(str(data["close"]) + "円で売り指値")

       # ここに売り注文のコードを入れる

       flag["order"]["exist"] = True
       flag["order"]["side"] = "SELL"
       flag["order"]["price"] = data["close"]

   return flag

買い・売りの注文を出す関数です。ただし、この関数を使うのは初回の注文のみです。
※2回目以降の注文はドテンになるので、決済用の関数で同時に注文も行います。

signal = donchian( data,last_data )

まず変数：signalにdonchian( data,last_data )からの返り値（ {"side":"BUY","price":highest}）を入れておきます。

if signal["side"] == "BUY":
    pp("過去{0}足の最高値{1}円を、直近の高値が{2}円でブレイク".format(term,signal["price"],data["high"]))
    pp(str(data["close"]) + "円で買い指値")

    # ここに買い注文のコードを入れる

    flag["order"]["exist"] = True
    flag["order"]["side"] = "BUY"
    flag["order"]["price"] = data["close"]

signal["side"] が "BUY"のとき、高値のブレイク状況・買い値の値を画面表示します。
買い注文実行後、flag["order"]内の各値を書き換えます。

"Sell"の時も同じ

・def check_order( flag )

#====================注文状況確認====================
def check_order( flag ):

   """ここに注文状況確認コード"""

   flag["order"]["exist"] = False
   flag["position"]["exist"] = True
   flag["position"]["side"] = flag["order"]["side"]
   flag["position"]["price"] = flag["order"]["price"]

   return flag

前回までと同じです。注文状況を確認し、約定したらflagの各値を書き換えます。

・def close_position( data,last_data,flag )

#====================成行決済&ドテン注文====================
def close_position( data,last_data,flag ):

   flag["position"]["count"] += 1
   signal = donchian( data,last_data )

   if flag["position"]["side"] == "BUY":
       if signal["side"] == "SELL":
           pp("過去{0}足の最安値{1}円を、直近の安値が{2}円でブレイク".format(term,signal["price"],data["low"]))
           pp("成行注文を出してポジションを決済します")

           # 決済の成行注文コードを入れる

           flag["position"]["exist"] = False
           flag["position"]["count"] = 0

           pp("さらに" + str(data["close"]) + "円で売りの指値注文を入れてドテンします")

           # ここに売り注文のコードを入れる

           flag["order"]["exist"] = True
           flag["order"]["side"] = "SELL"

   if flag["position"]["side"] == "SELL":
       if signal["side"] == "BUY":
           pp("過去{0}足の最高値{1}円を、直近の高値が{2}円でブレイク".format(term,signal["price"],data["high"]))
           pp("成行注文を出してポジションを決済します")

           # 決済の成行注文コードを入れる

           flag["position"]["exist"] = False
           flag["position"]["count"] = 0

           pp("さらに" + str(data["close"]) + "円で買いの指値注文を入れてドテンします")

           # ここに買い注文のコードを入れる

           flag["order"]["exist"] = True
           flag["order"]["side"] = "BUY"

   return flag

決済&ドテン注文を行う関数です。

 flag["position"]["count"] += 1
 signal = donchian( data,last_data )

最初にポジションのカウントと、donchian( data,last_data )からの返り値を設定しておきます。

if flag["position"]["side"] == "BUY":
   if signal["side"] == "SELL":
       pp("過去{0}足の最安値{1}円を、直近の安値が{2}円でブレイク".format(term,signal["price"],data["low"]))
       pp("成行注文を出してポジションを決済します")

       # 決済の成行注文コードを入れる

       flag["position"]["exist"] = False
       flag["position"]["count"] = 0

       pp("さらに" + str(data["close"]) + "円で売りの指値注文を入れてドテンします")

       # ここに売り注文のコードを入れる

       flag["order"]["exist"] = True
       flag["order"]["side"] = "SELL"

flag["position"]["side"]が"BUY"で、signal["side"] が "SELL"の時、安値のブレイク状況・売り注文の表示をし、売り注文を行います。

その後、falg["position"]内の各値を書き換え、ドテン注文を行い、flag["order"]内の各値を書き換えます。

if flag["position"]["side"] == "SELL"の時も同様です。

・main()

#====================メイン====================
def main():
   price = get_price(chart_min)
   
   last_data = []

   flag = {
       "order":{
           "exist" : False,
           "side"  : "",
           "count" : 0
       },
       "position":{
           "exist" : False,
           "side"  : "",
           "count" : 0
       }
   }

   i = 0
   while i < end:

       # ドンチャンの判定に使う過去ｎ足分の安値・高値データを準備する
       if len(last_data) < term:
           last_data.append(price[i])
           print_price(price[i])
           time.sleep(wait)
           i += 1

           #if文の条件を満たすまでループ処理
           continue

       data = price[i]
       print_price(data)

       if flag["order"]["exist"]:
           flag = check_order( flag )
       elif flag["position"]["exist"]:
           flag = close_position( data,last_data,flag )
       else:
           flag = entry_signal( data,last_data,flag )


       # 過去データをｎ個ピッタリに保つために先頭を削除
       del last_data[0]

       last_data.append( data )
       i += 1
       time.sleep(wait)

メイン関数です。

price = get_price(chart_min)
last_data = []

flag = {
   "order":{
       "exist" : False,
       "side"  : "",
       "count" : 0
   },
   "position":{
       "exist" : False,
       "side"  : "",
       "count" : 0
   }
}
i = 0

まず初期値の設定をします。ローソク足データを変数：priceに、last_dataをリストを入れる変数として設定し、flagの初期値も与えておきます。
while文用のiも0にしておきます。

while i < end:

   # ドンチャンの判定に使う過去term足分の安値・高値データを準備する
   if len(last_data) < term:
       last_data.append(price[i])
       print_price(price[i])
       time.sleep(wait)
       i += 1

       #if文の条件を満たすまでループ処理
       continue

iが初期設定のendより小さい場合、while文を実行します。

last_data内のデータの個数（len(last_data)）がtermより小さい場合、last_dataにprice[i]を追加し、画面表示し、iを+1します。
最後のcontinueでwhile文の最初の行に戻り、if len(last_data) < termを満たすまで実行します。（今回の場合20回）

data = price[i]
print_price(data)

if flag["order"]["exist"]:
   flag = check_order( flag )
elif flag["position"]["exist"]:
   flag = close_position( data,last_data,flag )
else:
   flag = entry_signal( data,last_data,flag )

while文から抜け出したら、data に price[i]を入れ、画面表示します。

①flag["order"]["exist"]がTrueの場合、check_order( flag )を呼び出し、返り値をflagに入れます。

②flag["position"]["exist"]がTrueの場合、close_position( data,last_data,flag )を呼び出し、返り値をflagに入れます。

③上記以外の場合、entry_signal( data,last_data,flag )を呼び出し、返り値をflagに入れます。

# 過去データをterm個に保つために先頭を削除
del last_data[0]

last_data.append( data )
i += 1
time.sleep(wait)

del last_data[0]で、last_data内の最初のデータを削除し、
last_data.append( data )でlast_dataにdataを追加します。
最後にiを+1して終了です！

解説終わり！

◆実行結果

実行するとこんな感じです。
コピペして実行するだけで動くので、やってみてくださいッ！

次回はこのコードで損益計算をやってみます。