Bot

仮想通貨botの開発を本格的に始めてみる#22(2023/10/1)

2023年10月1日

前回の内容に引き続き、今回も仮想通貨botの開発状況をまとめていきます。

今回はこちらの記事を参考にして、「ドンチャン・チャネル・ブレイクアウトのバックテスト検証」を行いました。

宿題：平均保有期間の追記

こちらの記事を参考にして「平均保有期間」も算出できるようにコードを書き換えました。

flag変数に買いと売りの保有期間を示す変数を追加しました。

その変数を集計用の関数とコンソール出力用の関数にそれぞれ組み込んで完成です。

ドンチャン・チャネル・ブレイクアウトとは

ブレイクアウトとは、過去の一定期間における揉み合いを脱した瞬間をエントリーのシグナルとして捉える手法です。

ドンチャン・チャネル・ブレイクアウトは、ブレイクアウト手法の一つで「過去一定期間における最高値(または最安値)の更新」に注目します。

例えば、「過去20期間における高値を更新したときに買いでエントリーする」「過去20期間の安値を更新したときに売りでエントリーする」場合などが挙げられます。

ブレイクアウト手法には、他にもオープニング・レンジ・ブレイクアウトなどがあります。

詳しい解説は、ジョン・ヒル著「究極のトレーディングガイド」などに載っています。

今回はロジックとしてコードに落とし込みやすいため「ドンチャンブレイクアウト」を取り上げてコードを書いていきます。

ドンチャンブレイクアウトのコードを書く

今回書いたコードは以下の通り。

詳しい解説はこちらの記事で。

import requests
from datetime import datetime
import time

chart_sec = 3600
term = 20
wait = 0

def get_price(min , before=0, after=0):
    price = []
    params = {"periods" : min }
    if before != 0:
        params["before"] = before
    if after != 0:
        params["after"] = after
    
    response = requests.get("https://api.cryptowat.ch/markets/bitflyer/btcfxjpy/ohlc",params)
    data = response.json()
    
    if data["result"][str(min)] is not None:
        for i in data["result"][str(min)]:
            if i[1] != 0 and i[2] != 0 and i[3] != 0 and i[4] != 0:
                price.append({ "close_time" : i[0],
                              "close_time_dt" : datetime.fromtimestamp(i[0]).strftime('%Y/%m/%d %H:%M'),
                              "open_price" : i[1],
                              "high_price" : i[2],
                              "low_price" : i[3],
                              "close_price" : i[4] })
        return price
    else:
        print("データが存在しません")
        return None

def print_price( data ):
    print( "時間: " + datetime.fromtimestamp(data["close_time"]).strftime('%Y/%m/%d %H:%M') + " 高値： " + str(data["high_price"]) + " 安値： " + str(data["low_price"]) )


def donchian( data,last_data ):
    
    highest = max(i["high_price"] for i in last_data)
    if data["high_price"] > highest:
        return {"side":"BUY","price":highest}
    
    lowest = min(i["low_price"] for i in last_data)
    if data["low_price"] < 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":
        print("過去{0}足の最高値{1}円を、直近の高値が{2}円でブレイクしました".format(term,signal["price"],data["high_price"]))
        print(str(data["close_price"]) + "円で買いの指し値注文を出します")
        
        #買い注文のコード挿入
        
        flag["order"]["exist"] = True
        flag["order"]["side"] = "SELL"
    
    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 close_position( data,last_data,flag ):
    
    flag["position"]["count"] += 1
    signal = donchian( data,last_data )
    
    if flag["position"]["side"] == "BUY":
        if signal["side"] == "SELL":
            print("過去{0}足の最安値{1}円を、直近の安値が{2}円でブレイクしました".format(term,signal["price"],data["low_price"]))
            print("成行注文を出してポジションを決済します")
            
            #決済の成行注文コード挿入
            
            flag["position"]["exist"] = False
            flag["position"]["count"] = 0
            
            print("さらに" + str(data["close_price"]) + "円で売りの指し値注文を入れてドテンします")
            
            #売り注文のコードを挿入
            
            flag["order"]["exist"] = True
            flag["order"]["side"] = "SELL"
        
    if flag["position"]["side"] == "SELL":
        if signal["side"] == "BUY":
            print("過去{0}足の最高値{1}円を、直近の高値が{2}円でブレイクしました".format(term,signal["price"],data["high_price"]))
            print("成行注文を出してポジションを決済します")
            
            #決済の成り行き注文コードを挿入
            
            flag["position"]["exist"] = False
            flag["position"]["count"] = 0
            
            print("さらに" + str(data["close_price"]) + "円で買いの指し値注文を入れてドテンします")
            
            #買い注文のコードを挿入
            
            flag["order"]["exist"] = True
            flag["order"]["side"] = "BUY"
            
    return flag

price = get_price(chart_sec)
last_data = []

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

i = 0
while i < len(price):
    
    if len(last_data) < term:
        last_data.append(price[i])
        print_price(price[i])
        time.sleep(wait)
        i += 1
        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)

今回は全て手書きしました。

コードを書き写しながらそれぞれの意味を考えたため、結果的に理解が深まりました。

実装するロジックは主に以下の３つです。

  • 「過去20期間の値動きを観測する」
  • 「最高値(最安値)を更新したらエントリーする」
  • 「エントリーした後、過去20期間の最高値(最安値)を更新したら手仕舞いをして、逆方向にエントリー(ドテン)する」

「過去20期間」の部分は調整可能であるため、本格的な実装時にはエントリーと手仕舞いのタイミングを変えながら勝てるロジックを突き詰めていくことになります。

今回はデータ検証の練習なのでとりあえず「過去20期間」に設定しています。

データの削除と追加

del last_data[0]
    last_data.append( data )

↑のコードはデータの削除と追加を同時に行なっています。

delで最も古いローソク足のデータを削除して、appendで最も新しいローソク足のデータを追加しています。

この１文を書き加えておくことで、ローソク足のデータ数を常に20足分に保つことができます。

コードの実行

いくつかのエラーが出ましたが、修正を重ねて無事に稼働させることができました。

過去3週間分のデータに基づいて、実行されています。

こちらの出力結果をもとに、勝率の検証を行うコードを組み込んでいきます。

バックテスト実行

こちらの記事を参考にして、コードを書いて実行しました。

実行結果は以下の通り。↓170万円の負けです。

このままでは勝つのは難しそうですね。

パラメータを「5分足を使用」に変更してみましたが、それでも勝率は良くないです。

総損益はマイナス100万円ほどです。

この後も幾つかパラメータをいじってみましたが、CryptowatchのAPI使用制限に引っかかってしまったので、今回の検証はここまでです。

まとめ

Cryptowatchの使用制限に引っかかるとバックテストができなくなってしまうので、過去データは定期的にまとめて取得してファイル保存しておくようにします。

この点は、次回の記事でまとめます。

また、コードの内容をさらっと読み解けるようになってきました。

現段階では「このコードが何をやっているのか」を理解できるとコードの修正も素早く行うことができます。

また、「修正方法の検索」もかなり素早く行うことができます。

ちまちまと手書きでコードを書くことは一見面倒に思えますが、意味を理解できていないものをまるっとコピペすると学びが得られません。

1行ずつ書き写すことで「理解できていない部分に気づくことができる」ため、今後も手書きでコードを書き写しながら意味を考えたり調べたりすることを続けていきます。

-Bot