仮想通貨bot　勉強記録⑬

～赤三兵ロジック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()

◆今回やること

・注文と決済を自動化するテストコードを作る

BTCFXのエントリーと手仕舞い（決済）をpythonで自動化しよう | 文系でもわかる！BitcoinのBOT自動売買トレードの始め方

こちらを参考に、ロジックを確認⇒エントリー⇒決済　の流れを自動化するためのテストコードを作ります。いよいよbotっぽくなってきましたね！

作ったコードはこちら

from datetime import datetime
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_=1616684400) #2021/03/26 00:00:00

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

   data = d.json()
   data = data['result'][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): #ローソク足の条件判別
   #実体割合の算出
   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.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: return False

def buy_signal( data,last_data,flag ): #買いサインを出す関数
   if flag["buy_signal"] == 0 and check_candle( data ):#陽線1本目
       flag["buy_signal"] = 1
   elif flag["buy_signal"] == 1 and check_candle( data )  and check_ascend( data,last_data ):#陽線2本目
       flag["buy_signal"] = 2
   elif flag["buy_signal"] == 2 and check_candle( data )  and check_ascend( data,last_data ):#陽線3本目
       print("３本連続で陽線 なので" + str(data["close_price"]) + "で買い指値")
       # ここに買い注文のコードを入れる(今回は確認用なので入れない)

       flag["buy_signal"] = 3
       flag["order"] = True #未約定の注文があるサインを出す

   else:#陽線が途切れたりしたらシグナルリセット
       flag["buy_signal"] = 0

   return flag

def close_position( data,last_data,flag ):#決済用の関数

   if data["close_price"] < last_data["close_price"]:#陰線が出たら決済
       print("前回の終値を下回ったので" + str(data["close_price"]) + "で決済")
       #ここに決済注文のコードを入れる
       
       flag["position"] = False #ポジションを持ってないサインを出す

   return flag

def check_order( flag ):#注文監視用の関数

   # 注文状況を確認して通っていたら以下を実行
   # 一定時間で注文が通っていなければキャンセルする

   flag["order"] = False
   flag["position"] = True
   return flag

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

   flag = {
   "buy_signal":0,
   "sell_signal":0,
   "order":False,
   "position":False}

   i = 1
   
   while i < 200:
       if flag["order"]:
           flag = check_order(flag)
           
       data = get_price(i)

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

           if flag["position"]:
               flag = close_position( data,last_data,flag )

           else:
               flag = buy_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

       time.sleep(0.005)

main()

相変わらず長いっすね～～
でもちゃんと解説するから大丈夫やで👍

◆解説

前回のバックテスト用コードからの変化点を解説します。

・def buy_signal( data,last_data,flag )

def buy_signal( data,last_data,flag ): #買いサインを出す関数
   if flag["buy_signal"] == 0 and check_candle( data ):#陽線1本目
       flag["buy_signal"] = 1
   elif flag["buy_signal"] == 1 and check_candle( data )  and check_ascend( data,last_data ):#陽線2本目
       flag["buy_signal"] = 2
   elif flag["buy_signal"] == 2 and check_candle( data )  and check_ascend( data,last_data ):#陽線3本目
       print("３本連続で陽線 なので" + str(data["close_price"]) + "で買い指値")
       # ここに買い注文のコードを入れる(今回は確認用なので入れない)

       flag["buy_signal"] = 3
       flag["order"] = True #未約定の注文があるサインを出す

   else:#陽線が途切れたりしたらシグナルリセット
       flag["buy_signal"] = 0

   return flag

買いサインを出すための関数です・赤三兵の買いサインは”陽線が3連続で出ること”なので、それを判断する部分です。
main関数内に用意しているflag配列内の"buy_signal"の値に応じて判断します。

if flag["buy_signal"] == 0 and check_candle( data ):#陽線1本目
       flag["buy_signal"] = 1

上記の部分の意味は、
”もしbuy_signalの値が0かつcheck_candle(data)がTrueだった場合、buy_signalの値を1に書き換える”
です。

