仮想通貨Botで効率的にバックテストを回すためのコード
はじめまして。
技術系ブログはおろか、そもそもブログを書くのも今回が初めてのminaulと申します。大目に見て頂けますと幸いです。
今回は仮想通貨Botを作る上で欠かせない、バックテスト検証用のコードについて執筆してみたいと思います。
はじめに
まずはじめに簡単に用語解説しておくと、仮想通貨Botとは、仮想通貨のシステムトレードをプログラミングの実装により自動化したものになります。
バックテストとは、過去データに対して上記のBotを運用していた場合、どのような結果が得られたかを検証する方法です。
私自身、バックテストで戦略を考察している段階で、最初のうちはなにも考えずコード全体を書き殴っていたのですが、戦略が10、20と増えていくにつれ、めちゃくちゃ重くなっていき不便を感じるようになりました。
そこでようやく、綺麗にまとまったコードを作る必要性をひしひしと感じるようになり、本コードに至りました。
ちなみに、本記事の想定読者は、初〜中級者の方を対象にしています。
本コードの解説記事を見ても理解が難しいという方は、こちらの記事を先に読んで頂くことをお勧めします。
仮想通貨Botに関して、初心者の方向けに非常に分かりやすくまとまっています。
本コードの骨子も、こちらの記事を参考に作っています。
言語はPythonで、開発環境はJupyter Notebookです。
概要
本コードの概要について、簡単に説明します。
まず、本コードの大部分は共通ロジックとして扱っています。そのため、検証の際の可変部分は戦略ロジック部分のみです。
こうすることで、効率的にバックテストを回せるような設計にしています。
具体的には、設定部分、共通ロジック、戦略ロジック、メイン処理にまとめています。
つまり、最初に共通ロジックとその他設定などを実行すれば、検証時は戦略ロジックをいじるだけで検証を回せるということです。
順番に以下の通り、章立てでそれぞれ説明していきます。
設定
共通ロジック
戦略ロジック
メイン処理
1. 設定
本題に入る前に、最低限必要になる各種設定を行います。
ライブラリのインポート、ログ設定、基本値の設定です。
1.1 ライブラリのインポート
まずは本コードの実行に必要なライブラリを各自インポートします。
# ---------ライブラリ------------
import requests
import json
from datetime import datetime
import time
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
import talibNumPyやTA-Libについては事前にインストールする必要があります。
ちなみに、TA-Libとはテクニカル分析計算をカバーしたファイナンス用のライブラリです。200以上のテクニカル指標に対応しており、重宝される存在です。
1.2 ログ設定
次にコード全体のログ設定を行います。
# ---------ログ設定------------
from logging import getLogger, StreamHandler, FileHandler, Formatter
logger = getLogger('LoggingTest')
for h in logger.handlers[:]:
logger.removeHandler(h)
h.close()
logger.setLevel(10)
# ログ出力の必要がなければ、ファイル／コンソール設定はコメントアウト
# fh = FileHandler('log_bot_test' + '.log')
# logger.addHandler(fh)
# sh = StreamHandler()
# logger.addHandler(sh)
# formatter = Formatter('%(asctime)s: %(message)s', datefmt="%Y-%m-%d %H:%M:%S")
# fh.setFormatter(formatter)
# sh.setFormatter(formatter)ログ設定の際の注意点として、直接loggingをインポートしないようにしています。また、実行するたびにログが重複出力されないように、for文のところで実行毎に消しています。
ファイルやコンソールに出力したい場合は、コメントアウトの部分を外していただければ、それぞれ出力されるようになっています。
ログについては、このままコピペして頂ければ動きますし、必要なければ記述する必要もありません。記述しない場合は、以降で出てくるlogger.info()を都度コメントアウトする必要があります。
ログ設定に関しては、千差万別だと思うので、各自でお好みの設定にしていただければと思います。
1.3 設定値
共通ロジックの基本値を設定します。
# -----------設定項目---------------
chart_sec = 3600 # 1足の時間設定（秒）
stop_range = 2 # ストップのレンジ幅
trade_risk = 0.05 # 1トレードあたり口座の損失許容率
levarage = 2 # レバレッジ倍率の設定
start_funds = 1000000 # シミュレーション時の初期資金
wait = 0 # ループの待機時間
slippage = 0.00 # 手数料・スリッページ具体的な値に関しては、それぞれお好みで設定していただければと思います。新たな設定値を追加したい場合は、ここに追加して管理します。
ちなみに戦略ロジックの独自の基本値を設定する場合は、後述しますがここには記述しません。あくまで共通ロジックの基本値のみを設定します。
2. 共通ロジック
共通ロジックは3つのクラスに分けてコード管理します。
具体的には以下の通りです。
補助ツール
売買ロジック
バックテスト
2.1 補助ツール
補助系の関数は全てここに記述して管理します。
# ----------補助ツールのクラス-----------
class Tool:
# CryptowatchのAPIを使用する関数
def get_price(self, 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
# json形式のファイルから価格データを読み込む関数
def get_price_from_file(self, path , after, before):
file = open(path, "r", encoding='utf-8')
min = chart_sec
price = []
data = json.load(file)
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[after * -1 : before * -1]
else:
print("データが存在しません")
return None
# talibで使用可能なデータを準備する関数
def data_talib(self):
data_talib_open = []
data_talib_high = []
data_talib_low = []
data_talib_close = []
for i in range(len(price)):
data_talib_open.append(price[i]["open_price"])
data_talib_high.append(price[i]["high_price"])
data_talib_low.append(price[i]["low_price"])
data_talib_close.append(price[i]["close_price"])
ta = pd.DataFrame({
"Open" : data_talib_open,
"High" : data_talib_high,
"Low" : data_talib_low,
"Close" : data_talib_close
})
return ta
# 平均ボラティリティを計算する関数
def calculate_volatility(self):
ta = data_talib()
# ボラティリティはATRで管理（基本単位：14）
volatility = talib.ATR(ta["High"], ta["Low"], ta["Close"], timeperiod=14)
for i in range(14):
if np.isnan(volatility[i]):
volatility[i] = volatility[14] # NaN値の場合計算不可のため、擬似的に値を置き換える
for k in range(1, 15):
if np.isnan(volatility[k-1]):
volatility[k-1] = volatility[14] # 指値・売買ロジック用の置き換え
return volatility
# 注文ロットを計算する関数
def calculate_lot(self, flag):
lot = 0
balance = flag["records"]["funds"]
volatility = self.calculate_volatility()
stop = stop_range * volatility[i]
calc_lot = np.floor(balance * trade_risk / stop * 100) / 100
able_lot = np.floor(balance * levarage / data["close_price"] * 100) / 100
lot = min(able_lot, calc_lot)
flag["records"]["log"].append("現在のアカウント残高は{}円です".format(balance))
flag["records"]["log"].append("許容リスクから購入できる枚数は最大{}BTCまでです".format(calc_lot))
flag["records"]["log"].append("証拠金から購入できる枚数は最大{}BTCまでです".format(able_lot))
return lot, stop, flag合計5つの関数で形成されています。一つずつ説明していきます。
get_price
CryptowatchのAPIを使用して、仮想通貨（以降ビットコイン（BTC）とする）の価格を取得する関数です。最大で直近6,000件の価格データを返します。
get_price_from_file
json形式のファイルに保存されているデータから、価格を取得する関数です。ファイルに存在するデータ分を取得できるので、6,000件以上のデータを使って検証したい場合は、こちらを使います。また、返り値をprice[-1*after:-1*before]とすることで、任意のデータ件数・範囲を指定できます。ただし、独自でファイルを用意する必要があるため、上記と比してハードルは高めです。
data_talib
TA-Lib（以降talib）で必要となる価格データを集計する関数です。talibの引数に用意する価格データ（終値など）はリスト形式では計算エラーを吐いてしまいます。そのため、dataframeやndarrayの形式でデータを渡す必要があります。今回はOHLCをリスト化した後、dataframeに格納しています。
calculate_volatility
ボラティリティを計算する関数です。今回はボラティリティをATRで管理します。ATRとはTR（トゥルーレンジ）の期間平均です。ATRの計算は地味に面倒ですが、talibを用いることで1行で完結できます。先程のdata_talib関数を参照することで、高値・安値・終値の引数を渡して計算しています。
また、volatility[i]について、iが0〜13の間はNaN値を返すため、計算不可になってしまいます。そのため、擬似的にi=14で計算しています。指値の場合はさらに[i-1]を使うため、あわせて記述します。
calculate_lot
注文ロットを計算する関数です。リスク管理に基づくロット、レバレッジに基づくロットの2つを軸にロット管理を行います。ストップの計算も同時に行っています。ロット管理に関しては、かなりシンプルにしているのでカスタムの余地もあります。
2.2 売買ロジック
続いて、Botの核となる売買ロジックについての関数を見ていきます。
# ----------売買ロジックのクラス-----------
class EntryExit:
tool = Tool() # Toolクラスのインスタンス化（実体化）
back = BackTest() # BackTestクラス（先にBackTestクラスを実行する必要あり）
# エントリーシグナルを判定して成行注文を出す関数
def entry_signal(self, flag):
signal = logic_signal() # ロジックごとにシグナル関数が異なるが、ここで同様に参照する
if signal["side"] == "BUY":
logger.info("エントリーシグナルが発生しました")
lot, stop, flag = tool.calculate_lot(flag)
if lot > 0.01:
logger.info("{0}円で{1}BTCの買い注文を出します".format(data["close_price"], lot))
# ここに買い注文のコードを入れる
logger.info("{}円にストップを入れます".format(data["close_price"] - stop))
flag["order"]["exist"] = True
flag["order"]["side"] = "BUY"
flag["order"]["price"] = data["close_price"]
flag["order"]["lot"], flag["order"]["stop"] = lot, stop
else:
logger.info("注文可能枚数{}が、最低注文単位に満たなかったため注文を見送ります".format(lot))
return flag
# エントリーシグナルに対して指値注文を出す関数
def entry_signal_limit(self, flag):
signal = logic_signal()
if signal["side"] == "BUY":
logger.info("エントリーシグナルが発生しました")
lot, stop, flag = tool.calculate_lot(flag)
ATR = tool.calculate_volatility() # ボラの計算にATRを使っているので使い回す
delta = round(ATR[i-1] * entry_limit) # 指値距離はATRで管理
if lot > 0.01:
limit_1 = last_data[-1]["close_price"] - delta # 指値価格
logger.info("{0}円で{1}BTCの買い指値を出します".format(limit_1, lot))
if data["low_price"] < limit_1: # 指値の約定条件（ ＝ 現在の安値が一足前の指値を下回る ＝ 指値が刺さっている）
logger.info("１足前の指値({}円)が現在の安値({}円)を上回るので、指値は{}円で約定したものとみなす".format(limit_1, data["low_price"], limit_1))
# 指値注文のコードを入れる
logger.info("{}円にストップを入れます".format(limit_1 - stop))
flag["order"]["exist"] = True
flag["order"]["side"] = "BUY"
flag["order"]["price"] = limit_1
flag["order"]["lot"], flag["order"]["stop"] = lot, stop
else:
logger.info("指値価格({}円)が安値({}円)を下回るので、指値は約定しなかったものとみなす".format(limit_1, data["low_price"]))
else:
logger.info("注文可能枚数{}が、最低注文単位に満たなかったため注文を見送ります".format(lot))
return flag
# サーバーに出した注文が約定したか確認する関数
def check_order(self, flag):
# 注文が約定した際に、ポジションにフラグ変数を切り替える
flag["order"]["exist"] = False
flag["order"]["count"] = 0
flag["position"]["exist"] = True
flag["position"]["side"] = flag["order"]["side"]
flag["position"]["price"] = flag["order"]["price"]
flag["position"]["stop"] = flag["order"]["stop"]
flag["position"]["lot"] = flag["order"]["lot"]
return flag
# 手仕舞いのシグナルが出たら決済の成行注文
def close_position(self, flag):
if flag["position"]["exist"] == False:
return flag
flag["position"]["count"] += 1
signal = logic_signal()
if flag["position"]["side"] == "BUY":
if signal["side"] == "SELL":
logger.info("エグジットのシグナルが発生しました")
logger.info(str(data["close_price"]) + "円でポジションを決済します")
# 決済の注文コードを入れる
back.records(flag, data, data["close_price"])
flag["position"]["exist"] = False
flag["position"]["count"] = 0
return flag
# エグジットシグナルに対して指値注文で決済する関数
def close_position_limit(self, flag):
if flag["position"]["exist"] == False:
return flag
flag["position"]["count"] += 1
signal = logic_signal()
ATR = tool.calculate_volatility()
delta = round(ATR[i-1] * exit_limit)
if flag["position"]["side"] == "BUY":
if signal["side"] == "SELL":
logger.info("エグジットのシグナルが発生しました")
limit_2 = last_data[-1]["close_price"] + delta
logger.info("{0}円で売り指値を出します".format(limit_2))
if data["high_price"] > limit_2:
logger.info("１足前の指値({}円)が高値({}円)を下回ったので指値が約定したとみなし、ポジションを決済します".format(limit_2, data["high_price"]))
# 決済の指値注文コードを入れる
back.records(flag, data, limit_2) # 取引の結果を記録するrecords関数の引数にlimitを指定する
flag["position"]["exist"] = False
flag["position"]["count"] = 0
else:
logger.info("指値価格({}円)が高値({}円)を上回ったので、指値は約定しなかったとみなす".format(limit_2, data["high_price"]))
return flag
# 損切りラインにかかったら成行注文で決済する関数
def stop_position(self, flag):
if flag["position"]["side"] == "BUY":
stop_price = flag["position"]["price"] - flag["position"]["stop"]
if data["low_price"] < stop_price:
logger.info("{0}円の損切りラインに引っ掛かりました".format(stop_price))
ATR = tool.calculate_volatility()
stop_price = round(stop_price - 2 * ATR[i-1] / (chart_sec / 60))
logger.info(str(stop_price) + "円あたりで成行注文を出してポジションを決済します。\n")
# 決済の注文コードを入れる
back.records(flag, data, stop_price, "STOP")
flag["position"]["exist"] = False
flag["position"]["count"] = 0
return flag成行注文と指値注文の2パターンに対応できるように、それぞれ関数を定義しています。
また、買いのみに対応しているため、売りを導入したい場合は、各自でコードを付け足してください。
以降は、基本的に買いのみのシンプルなロジックで進めていきます。
entry_signal
エントリーシグナルに対して、成行注文を行う関数です。エントリーシグナルは、戦略ロジックによって各々定義されます。注文状況などはフラグ変数で管理しています。
entry_signal_limit
エントリーシグナルに対して、指値注文を行う関数です。指値距離はATRで管理していますが、お好みで結構です。ちなみに本コードの指値特有のロジックとして、1足前の情報を使うことに注意が必要です。指値の約定条件については、コード中の説明文の通りです。また、指値が刺さらなければ1足で注文をキャンセルする設計になっています。持続させたい場合は、else時にcount変数などを加えると改良できそうです。
check_order
注文の約定状況を確認する関数です。約定時にこの関数を呼び出して、フラグ変数をorderからpositionに切り替えます。
close_position
エグジットシグナルに対して、成行注文で決済する関数です。エグジットシグナルも戦略ロジックによって異なります。順序が逆転しますが、エグジット時にrecords関数を呼ぶ必要があるため、ここでバックテストクラスから呼んでいます。
close_position_limit
エグジットシグナルに対して、指値注文で決済する関数です。エントリー時とは逆に、高値の情報を使って指値が刺さっているかを確認します。
stop_position
損切りラインに対して、ストップの成行注文を出す関数です。安値が損切りラインに引っ掛かった場合に発動し、ストップ価格で決済します。
2.3 バックテスト
バックテストの結果集計に必要な関数を集約したクラスです。
# ----------バックテストのクラス-----------
class BackTest:
# 各トレードのパフォーマンスを記録する関数
def records(self, flag, data, close_price, close_type=None):
# 取引手数料等の計算
entry_price = int(round(flag["position"]["price"] * flag["position"]["lot"]))
exit_price = int(round(close_price * flag["position"]["lot"]))
trade_cost = round(exit_price * slippage)
logger.info("スリッページ・手数料として " + str(trade_cost) + "円を考慮します")
flag["records"]["slippage"].append(trade_cost)
# 手仕舞った日時と保有期間を記録
flag["records"]["date"].append(data["close_time_dt"])
flag["records"]["holding-periods"].append(flag["position"]["count"])
# 損切りにかかった回数をカウント
if close_type == "STOP":
flag["records"]["stop-count"].append(1)
else:
flag["records"]["stop-count"].append(0)
# 値幅の計算
buy_profit = exit_price - entry_price - trade_cost
sell_profit = entry_price - exit_price - trade_cost
# 利益が出てるかの計算
if flag["position"]["side"] == "BUY":
flag["records"]["side"].append("BUY")
flag["records"]["profit"].append(buy_profit)
flag["records"]["return"].append(round(buy_profit / entry_price * 100, 4))
flag["records"]["funds"] = flag["records"]["funds"] + buy_profit
if buy_profit > 0:
logger.info(str(buy_profit) + "円の利益です")
else:
logger.info(str(buy_profit) + "円の損失です")
return flag
# バックテストの集計用の関数
def backtest(self, flag):
# 本文からの移植
print("--------------------------")
print("テスト期間：")
print("開始時点 ： " + str(price[0]["close_time_dt"]))
print("終了時点 ： " + str(price[-1]["close_time_dt"]))
print(str(len(price)) + "件のローソク足データで検証")
print("--------------------------")
# 成績を記録したPandas DataFrameを作成
records = pd.DataFrame({
"Date" : pd.to_datetime(flag["records"]["date"]),
"Profit" : flag["records"]["profit"],
"Side" : flag["records"]["side"],
"Rate" : flag["records"]["return"],
"Stop" : flag["records"]["stop-count"],
"Periods" : flag["records"]["holding-periods"],
"Slippage" : flag["records"]["slippage"]
})
# 連敗回数をカウントする
consecutive_defeats = []
defeats = 0
for p in flag["records"]["profit"]:
if p < 0:
defeats += 1
else:
consecutive_defeats.append(defeats)
defeats = 0
# テスト日数を集計
time_period = datetime.fromtimestamp(last_data[-1]["close_time"]) - datetime.fromtimestamp(last_data[0]["close_time"])
time_period = int(time_period.days)
# 総損益の列を追加する
records["Gross"] = records.Profit.cumsum()
# 資産推移の列を追加する
records["Funds"] = records.Gross + start_funds
# 最大ドローダウンの列を追加する
records["Drawdown"] = records.Funds.cummax().subtract(records.Funds)
records["DrawdownRate"] = round(records.Drawdown / records.Funds.cummax() * 100, 1)
# 買いエントリーと売りエントリーだけをそれぞれ抽出する
buy_records = records[records.Side.isin(["BUY"])]
# sell_records = records[records.Side.isin(["SELL"])]
# 月別のデータを集計する
records["月別集計"] = pd.to_datetime(records.Date.apply(lambda x: x.strftime('%Y/%m')))
grouped = records.groupby("月別集計")
month_records = pd.DataFrame({
"Number" : grouped.Profit.count(),
"Gross" : grouped.Profit.sum(),
"Funds" : grouped.Funds.last(),
"Rate" : round(grouped.Rate.mean(), 2),
"Drawdown" : grouped.Drawdown.max(),
"Periods" : grouped.Periods.mean()
})
print("バックテストの結果")
print("--------------------------")
print("買いエントリーの成績")
print("--------------------------")
print("トレード回数 : {}回".format(len(buy_records)))
print("勝率 : {}％".format(round(len(buy_records[buy_records.Profit > 0]) / len(buy_records) * 100, 1)))
print("平均リターン : {}％".format(round(buy_records.Rate.mean(), 2)))
print("総損益 : {}円".format(buy_records.Profit.sum()))
print("平均保有期間 : {}足分".format(round(buy_records.Periods.mean(), 1)))
print("損切りの回数 : {}回".format(buy_records.Stop.sum()))
print("--------------------------")
print("総合パフォーマンス")
print("--------------------------")
print("全トレード数 : {}回".format(len(records)))
print("勝率 : {}％".format(round(len(records[records.Profit > 0]) / len(records) * 100, 1)))
print("平均リターン : {}％".format(round(records.Rate.mean(), 2)))
print("平均保有期間 : {}足分".format(round(records.Periods.mean(), 1)))
print("損切りの回数 : {}回".format(records.Stop.sum()))
print("")
print("最大の勝ちトレード : {}円".format(records.Profit.max()))
print("最大の負けトレード : {}円".format(records.Profit.min()))
print("最大ドローダウン : {0}円／{1}％".format(-1 * records.Drawdown.max(), -1 * records.DrawdownRate.loc[records.Drawdown.idxmax()]))
print("最大連敗回数 : {}回".format(max(consecutive_defeats)))
print("利益合計 : {}円".format(records[records.Profit > 0].Profit.sum()))
print("損失合計 : {}円".format(records[records.Profit < 0].Profit.sum()))
print("最終損益 : {}円".format(records.Profit.sum()))
print("")
print("初期資金 : {}円".format(start_funds))
print("最終資金 : {}円".format(records.Funds.iloc[-1]))
print("運用成績 : {}％".format(round(records.Funds.iloc[-1] / start_funds * 100, 2)))
print("手数料合計 : {}円".format(-1 * records.Slippage.sum()))
print("-----------------------------------")
print("各成績指標")
print("-----------------------------------")
print("CAGR(年間成長率) : {}％".format(round((records.Funds.iloc[-1] / start_funds) ** (time_period / 365) * 100 - 100, 2)))
print("MARレシオ : {}".format(round((records.Funds.iloc[-1] / start_funds - 1) * 100 / records.DrawdownRate.max(), 2)))
print("シャープレシオ : {}".format(round(records.Rate.mean() / records.Rate.std(), 2)))
print("プロフィットファクター : {}".format(round(records[records.Profit>0].Profit.sum() / abs(records[records.Profit<0].Profit.sum()), 2)))
print("損益レシオ : {}".format(round(records[records.Profit>0].Rate.mean() / abs(records[records.Profit<0].Rate.mean()), 2)))
print("-----------------------------------")
print("月別の成績")
for index,row in month_records.iterrows():
print("-----------------------------------")
print("{0}年{1}月の成績".format(index.year, index.month))
print("-----------------------------------")
print("トレード数 : {}回".format(row.Number.astype(int)))
print("月間損益 : {}円".format(row.Gross.astype(int)))
print("平均リターン : {}％".format(row.Rate))
print("継続ドローダウン : {}円".format(-1 * row.Drawdown.astype(int)))
print("月末資金 : {}円".format(row.Funds.astype(int)))
# 損益曲線をプロット
plt.plot(records.Date, records.Funds)
plt.xlabel("Date")
plt.ylabel("Balance")
plt.xticks(rotation=50) # x軸の目盛りを50度回転
plt.show()
# フラグ変数管理の関数
def flags(self):
flag = {
"order" : {
"exist" : False,
"side" : "",
"price" : 0,
"stop" : 0,
"ATR" : 0,
"lot" : 0,
"count" : 0
},
"position" : {
"exist" : False,
"side" : "",
"price" : 0,
"stop" : 0,
"ATR" :0,
"lot" : 0,
"count" : 0
},
"records" : {
"date" : [],
"profit" : [],
"return" : [],
"side" : [],
"stop-count" : [],
"funds" : start_funds,
"holding-periods" : [],
"slippage" : [],
"log" : []
}
}
return flagロジックを回した時の取引結果の集計と各指標の計算を行っています。
records
トレードの結果を記録する関数です。エントリーとエグジット時の価格差から、利益が出たか損失を出したかのパフォーマンスを算出します。
backtest
バックテストの集計・結果を出す関数です。成績をdataframeに記録し、そこから各々の結果や成績指標などを算出しています。バックテストでどういった指標を検証するかは、作成者の好みに左右されます。リターンやリスク、堅牢性など、検証したい指標があればここを各自カスタムしていくことになります。
また、図表の可視化についても色々とカスタムできます。
ちなみに冒頭のprint文の箇所は、メイン処理の本文を軽くするために、移植しています。
flags
フラグ変数を管理する関数です。大きくorder, position, recordsに分けて、それぞれフラグ変数を使い、ロジック全体の流れをコントロールしています。ちなみにこちらも、本文のコード量削減のためにバックテストクラスに移しています。
3. 戦略ロジック
今回の戦略ロジックは、ボリンジャーバンドを採用してみます。簡単に説明すると、移動平均線（middle）と上下の標準偏差（upper, lower）の範囲から形成されるテクニカル指標です。基本的には±2σのラインが用いられます。
また、今回は折角なので指値で書いてみようと思います。
3.1 設定値
戦略ロジックに入る前に、予め基本値の設定をしておきます。
# -----------設定項目---------------
bbands_period = 10 # ボリンジャーバンドに用いる期間設定
std_value = 2 # ボリンジャーバンドに用いる標準偏差の数値設定
entry_limit = 1.0 # エントリー時の指値距離に用いるATRの係数（参考値：0.1〜1.5）
exit_limit = 1.0 # エグジット時のdeltaATR係数（上と同じ）ここで、個別に戦略ロジック毎に設定値を変更します。
今回はボリンジャーバンド戦略なので、それに付随する基本値を設定します。
また、指値距離の係数もここで設定しています。
3.2 個別ロジック
いよいよ、ボリンジャーバンドの戦略部分です。様々な戦略のバックテストを回す場合、基本的に変えるのはこの部分のみです。
# ------------ロジックの個別部分-----------------
# ボリンジャーバンドの値を計算する関数
def calculate_bbands(value_1, value_2):
ta = tool.data_talib()
# talibを用いたボリンジャーバンドの計算
bbands_upper, bbands_middle, bbands_lower = talib.BBANDS(ta["Close"], timeperiod=value_1, nbdevup=value_2, nbdevdn=value_2, matype=0)
return bbands_upper, bbands_middle, bbands_lower
# ボリンジャーバンドのエントリー/エグジットシグナルを判定する関数
def logic_signal():
# 計算に用いる数値の準備
bbands_upper, bbands_middle, bbands_lower = calculate_bbands(bbands_period, std_value)
# -1ずつずらすことに注意（指値の場合は過去の情報を使う必要があるため）
# エントリーサイン：+２σのラインを現在の株価が上抜け
if last_data[-1]["high_price"] > bbands_upper[i-1]:
return {"side":"BUY", "price":last_data[-1]["close_price"]}
# エグジットサイン：±０σのラインを現在の株価が下抜け
if last_data[-1]["low_price"] < bbands_middle[i-1]:
return {"side":"SELL", "price":last_data[-1]["close_price"]}
return {"side":None, "price":0}可変部分はここだけです。効率を重視してコンパクトにまとめています。
戦略に関しては、+2σ, ±0σをシグナルラインとする基本的な順張り戦略です。
calculate_bbands
ボリンジャーバンドの値を計算する関数です。以下のlogic_signal関数で使うため、事前に準備します。ボリンジャーバンド（以降bbands）はtalibを使うことで手軽に計算できます。bbandsにおけるupper, middle, lowerの3つの値を返します。計算に支障はないため、NaN値に対して特に処理はしません。
ここで注意ですが、talibを使ってテクニカル指標を計算する場合は、先にToolクラスをインスタンス化する必要があります。
logic_signal
戦略ロジックのエントリー／エグジットシグナルを判定する関数です。各シグナルの発生条件が満たされれば、方向と価格を返します。bbandsに限らず様々なテクニカル指標に対して、シグナル発生時に注文／決済の命令を与える心臓部となります。今回は指値のため、1足前の情報からシグナルを得ています。成行の場合はlast_data[-1]をdataに、bbands_upper[i-1], middle[i-1]をupper[i], middle[i]に変更する必要があります。
4. メイン処理
最後に、ロジック全体のメイン処理をループで回します。
戦略ロジックを変えた場合は、このメイン処理も同時に実行しなければ反映はされません。同一ロジックであっても、メイン処理に関わる数値を変更する場合は、メイン処理も実行しないと、バックテストの結果には反映されないので注意してください。
メイン処理の前に、メイン処理で使う共通ロジックの3つのクラスをインスタンス化する必要があります。説明が遅れましたが、クラスをインスタンス（実体化）することで、クラス内の関数を使用できるようになります。
# -----------クラスのインスタンス化----------------
tool = Tool()
enex = EntryExit()
back = BackTest()前述した通り、戦略ロジックにtalibを使う場合は、先にこれらのインスタンス化を実行しても良いでしょう。
さて、いよいよ最後のメイン処理の部分に入ります。
# -----------メイン処理--------------
# 価格チャートを取得
price = tool.get_price(chart_sec, after=1640962800)
# フラグ変数
flag = back.flags()
need_term = 1
last_data = []
i = 0
# メインのループ文
while i < len(price):
# 指値のため、一足遅く動かすので、last_dataが少し必要
if len(last_data) < need_term:
last_data.append(price[i])
i += 1
continue
data = price[i]
if flag["order"]["exist"]:
flag = enex.check_order(flag)
if flag["position"]["exist"]:
# flag = enex.stop_position(flag)
flag = enex.close_position_limit(flag)
else:
flag = enex.entry_signal_limit(flag)
last_data.append(data)
i += 1
time.sleep(wait)
back.backtest(flag)かなりスッキリとまとまったと思います。
順次説明していきます。
まず、価格データを取得します。今回は1時間足で、2022/1/1〜のデータで検証しようと思います。
次に、指値の計算に用いるため、メイン処理を回す前に事前にlast_dataを用意しておきます。need_termを設定することで、その分のlast_dataを蓄積します。指値に必要な最低期間は1です。
メインループを回す前にlast_dataを疑似的に準備することで、[i]に関するズレを無くすことができます。
メイン処理のループ中はデータを更新しながら、主にEntryExitクラスで売買ロジックを回しています。今回はストップを使っていませんが、必要に応じてコメントアウトを外して使ってください。
最後に、バックテスト関数を呼び出して、バックテストの結果を可視化して検証終了です。
以下、今回のパフォーマンス結果を示して、締めとしたいと思います。
結果
今回のバックテストの結果は以下の通りです。
--------------------------
テスト期間：
開始時点 ： 2022/01/01 00:00
終了時点 ： 2022/06/21 13:00
4117件のローソク足データで検証
--------------------------
バックテストの結果
--------------------------
買いエントリーの成績
--------------------------
トレード回数 : 66回
勝率 : 56.1％
平均リターン : 0.02％
総損益 : -139210円
平均保有期間 : 13.5足分
損切りの回数 : 0回
--------------------------
総合パフォーマンス
--------------------------
全トレード数 : 66回
勝率 : 56.1％
平均リターン : 0.02％
平均保有期間 : 13.5足分
損切りの回数 : 0回
最大の勝ちトレード : 82307円
最大の負けトレード : -176304円
最大ドローダウン : -259695円／-26.9％
最大連敗回数 : 4回
利益合計 : 763085円
損失合計 : -902295円
最終損益 : -139210円
初期資金 : 1000000円
最終資金 : 860790円
運用成績 : 86.08％
手数料合計 : 0円
-----------------------------------
各成績指標
-----------------------------------
CAGR(年間成長率) : -6.78％
MARレシオ : -0.52
シャープレシオ : 0.01
プロフィットファクター : 0.85
損益レシオ : 0.8
-----------------------------------
この結果を見て、どのように判断し、実運用に活かしていくかは人それぞれです。
パラメータを変えたり、期間設定を変えたり、はたまた戦略自体を大きく変えたらバックテスト結果はどう変化するのかなど、考察の余地は尽きません。
結果については、手短に終わらせようと思います。
今回の記事の趣旨はあくまでバックテストの結果論ではなく、バックテストを効率的に回すためのエンジニアリング的方法論です。
結果の考察などは、それぞれみなさんに任せたいと思います。
というよりむしろ、この試行錯誤の過程に、Botを自作する醍醐味があるような感じがします。
最後に
以上が、バックテストにおける効率的な検証コードの全体になります。
戦略ロジックに関しては、色々なテクニカル指標のバックテスト結果などの記事も今後出していければと考えています。
ブログ投稿は初めてなもので、コードなど含めちゃんと書けているか不安なところではありますが、そろそろ書き納めたいと思います。
それでは、ここまで読んでいただきありがとうございました。