仮想通貨bot　勉強記録⑫

残業パラダイスでなかなか更新できなかったぜ。

～赤三兵ロジックの動作確認～

◆前回までのあらすじ

Bybitで動く赤三兵のロジックを作成しました。

import pybybit
from datetime import datetime,timedelta
import time
from rich import print as pp

def get_price(min):#最新ローソク足を取得
   t = datetime.now() - timedelta(hours=1)#今から1時間前の時刻を取得
   t = int(t.timestamp())#タイムスタンプ形式に変換&float型からint型に変換
   d = bybit.rest.inverse.public_kline_list(
         symbol = "BTCUSD",
         interval=min,
         from_=t).json()
   last_data = d['result'][-2]

   #返り値取得
   return { "close_time"  : last_data['open_time'],
            "open_price"  : last_data['open'],
            "high_price"  : last_data['high'],
            "low_price"   : last_data['low'],
            "close_price" : last_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):#ローソク足の条件判別
       #実体割合の算出
       realbody_rate = abs(float(data["close_price"]) - float(data["open_price"])) / float(data["high_price"])-float(data["low_price"])
       #実体の大きさの算出
       increase_rate = float(data["close_price"]) / float(data["open_price"]) - 1

       if data["close_price"] < data["open_price"] : return False#ローソク足が赤だったらFalse
       elif increase_rate < 0.0005 : return False#実体の大きさが0.05%未満ならFalse
       elif realbody_rate < 0.5 : return False#実体の割合がローソク足の50%未満なら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:#それ以外はFalse
       return False

def main():
   last_data = get_price(1)
   print_price( last_data )
   flag = 0

   while True:
       data = get_price(1)

       if data["close_time"] != last_data["close_time"]:
           print_price( data )

           if flag == 0 and check_candle( data ):
               flag = 1
           elif flag == 1 and check_candle( data )  and check_ascend( data,last_data ):
               print("２本連続で陽線")
               flag = 2
           elif flag == 2 and check_candle( data )  and check_ascend( data,last_data ):
               print("３本連続で陽線 なので 買い！")
               flag = 3
           else:
              flag = 0

           last_data["close_time"] = data["close_time"]
           last_data["open_price"] = data["open_price"]
           last_data["close_price"] = data["close_price"]

           time.sleep(10)

main()

このコード動くんですけど、間違ってました。。。

def check_candle(data):#ローソク足の条件判別
       #実体割合の算出
       realbody_rate = abs(float(data["close_price"]) - float(data["open_price"])) / float(data["high_price"])-float(data["low_price"])
       #実体の大きさの算出
       increase_rate = float(data["close_price"]) / float(data["open_price"]) - 1

       if data["close_price"] < data["open_price"] : return False #ローソク足が赤だったらFalse
       elif increase_rate < 0.0005 : return False #実体の大きさが0.05%未満ならFalse
       elif realbody_rate < 0.5 : return False #実体の割合がローソク足の50%未満ならFalse
       else : return True #上記すべて条件が当てはまらなければTrue

この関数の部分で、Falseの後にスペースを入れずに#でコメントを入れてたんですけど、このせいでFalseが機能してませんでした。

◆今回やること

・過去の価格データで売買シグナルを検証する

過去のCryptowatchの価格データで売買シグナルを検証をする | 文系でもわかる！BitcoinのBOT自動売買トレードの始め方

こちらを参考に、過去データで作成したシグナルの検証を行います。
いわゆるバックテストというやつです。

作ったコードはこちら。

from datetime import datetime,timedelta
import time
from rich import print as pp
import pybybit

apis = [
  'プライベートキー',
  'シークレットキー'
]

bybit = pybybit.API(*apis, testnet=True)

d = bybit.rest.inverse.public_kline_list(
     symbol = "BTCUSD",
     interval=1,
     from_=1616293214)

def get_price(i):#最新ローソク足を取得

   data = d.json()
   last_data = data['result'][i]

   #返り値取得
   return { "close_time"  : last_data['open_time'],
       "open_price"  : float(last_data['open']),
       "high_price"  : float(last_data['high']),
       "low_price"   : float(last_data['low']),
       "close_price" : float(last_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):#ローソク足の条件判別
       #実体割合の算出
       realbody_rate = abs(data["close_price"] - data["open_price"]) / (data["high_price"]-data["low_price"])
       #実体の大きさの算出
       increase_rate = (data["close_price"] / data["open_price"]) - 1

       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:#それ以外はFalse
       return False

def main():
   last_data = get_price(0)
   print_price( last_data )

   flag = 0
   i = 1

   while i < 50:
       data = get_price(i)

       if data["close_time"] != last_data["close_time"]:
           print_price( data )

           if flag == 0 and check_candle( data ):
               flag = 1
           elif flag == 1 and check_candle( data )  and check_ascend( data,last_data ):
               pp("２本連続で陽線")
               flag = 2
           elif flag == 2 and check_candle( data )  and check_ascend( data,last_data ):
               pp("３本連続で陽線 なので 買い！")
               flag = 3
           else:
              flag = 0

           last_data["close_time"] = data["close_time"]
           last_data["open_price"] = data["open_price"]
           last_data["close_price"] = data["close_price"]

           i += 1
           time.sleep(0.01)

main()

前回のコードとほとんど変わらんです。
サクッと解説していきます。

◆解説

・変化点①

bybit = pybybit.API(*apis, testnet=True)

d = bybit.rest.inverse.public_kline_list(
     symbol = "BTCUSD",
     interval=1,
     from_=1616293214)

最初の関数より前の部分です。
バックテストでは一度ローソク足のデータを取得したら、そのデータでテストを行うので、取得コードを関数の外に出しています。

・変化点②

def get_price(i):#最新ローソク足を取得

   data = d.json()
   last_data = data['result'][i]

   #返り値取得
   return { "close_time"  : last_data['open_time'],
            "open_price"  : float(last_data['open']),
            "high_price"  : float(last_data['high']),
            "low_price"   : float(last_data['low']),
            "close_price" : float(last_data['close'])}

価格データを取得する関数の部分です。
関数の引数にiを設定して、取得したデータのi番目を呼び出せるようにしてあります。

・変化点③

 i = 1

   while i < 50:
       data = get_price(i)

       if data["close_time"] != last_data["close_time"]:
           print_price( data )

           if flag == 0 and check_candle( data ):
               flag = 1
           elif flag == 1 and check_candle( data )  and check_ascend( data,last_data ):
               pp("２本連続で陽線")
               flag = 2
           elif flag == 2 and check_candle( data )  and check_ascend( data,last_data ):
               pp("３本連続で陽線 なので 買い！")
               flag = 3
           else:
              flag = 0

           last_data["close_time"] = data["close_time"]
           last_data["open_price"] = data["open_price"]
           last_data["close_price"] = data["close_price"]

           i += 1
           time.sleep(0.01)

while文の前でi=1を代入し、while文で"iが50に到達するまで”の条件でループさせています。
最後から2行目の部分に　i += 1　とありますが、iに1ずつ加算させるコードです。
このwhile文では、iが1から50になるまで0.01秒ごとに処理を繰り返します。

変化点は以上です。ちなみに、bybitのローソク足情報の取得は最大200本なので、iは最大200です。

実行結果はこんな感じ。

簡単に過去データでテストができます！

バックテストができるようになりました。
今回はここまで。