自動売買BOTの世界では、長期的に使える強いトレードシステムのことを「堅牢性（ロバスト性）がある」と表現します。わかりやすい言葉でいうと「環境の変化に強く安定したシステム」のことです。
「堅牢性がある」とは、単に期待リターンが高いというだけではありません。以下のような特徴を備えた売買ロジックのことをいいます。
１．期待リターンに対してリスクが低い
２．毎月の期待リターンのバラツキ具合が小さい
３．最大ドローダウンが小さい
４．最適化された固定値のパラメーターが少ない
前回の記事では、自動売買BOTの勝率・期待リターンなどを計算する方法を勉強しましたが、それだけではシステムの堅牢性は評価できません。
この記事では、ドンチャン・ブレイクアウトBOTの堅牢性を評価するために資産推移のグラフを描写して、さらに最大ドローダウンを計算する方法を解説していきます。
１）グラフをpythonで描画する方法
先にpythonでグラフを描画する方法だけ解説しておきましょう。
大丈夫、めちゃくちゃ簡単です！ まずは練習としてBTCの日々の終値の価格グラフを作成してみましょう。以下のように書くだけです。
import matplotlib.pyplot as plt
import pandas as pd
# 日付の配列データ（x軸）
date_list = [
"2018/1/1",
"2018/1/2",
"2018/1/3",
"2018/1/4",
"2018/1/5",
"2018/1/6",
"2018/1/7",
"2018/1/8",
"2018/1/9",
"2018/1/10",
"2018/1/11",
"2018/1/12",
"2018/1/13",
"2018/1/14"
]
# BTC価格（ドル）のデータ（y軸）
price_list = [
13657,
14982,
15201,
15599,
17429,
17527,
16477,
15170,
14595,
14973,
13405,
13980,
14360,
13772
]
# 日付データに変換
date_list = pd.to_datetime(date_list)
# plot(x,y)でx軸とy軸を指定
plt.plot( date_list,price_list )
plt.xlabel("Date")
plt.ylabel("Price")
# 描写を実行
plt.show()
グラフの描画には matplotlib というライブラリを使います。
これはAnacondaを使ってpythonをインストールした方であれば、最初から入っていますので、以下の１行を先頭に書くだけで使用できます。
import matplotlib.pyplot as plt
コードの解説
まずX軸とY軸にそれぞれ描画したいデータを配列として用意します。
それぞれの要素の数が一致していないといけないので、注意してください。今回は、日付データ（date_list）と価格データ（price_list）にそれぞれ14個ずつのデータを用意しました。
次に、日付データを「テキスト型」から「日付型」に変換します。なぜ日付型に変換するのかというと、テキスト型のままだと以下のようなデータがすべてX軸に等間隔で並んでしまうからです。
例） date = ["1/1","1/2","3/2","3/4","4/2"]
テキスト型のままだとこの５個のデータは、同じ目盛り上に等しい間隔で並んでしまいます。1/1～1/2と、1/2～3/2が、同じ間隔で並んでしまうと、後ほど資産推移の曲線などを描画するときに困りますね。
これを避けるために、以下のコードで日付型に変換しておきます。
import pandas as pd # 日付データに変換 date_list = pd.to_datetime(date_list)
ここでは変換にpandasというライブラリを使用しています。こちらもAnacondaでPythonをインストールした方であれば、最初からインストールされています。そうでない方はpipを使ってください。
実行結果
当サイトと同じ方法で（Anacondaで）Pythonをインストールしている方なら、すでにSpyderという実行環境がインストールされています。そのため、上記のコードを実行すると、自動的に以下のようなウィンドウが立ち上がり、グラフの描画が実行されます。
グラフの描画の説明はこれだけです！
２）資産曲線を描く
まずは資産カーブを描くために、ポジションを手仕舞うたびにその時点での損益（資産推移）を記録するコードを作っておきましょう。やり方は前回までの記事と全く同じです。
データを記録する準備
資産推移のグラフを描くために必要なデータは以下の２つです。
１）各ポジションを手仕舞った時点での資産額
２）各ポジションを手仕舞った時点の日付（時間）
まずは資産額と日付を記録するために、flag変数に以下を追加しておきましょう。
"records":{
"buy-count": 0,
"buy-winning" : 0,
#(中略)
"date":[], # 追加
"gross-profit":[0],# 追加
"slippage":[],
"log":[]
}
総損益を記録するための配列データには、初期値として０を入れておきます。
そして手仕舞いの関数が呼ばれるたびに、その時点での日付データと総損益を append で記録しておきます。 以下のように書けばいいですね。
# 各トレードのパフォーマンスを記録する関数 def records(flag,data): # 手仕舞った日時の記録 flag["records"]["date"].append(data["close_time_dt"]) if flag["position"]["side"] == "BUY": # 追記部分 flag["records"]["gross-profit"].append( flag["records"]["gross-profit"][-1] + buy_profit ) if flag["position"]["side"] == "SELL": # 追記部分 flag["records"]["gross-profit"].append( flag["records"]["gross-profit"][-1] + sell_profit )
各トレードの成績を記録する関数（def records）で、ポジションを手仕舞ったときの日付データを記録しておきます。これはローソク足のデータを記録する変数（data）にある日時情報（data[“close_time_dt”]）を代入するだけです。
また、総損益を記録する変数（gross-profit）の前回の値（[-1]）を取り出し、そこに今回の最新の損益（buy_profit/sell_profit）を足した数字を、appendで配列の末尾に追加します。
これで準備は完了です。
グラフを描画する
最後にバックテスト集計用の関数（def backtest）の中で資産カーブを描画するためのコードを追加します。
# 損益曲線をプロット
del flag["records"]["gross-profit"][0] # X軸/Y軸のデータ数を揃えるため、先頭の0を削除
date_list = pd.to_datetime( flag["records"]["date"] ) # 日付型に変換
plt.plot( date_list, flag["records"]["gross-profit"] )
plt.xlabel("Date")
plt.ylabel("Balance")
plt.xticks(rotation=50) # X軸の目盛りを50度回転
plt.show()
最初の例と全く同じですね。
flag[records][gross-profit]には、初期値として０円が入っており、日付データと数が一致しないため、del [0]で先頭の０円を削除しています。また、pd.to_datetime()で、日付のテキストデータを日付型に変換しています。
では実行してみましょう！
実行結果
以下は、30期間のドンチャン・ブレイクアウト（損切なし・ドテン売買）を、１時間足を使って 2017/8/9 ～ 2018/4/16 までの期間、6000足分のローソク足データを使って検証した結果です。
▽ 最終結果
▽ 資産カーブ
右肩上がりといえば右肩上がりですが、何度か大きなドローダウン（利益が大きく削られる局面）を経験していることがわかりますね。では、この売買システムの最大ドローダウンはどのくらいの大きさなのでしょうか？
３）最大ドローダウンを計算する
まずpythonで最大ドローダウンの計算を実装するにあたって、ドローダウンとは何なのか？を明確に定義しておきましょう。
自動売買BOTを自作でプログラミングする最大のメリットは、あらゆるテクニカル指標・売買ロジック・バックテストの指標などをコーディングする過程で、正確にその意味を理解できるようになることです。
私もそうですが、文系で何となく投資本を読んでいるだけだと、ドローダウンの意味１つとっても、「何となくこういうものだろう」というイメージだけで理解してしまいがちです。しかし自分でコーディングするようになると、どんな指標・ロジックでも明確な定義を考える習慣がつきます。
定義を考える習慣がつくようになると、単に概念を本で学んで覚えているだけの人に比べて、その指標の活用方法や弱点に気づきやすくなります。
１．ドローダウンの定義
基本的にはドローダウンは、「ある時点での資産額と、それ以前の資産の最大額との落差のこと」と定義できます。
例えば、以下のように資産が推移したと過程しましょう。
１月 50万円
２月 100万円
３月 200万円 ← 直近の最大額
４月 180万円
５月 150万円 ← 最大ドローダウン
６月 170万円
７月 220万円 ← 直近の最大額を更新
８月 190万円
９月 200万円
10月 170万円 ← 最大ドローダウン
11月 210万円
この場合、大きなドローダウンは２つ存在します。
１つ目は、３月のピーク200万円から５月の150万円までのドローダウン（－50万円）です。もう１つは、７月のピーク（220万円）から10月の170万円までのドローダウン（－50万円）です。
この場合、最大ドローダウンは－50万円、ということになります。％表記にすると、50/220 = 0.2272… なので約22.7%のドローダウンです。
２．Pythonコードで実装する
ではこれを実装していきましょう。
手始めに、とりあえず、いつものように記録用の変数を追加します。以下のような変数を追加しておきましょう。
"records":{
"buy-count": 0,
"buy-winning" : 0,
# 追記
"drawdown": 0,
次にポジションを閉じるたびに呼ばれる「各トレードの成績を記録する関数」に、以下の３行を追加しましょう。
# 各トレードのパフォーマンスを記録する関数 def records(flag,data): # （中略） # ドローダウンの計算 drawdown = max(flag["records"]["gross-profit"]) - flag["records"]["gross-profit"][-1] if drawdown > flag["records"]["drawdown"]: flag["records"]["drawdown"] = drawdown
ポジションを閉じるたびに、その時点における過去の資産の最大額をmax（flag[records][gross-profit]）で探します。そこから、現時点での資産額（flag[records][gross-profit][-1]）を引いて現在のドローダウンを計算します。
そして現在のドローダウンが過去に記録したものよりも大きければ、flag[records][drawdown] に代入して値を更新します。これにより、最終的に「最大ドローダウン」だけが残ります。
そして最後に「バックテスト集計用の関数」に以下を追記します。
# バックテストの集計用の関数
def backtest(flag):
# （中略）
print("最大ドローダウン : {0}円 / {1}％".format(-1 * flag["records"]["drawdown"], -1 * round(flag["records"]["drawdown"]/max(flag["records"]["gross-profit"])*100,1) ))
ここで最大ドローダウンの金額を表示するとともに、そのパーセンテージも計算して表示しています。ではこれを実行してみましょう。
３．実行結果
以下のようになりました。
もし過去の半年間に１時間足で30期間のドンチャンブレイクアウトを採用していたら、この部分の最大ドローダウンが約48万円（－23％）だったということですね。
※ ２月中盤～後半にかけて、資産207万円 ⇒ 159万円のドローダウン
３）もし２月から自動売買BOTの運用を開始していた場合
では、もし２月からドンチャン・ブレイクアウトの自動売買BOTの運用を開始していた場合はどうなっていたのでしょうか？ これを最後に検証して、今回の記事を終えておきましょう。
以下のコードの部分を変更するだけです。
これを実行すると、以下のようになります。
▽ バックテストの結果
▽ 資産カーブ
１枚のBTCを売買したとすると、開始早々－20万円のドローダウンに見舞われることになります。その後、＋20万円まで持ち直しますが、そこから－約50万円のドローダウンを食らいます。
長期的には期待値プラスの売買ロジックでも、このように開始早々、大きなマイナスを食らったり、数カ月連続してプラス転換しない場合もあります。このようなときに、長期的な期待値を信じて我慢できるかどうか、あるいは「相場が変わったからもうこのシステムは通用しないんだ…」と諦めるべきかは、判断が難しいところです。
パラメーターの過剰最適化がされていない、バックテストのデータ数が十分である、など、堅牢性の高いシステムであればあるほど、このような局面でも期待値を信じて我慢することが可能になります。この章のはじめで、サンプル数の重要性などを説明したのはそのためです。
今回の勉強で使ったコード
最後に今回の記事で作ったpythonコードを掲載しておきましょう。
import requests
from datetime import datetime
import time
import numpy as np
import matplotlib.pyplot as plt
import pandas as pd
#-----設定項目
chart_sec = 3600 # １時間足を使用
term = 30 # 過去ｎ足の設定
wait = 0 # ループの待機時間
lot = 1 # BTCの注文枚数
slippage = 0.001 # 手数料・スリッページ
# CryptowatchのAPIを使用する関数
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 log_price( data,flag ):
log = "時間： " + datetime.fromtimestamp(data["close_time"]).strftime('%Y/%m/%d %H:%M') + " 高値： " + str(data["high_price"]) + " 安値： " + str(data["low_price"]) + "\n"
flag["records"]["log"].append(log)
return flag
# ドンチャンブレイクを判定する関数
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":
flag["records"]["log"].append("過去{0}足の最高値{1}円を、直近の高値が{2}円でブレイクしました\n".format(term,signal["price"],data["high_price"]))
flag["records"]["log"].append(str(data["close_price"]) + "円で買いの指値注文を出します\n")
# ここに買い注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "BUY"
flag["order"]["price"] = round(data["close_price"] * lot)
if signal["side"] == "SELL":
flag["records"]["log"].append("過去{0}足の最安値{1}円を、直近の安値が{2}円でブレイクしました\n".format(term,signal["price"],data["low_price"]))
flag["records"]["log"].append(str(data["close_price"]) + "円で売りの指値注文を出します\n")
# ここに売り注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "SELL"
flag["order"]["price"] = round(data["close_price"] * lot)
return flag
# サーバーに出した注文が約定したか確認する関数
def check_order( 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"]
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":
flag["records"]["log"].append("過去{0}足の最安値{1}円を、直近の安値が{2}円でブレイクしました\n".format(term,signal["price"],data["low_price"]))
flag["records"]["log"].append(str(data["close_price"]) + "円あたりで成行注文を出してポジションを決済します\n")
# 決済の成行注文コードを入れる
records( flag,data )
flag["position"]["exist"] = False
flag["position"]["count"] = 0
flag["records"]["log"].append("さらに" + str(data["close_price"]) + "円で売りの指値注文を入れてドテンします\n")
# ここに売り注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "SELL"
flag["order"]["price"] = round(data["close_price"] * lot)
if flag["position"]["side"] == "SELL":
if signal["side"] == "BUY":
flag["records"]["log"].append("過去{0}足の最高値{1}円を、直近の高値が{2}円でブレイクしました\n".format(term,signal["price"],data["high_price"]))
flag["records"]["log"].append(str(data["close_price"]) + "円あたりで成行注文を出してポジションを決済します\n")
# 決済の成行注文コードを入れる
records( flag,data )
flag["position"]["exist"] = False
flag["position"]["count"] = 0
flag["records"]["log"].append("さらに" + str(data["close_price"]) + "円で買いの指値注文を入れてドテンします\n")
# ここに買い注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "BUY"
flag["order"]["price"] = round(data["close_price"] * lot)
return flag
# 各トレードのパフォーマンスを記録する関数
def records(flag,data):
# 取引手数料等の計算
entry_price = flag["position"]["price"]
exit_price = round(data["close_price"] * lot)
trade_cost = round( exit_price * slippage )
log = "スリッページ・手数料として " + str(trade_cost) + "円を考慮します\n"
flag["records"]["log"].append(log)
flag["records"]["slippage"].append(trade_cost)
# 手仕舞った日時の記録
flag["records"]["date"].append(data["close_time_dt"])
# 値幅の計算
buy_profit = exit_price - entry_price - trade_cost
sell_profit = entry_price - exit_price - trade_cost
# 利益が出てるかの計算
if flag["position"]["side"] == "BUY":
flag["records"]["buy-count"] += 1
flag["records"]["buy-profit"].append( buy_profit )
flag["records"]["gross-profit"].append( flag["records"]["gross-profit"][-1] + buy_profit )
flag["records"]["buy-return"].append( round( buy_profit / entry_price * 100, 4 ))
flag["records"]["buy-holding-periods"].append( flag["position"]["count"] )
if buy_profit > 0:
flag["records"]["buy-winning"] += 1
log = str(buy_profit) + "円の利益です\n"
flag["records"]["log"].append(log)
else:
log = str(buy_profit) + "円の損失です\n"
flag["records"]["log"].append(log)
if flag["position"]["side"] == "SELL":
flag["records"]["sell-count"] += 1
flag["records"]["sell-profit"].append( sell_profit )
flag["records"]["gross-profit"].append( flag["records"]["gross-profit"][-1] + sell_profit )
flag["records"]["sell-return"].append( round( sell_profit / entry_price * 100, 4 ))
flag["records"]["sell-holding-periods"].append( flag["position"]["count"] )
if sell_profit > 0:
flag["records"]["sell-winning"] += 1
log = str(sell_profit) + "円の利益です\n"
flag["records"]["log"].append(log)
else:
log = str(sell_profit) + "円の損失です\n"
flag["records"]["log"].append(log)
# ドローダウンの計算
drawdown = max(flag["records"]["gross-profit"]) - flag["records"]["gross-profit"][-1]
if drawdown > flag["records"]["drawdown"]:
flag["records"]["drawdown"] = drawdown
return flag
# バックテストの集計用の関数
def backtest(flag):
buy_gross_profit = np.sum(flag["records"]["buy-profit"])
sell_gross_profit = np.sum(flag["records"]["sell-profit"])
print("バックテストの結果")
print("--------------------------")
print("買いエントリの成績")
print("--------------------------")
print("トレード回数 : {}回".format(flag["records"]["buy-count"] ))
print("勝率 : {}％".format(round(flag["records"]["buy-winning"] / flag["records"]["buy-count"] * 100,1)))
print("平均リターン : {}％".format(round(np.average(flag["records"]["buy-return"]),2)))
print("総損益 : {}円".format( np.sum(flag["records"]["buy-profit"]) ))
print("平均保有期間 : {}足分".format( round(np.average(flag["records"]["buy-holding-periods"]),1) ))
print("--------------------------")
print("売りエントリの成績")
print("--------------------------")
print("トレード回数 : {}回".format(flag["records"]["sell-count"] ))
print("勝率 : {}％".format(round(flag["records"]["sell-winning"] / flag["records"]["sell-count"] * 100,1)))
print("平均リターン : {}％".format(round(np.average(flag["records"]["sell-return"]),2)))
print("総損益 : {}円".format( np.sum(flag["records"]["sell-profit"]) ))
print("平均保有期間 : {}足分".format( round(np.average(flag["records"]["sell-holding-periods"]),1) ))
print("--------------------------")
print("総合の成績")
print("--------------------------")
print("最大ドローダウン : {0}円 / {1}％".format(-1 * flag["records"]["drawdown"], -1 * round(flag["records"]["drawdown"]/max(flag["records"]["gross-profit"])*100,1) ))
print("総損益 : {}円".format( flag["records"]["gross-profit"][-1] ))
print("手数料合計 : {}円".format( -1 * np.sum(flag["records"]["slippage"]) ))
# ログファイルの出力
file = open("./{0}-log.txt".format(datetime.now().strftime("%Y-%m-%d-%H-%M")),'wt',encoding='utf-8')
file.writelines(flag["records"]["log"])
# 損益曲線をプロット
del flag["records"]["gross-profit"][0]
date_list = pd.to_datetime( flag["records"]["date"] )
plt.plot( date_list, flag["records"]["gross-profit"] )
plt.xlabel("Date")
plt.ylabel("Balance")
plt.xticks(rotation=50) # X軸の目盛りを50度回転
plt.show()
# ここからメイン処理
# 価格チャートを取得
price = get_price(chart_sec,after=1483228800)
flag = {
"order":{
"exist" : False,
"side" : "",
"price" : 0,
"count" : 0
},
"position":{
"exist" : False,
"side" : "",
"price": 0,
"count":0
},
"records":{
"buy-count": 0,
"buy-winning" : 0,
"buy-return":[],
"buy-profit": [],
"buy-holding-periods":[],
"sell-count": 0,
"sell-winning" : 0,
"sell-return":[],
"sell-profit":[],
"sell-holding-periods":[],
"drawdown": 0,
"date":[],
"gross-profit":[0],
"slippage":[],
"log":[]
}
}
last_data = []
i = 0
while i < len(price):
# ドンチャンの判定に使う過去30足分の安値・高値データを準備する
if len(last_data) < term:
last_data.append(price[i])
flag = log_price(price[i],flag)
time.sleep(wait)
i += 1
continue
data = price[i]
flag = log_price(data,flag)
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 )
# 過去データを30個に保つために先頭を削除
del last_data[0]
last_data.append( data )
i += 1
time.sleep(wait)
print("--------------------------")
print("テスト期間：")
print("開始時点 : " + str(price[0]["close_time_dt"]))
print("終了時点 : " + str(price[-1]["close_time_dt"]))
print(str(len(price)) + "件のローソク足データで検証")
print("--------------------------")
backtest(flag)
はじめまして。いつも参考にさせていただいています。
（とはいえ、僕は実験するばかりで実践しないので、利益はサッパリですが）
一番最初の「１）グラフをpythonで描画する方法」のコードですが、
price_listが文字列のリストになっているので、意図した通りにグラフが表示されないと思います。
整数のリストとすれば、例のグラフ通りになりますね。
小さいことですが、お礼をかねて書かせていただきます。
ご指摘ありがとうございます。
本当ですね、価格が文字列になっていました…！
数字に修正しました、ありがとうございますm(__)m