check_candle関数の解説は、前々回の記事でやってます。

   elif flag["buy_signal"] == 2 and check_candle( data )  and check_ascend( data,last_data ):#陽線3本目
       print("３本連続で陽線 なので" + str(data["close_price"]) + "で買い指値")
       # ここに買い注文のコードを入れる(今回は確認用なので入れない)

       flag["buy_signal"] = 3
       flag["order"] = True #未約定の注文があるサインを出す

   else:#陽線が途切れたりしたらシグナルリセット
       flag["buy_signal"] = 0

#で各行の解説をしているのでわかると思いますが、陽線が3連続になった時点で買い注文を出し、更にflag["order"]をTrueに書き換えます。（今回は買い注文コードは入れてません）
陽線が2連続で終わったり、check_candleの条件を満たさない場合はflag["buy_signal"]を0に書き換え、シグナルをリセットします。

最後にflagの値を返します。

・def close_position( data,last_data,flag )

def close_position( data,last_data,flag ):#決済用の関数

   if data["close_price"] < last_data["close_price"]:#陰線が出たら決済
       print("前回の終値を下回ったので" + str(data["close_price"]) + "で決済")
       #ここに決済注文のコードを入れる
       
       flag["position"] = False #ポジションを持ってないサインを出す

   return flag

ポジションを持っているときに、決済の判断をするためのコードです。
今回の決済条件は、”ポジション所持中に1度でも陰線が出ること”です。

決済を行ったらポジションは解消されるので、flag["position"]をFalseに書き換えます。

最後にflagの値を返します。

・def check_order( flag )

def check_order( flag ):

   # 注文状況を確認して通っていたら以下を実行
   # 一定時間で注文が通っていなければキャンセルする

   flag["order"] = False
   flag["position"] = True
   return flag

注文の状況を確認するためのコードです。
この関数が呼び出されたとき、flag["order"]をFalseに書き換え、flag["position"]をTrueに書き換えます。
意味は”注文が約定されてポジションを所持したよ”です。

注文状況を確認するコードは書いていませんが、実際に動かすときは勉強記録⑦で扱った内容を使います。

最後にflagの値を返します。

・def main()

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

   flag = {
   "buy_signal":0,
   "sell_signal":0,
   "order":False,
   "position":False}

   i = 1
   
   while i < 200:
       if flag["order"]:
           flag = check_order(flag)
           
       data = get_price(i)

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

           if flag["position"]:
               flag = close_position( data,last_data,flag )

           else:
               flag = buy_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

       time.sleep(0.005)

いよいよメイン関数です。サブ関数を使ってロジックを動かす関数です。

   last_data = get_price(0)
   print_price( last_data )

   flag = {
   "buy_signal":0,
   "sell_signal":0,
   "order":False,
   "position":False}

   i = 1

まずはget_price関数、print_price関数で最初のローソク足の価格表示を行い、そのあと各初期値の設定を行います。

 while i < 200:
       if flag["order"]:
           flag = check_order(flag)
           
       data = get_price(i)

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

           if flag["position"]:
               flag = close_position( data,last_data,flag )

           else:
               flag = buy_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

       time.sleep(0.005)

初期値の設置を終えたら、while文でループに入ります。ループ条件は”iが200になるまで”です。
上から順に説明していきます。

if flag["order"]:
           flag = check_order(flag)
           
       data = get_price(i)

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

flag["order"]がTrueの時、flagの値にcheck_order(flag)の戻り値を入れます。
get_price(i)でdataにi番目のローソク足情報を入れます。
ローソク足情報が更新されたら、print_price( data )で表示します。

if flag["position"]:
               flag = close_position( data,last_data,flag )

           else:
               flag = buy_signal( data,last_data,flag )

flag["position"]がTrueだった場合、flagにclose_position( data,last_data,flag )の戻り値を入れます。

flag["position"]がFalseだった場合、flagにbuy_position( 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

       time.sleep(0.005)

以上の処理が終わったら、last_dataにdataの値を書き込みます（前回のデータを最新のデータに書き換える）
更にiの値を+1して、time.sleep(0.005)で0.005秒待ち、while文を再度ループします。

実行するとこんな感じになります。

いい感じ！

コードが長くなってくると解説もむずい、、、
サブ関数の意味を理解すれば、メイン関数の内容もサクッと理解できます。

今回はここまで！