仮想通貨bot勉強記録㉕
～ロジックに自動損切機能を追加する～
◆前回までのあらすじ
総当たりテストで、最適パラメータを探せるようになりました。
・閑話休題
いや～ビットコさん落ちましたね。。。
黄色のラインだと思ってたら、青の平行チャネルでしたって感じかな？
2月からのトレンドが継続するなら、5月頭頃に70,000$到達しそうですな。
◆今回やること
・資金管理のため、自動で損切りできるようにする
一回のトレードで許容する最大損失額を予測・コントロールするため、１回のトレードで取れるポジションの量を「1回のトレードで許容できる最大損失額」から自動的に計算できるようにします。
こちらの記事をパク、、、参考にさせて頂いています↓
損切の基準として、ATRを使用します。ATRの説明はビットバンクの記事↓が分かりやすかった。
ATR（Average True Range）は、簡単に言うと、任意の期間の中での値動きの平均です。この値を使って損切りの判断をを行います。
作成コードはこちらです↓
from datetime import datetime
import pybybit
import time
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
#====================API設定====================
apis = [
'プライベートキー',
'シークレットキー'
]
bybit = pybybit.API(*apis, testnet=True)
#===============================================
#====================バックテストの初期設定値====================
lot = 1000 # 1トレードのロット($)
slippage = 0.001 # 手数料やスリッページ（0.075％初期値）
wait = 0 # 待機時間
start = '2019/06/01 09:00' # ローソク足取得開始時刻
get_start = int(datetime.strptime(start, '%Y/%m/%d %H:%M').timestamp()) #タイムスタンプ変換
n = 50 # ローソク足取得リクエスト回数
#/////////////////////////////////////////////////////////////////
stop_range = 1 # ATRの何倍を損切りラインにするか
volatility_term = 14 # ATRを算出するための足の数
#/////////////////////////////////////////////////////////////////
#====================バックテストのパラメーター設定====================
chart_min_list = [ 120,240 ] # テストに使う時間軸(1 3 5 15 30 60 120 240 360 720 "D" "M" "W")
buy_term_list = np.arange(20,40,5) # テストに使う上値ブレイクアウトの期間
sell_term_list = np.arange(20,40,5) # テストに使う下値ブレイクアウトの期間
judge_price_list = [
{"BUY":"close","SELL":"close"}, # ブレイクアウト判定に終値を使用
{"BUY":"high","SELL":"low"} # ブレイクアウト判定に高値・安値を使用
]
#====================APIから価格データ取得(ローソク足の本数指定)====================
def get_price_from_API(chart_min,get_start,n):
price = []
#200*n本のローソク足を取得して、price[]に入れる
for o in range(n):
#pybybitでローソク足取得
k = bybit.rest.inverse.public_kline_list(
symbol = "BTCUSD",
interval= chart_min,
from_ = get_start
).json()
#priceに取得したデータを入れる
price += k["result"]
#200本x足の長さ分だけタイムスタンプを進める
if chart_min =="D":
get_start += 200*60*1440
else:
get_start += 200*60*chart_min
get_start = int(datetime.strptime(start, '%Y/%m/%d %H:%M').timestamp())
return price
#====================パラメータぶんのローソク足をリスト化する====================
def get_price_amount(chart_min_list):
price_list = {} #ローソク足を入れる変数
for chart_min in chart_min_list: #for文(chart_min_listの数だけ処理を行う)
print("{0}分足取得中".format([chart_min]))
price_list[chart_min] = get_price_from_API(chart_min,get_start,n) #chart_min分足のローソク足取得リクエストをn回行う
return price_list
#====================ファイルから価格データを読み込む====================
"""
def get_price_from_file(path):
file = open(path,'r',encoding='utf-8')
price = json.load(file)
print(price)
return price
"""
#====================時間と高値・安値をログに記録====================
def log_price( data,flag ):
log = "時間： " + datetime.fromtimestamp(data["open_time"]).strftime('%Y/%m/%d %H:%M') + " 高値： " + str(data["high"]) + " 安値： " + str(data["low"]) + " 終値： " + str(data["close"]) + "\n"
flag["records"]["log"].append(log)
return flag
#/////////////////////////////////////////////////////////////////////////////
# 期間の平均ボラティリティを計算する関数
def calculate_volatility( last_data ):
high_sum = sum(float(i["high"]) for i in last_data[-1 * volatility_term :])
low_sum = sum(float(i["low"]) for i in last_data[-1 * volatility_term :])
volatility = round((high_sum - low_sum) / volatility_term)
return volatility
#/////////////////////////////////////////////////////////////////////////////
#====================ロジック判定====================
def donchian( data,last_data,buy_term,sell_term,judge_price ):
highest = max(i["high"] for i in last_data[(-1*buy_term):])
if data[ judge_price["BUY"]] > highest:
return {"side":"BUY","price":highest}
lowest = min(i["low"] for i in last_data[(-1*sell_term):])
if data[judge_price["SELL"]] < lowest:
return {"side":"SELL","price":lowest}
return {"side" : None , "price":0}
#====================買い・売り注文====================
def entry_signal( data,last_data,flag,buy_term,sell_term,judge_price ):
signal = donchian( data,last_data,buy_term,sell_term,judge_price )
if signal["side"] == "BUY":
flag["records"]["log"].append("過去{0}足の最高値{1}$を、直近の価格が{2}$でブレイクしました\n".format(buy_term,signal["price"],data[judge_price["BUY"]]))
flag["records"]["log"].append(data["close"] + "$で買いの指値注文を出します\n")
# ここに買い注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "BUY"
flag["order"]["price"] = float(data["close"])
#///////////////////////////////////////////////////////
flag["order"]["ATR"] = calculate_volatility( last_data )
#///////////////////////////////////////////////////////
if signal["side"] == "SELL":
flag["records"]["log"].append("過去{0}足の最高値{1}$を、直近の価格が{2}$でブレイクしました\n".format(sell_term,signal["price"],data[judge_price["SELL"]]))
flag["records"]["log"].append(data["close"] + "$で買いの指値注文を出します\n")
# ここに売り注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "SELL"
flag["order"]["price"] = float(data["close"])
#///////////////////////////////////////////////////////
flag["order"]["ATR"] = calculate_volatility( last_data )
#///////////////////////////////////////////////////////
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"]
#/////////////////////////////////////////////
flag["position"]["ATR"] = flag["order"]["ATR"]
#/////////////////////////////////////////////
return flag
#////////////////////////////////////////////////
#====================損切確認====================
def stop_position( data,flag,last_data,chart_min ):
if flag["position"]["side"] == "BUY":
stop_price = float(flag["position"]["price"]) - flag["position"]["ATR"] * stop_range
if float(data["low"]) < stop_price:
flag["records"]["log"].append("{0}$の損切ラインに引っかかりました。\n".format( stop_price ))
flag["records"]["log"].append(str(round(stop_price)) + "$あたりで成行注文を出してポジションを決済します\n")
# 決済の成行注文コードを入れる
records( flag,data,stop_price,"STOP" )
flag["position"]["exist"] = False
flag["position"]["count"] = 0
if flag["position"]["side"] == "SELL":
stop_price = float(flag["position"]["price"]) + flag["position"]["ATR"] * stop_range
if float(data["high"]) > stop_price:
flag["records"]["log"].append("{0}$の損切ラインに引っかかりました。\n".format( stop_price ))
flag["records"]["log"].append(str(round(stop_price)) + "$あたりで成行注文を出してポジションを決済します\n")
# 決済の成行注文コードを入れる
records( flag,data,stop_price,"STOP" )
flag["position"]["exist"] = False
flag["position"]["count"] = 0
return flag
#////////////////////////////////////////////////
#====================成行決済&ドテン注文====================
def close_position( data,last_data,flag,buy_term,sell_term,judge_price ):
#/////////////////////////////////////
#既にに損切りに掛かっていたら何もしない
if flag["position"]["exist"] == False:
return flag
#/////////////////////////////////////
flag["position"]["count"] += 1
signal = donchian( data,last_data,buy_term,sell_term,judge_price )
if flag["position"]["side"] == "BUY":
if signal["side"] == "SELL":
flag["records"]["log"].append("過去{0}足の最安値{1}$を、直近の価格が{2}$でブレイクしました\n".format(sell_term,signal["price"],data[judge_price["SELL"]]))
flag["records"]["log"].append(data["close"] + "$あたりで成行注文を出してポジションを決済します\n")
# 成行決済注文コードを入れる
records( flag,data,data["close"] )
flag["position"]["exist"] = False
flag["position"]["count"] = 0
flag["records"]["log"].append("さらに" + data["close"] + "$で売りの指値注文を入れてドテンします\n")
# 売り指値注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "SELL"
flag["order"]["price"] = data["close"]
if flag["position"]["side"] == "SELL":
if signal["side"] == "BUY":
flag["records"]["log"].append("過去{0}足の最高値{1}$を、直近の価格が{2}$でブレイクしました\n".format(buy_term,signal["price"],data[judge_price["BUY"]]))
flag["records"]["log"].append(data["close"] + "$あたりで成行注文を出してポジションを決済します\n")
# 成行決済注文コードを入れる
records( flag,data,data["close"] )
flag["position"]["exist"] = False
flag["position"]["count"] = 0
flag["records"]["log"].append("さらに" + data["close"] + "$で買いの指値注文を入れてドテンします\n")
# 買い指値注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "BUY"
flag["order"]["price"] = data["close"]
return flag
#====================トレードパフォーマンス確認====================
def records(flag,data,close_price, close_type=None):
#手数料等の計算
entry_price = float(flag["position"]["price"])
exit_price = float(data["close"])
trade_cost = lot * slippage
flag["records"]["slippage"].append(trade_cost)
flag["records"]["log"].append("スリッページ・手数料として " + str(trade_cost) + "$を考慮します\n")
# 決済日時,ポジションの保有期間を記録
flag["records"]["date"].append(datetime.fromtimestamp(data["open_time"]).strftime('%Y/%m/%d %H:%M'))
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_Price_range = exit_price - entry_price
sell_Price_range = entry_price - exit_price
# 利益率の計算
buy_return = buy_Price_range/entry_price
sell_return = sell_Price_range/entry_price
#利益・損失の確認
if flag["position"]["side"] == "BUY":
flag["records"]["return"].append( buy_return ) #獲得リターンを記録
flag["records"]["side"].append( flag["position"]["side"] ) #買いか売りかを記録
if buy_return > 0:
log = str(buy_return*lot) + "$の利益です\n"
flag["records"]["log"].append(log)
else:
log = str(buy_return*lot) + "$の損失です\n"
flag["records"]["log"].append(log)
if flag["position"]["side"] == "SELL":
flag["records"]["return"].append( sell_return ) #獲得リターンを記録
flag["records"]["side"].append( flag["position"]["side"] ) #買いか売りかを記録
if sell_return > 0:
log = str(sell_return*lot) + "$の利益です\n"
flag["records"]["log"].append(log)
else:
log = str(sell_return*lot) + "$の損失です\n"
flag["records"]["log"].append(log)
return flag
#====================損益曲線をプロット====================
def plot(records,buy_term,sell_term,judge_price,interval):
plt.plot( records.Date, records.Gross ) #X軸、Y軸の値を指定
plt.xlabel("Date") #X軸のラベル名
plt.ylabel("Balance") #Y軸のラベル名
plt.xticks(rotation=50) # X軸の目盛りを50度回転
plt.title("buy_term:{0},sell_term:{1},judge:{2},Interval:{3}".format(buy_term,sell_term,judge_price,interval))
plt.show() #グラフの表示
#====================ファイルを出力====================
def File_output(df,flag):
#file = open("log/donchian-{0}-log.txt".format(datetime.now().strftime("%Y-%m-%d-%H-%M")),'wt',encoding='utf-8')
#file.writelines(flag["records"]["log"])
#pandasのdfをcsvで出力
df.to_csv("log/donchian-{0}-records.csv".format(datetime.now().strftime("%Y-%m-%d-%H-%M")))
#====================バックテストの集計====================
def backtest( flag,buy_term,sell_term,judge_price,interval ):
# 成績を記録したpandas DataFrameを作成
records = pd.DataFrame({
"Date" : pd.to_datetime(flag["records"]["date"]),#決済日時
"Side" : flag["records"]["side"], #ポジションの側
#//////////////////////////////////////////////////////////////////////////
"Stop" : flag["records"]["stop-count"], #損切りを行った回数
#//////////////////////////////////////////////////////////////////////////
"Rate" : flag["records"]["return"], #獲得レート
"Periods" : flag["records"]["holding-periods"], #ポジション保有期間
"Slippage" : flag["records"]["slippage"] #手数料等
})
# 獲得利益の列を追加
records["Profit"] = records.Rate*lot
#////////////////////////////////////////////////
# 連敗回数をカウントする
consecutive_lose = []
defeats = 0
for p in flag["records"]["return"]:
if p < 0:
defeats += 1
else:
consecutive_lose.append( defeats )
defeats = 0
#////////////////////////////////////////////////
# 総利益の列を追加
records["Gross"] = records.Profit.cumsum()
# ドローダウンの列を追加
records["Drawdown"] = records.Gross.cummax().subtract( abs(records.Gross) )
records["DrawdownRate"] = records.Drawdown / records.Gross.cummax() * 100
print("\nバックテスト結果")
print("==============================")
print("\n------------総合成績--------------")
print("全トレード数 : {}回".format(len(records) ))
print("勝率 : {}％".format(round(len(records[records.Profit>0]) / len(records) * 100,1)))
print("平均リターン : {}％".format(round(records.Rate.mean()*100,2)))
print("平均保有期間 : {}足".format(round(records.Periods.mean(),1) ))
#/////////////////////////////////////////////////////////////////
print("損切りの回数 : {}回".format( records.Stop.sum() ))
print("最大連敗回数 : {}回".format( max(consecutive_lose) ))
#/////////////////////////////////////////////////////////////////
print("最大の勝ちトレード : {}$".format((round(records.Profit.max(),2))))
print("最大の負けトレード : {}$".format((round(records.Profit.min(),2))))
print("最大ドローダウン : {0}$ / {1}％".format(round(-1 * records.Drawdown.max()), round( records.DrawdownRate.loc[records.Drawdown.idxmax()] )))
print("利益合計 : {}$".format((round(records[records.Profit>0].Profit.sum(),2))))
print("損失合計 : {}$".format(round(records[records.Profit<0].Profit.sum(),2),))
print("手数料合計 : {}$".format(-1 * records.Slippage.sum() ))
print("最終損益 : {}$".format((round(records.Profit.sum()-(records.Slippage.sum()) ,2))))
print("==============================")
result = {
"Trade-count" : len(records), #トレード回数
"Win-rate" : round(len(records[records.Profit>0]) / len(records) * 100,1),#勝率
"Return-ave" : round(records.Rate.mean(),2), #平均リターン
"DD-rate-max" : -1 * round( records.DrawdownRate.loc[records.Drawdown.idxmax()] ), #最大ドローダウンレート
"Gross" : records.Profit.sum()-(records.Slippage.sum()), #最終損益
"PF" : round( -1 * (records[records.Profit>0].Profit.sum() / records[records.Profit<0].Profit.sum()) ,2)#プロフィットファクター
}
#plot(records,buy_term,sell_term,judge_price,interval)
return result
#====================テスト&集計====================
def aggregate(volatility_term):
# chart_min_listのローソク足リストを取得
price_list = get_price_amount(chart_min_list)
# テストごとの各パラメーターの組み合わせと結果を記録する配列を準備
param = {
"buy_term" : [],
"sell_term" : [],
"chart_min" : [],
"judge_price" : []
}
all_result = {
"count" : [],
"winRate" : [],
"returnRate" : [],
"Drawdown" : [],
"ProfitFactor" : [],
"Gross" : []
}
# 総当たりのためのfor文の準備
combinations = [(chart_min, buy_term, sell_term, judge_price)
for chart_min in chart_min_list
for buy_term in buy_term_list
for sell_term in sell_term_list
for judge_price in judge_price_list]
# 総当たり処理
for chart_min, buy_term, sell_term, judge_price in combinations:
price = price_list[ chart_min ]
last_data = []
need_term = max(buy_term,sell_term,volatility_term)
i = 0
# フラッグ変数の初期化
flag = {
"order":{
"exist" : False,
"side" : "",
"price" : 0,
"count" : 0,
"ATR" : 0
},
"position":{
"exist" : False,
"side" : "",
"price" : 0,
"count" :0,
"ATR" :0
},
"records":{
"date":[],
"return":[],
"side":[],
"stop-count" :[],
"holding-periods":[],
"slippage":[],
"log":[]
}
}
# price全数でバックテストを行う（ローソク足を6000本取得していたら6000回）
while i < len(price):
# ドンチャンの判定に使う期間分の安値・高値データを準備する
if len(last_data) < need_term:
last_data.append(price[i])
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"]:
stop_position( data,flag,last_data,chart_min )
flag = close_position( data,last_data,flag,buy_term,sell_term,judge_price )
else:
flag = entry_signal( data,last_data,flag,buy_term,sell_term,judge_price )
last_data.append( data )
i += 1
time.sleep(wait)
print("テスト期間 ")
print("==============================")
print("開始時点 : " + str(datetime.fromtimestamp(float(price[0]["open_time"]))))
print("終了時点 : " + str(datetime.fromtimestamp(float(price[-1]["open_time"]))))
print("時間足 : {0}".format(chart_min))
print("パラメータ１ : " + str(buy_term) + "期間 / 買い" )
print("パラメータ２ : " + str(sell_term) + "期間 / 売り" )
print("パラメータ３ : " + str(judge_price) + "")
print(str(len(price)) + "件のローソク足データで検証")
print("==============================")
result = backtest( flag,buy_term,sell_term,judge_price,chart_min )
# 今回のループで使ったパラメータの組み合わせを配列に記録する
param["buy_term"].append( buy_term )
param["sell_term"].append( sell_term )
param["chart_min"].append( chart_min )
if judge_price["BUY"] == "high":
param["judge_price"].append( "high/low" )
else:
param["judge_price"].append( "open/close" )
# 今回のループのバックテスト結果を配列に記録する
all_result["count"].append( result["Trade-count"] )
all_result["winRate"].append( result["Win-rate"] )
all_result["returnRate"].append( result["Return-ave" ] )
all_result["Drawdown"].append( result["DD-rate-max"] )
all_result["ProfitFactor"].append( result["PF"] )
all_result["Gross"].append( result["Gross"] )
return param,all_result,flag
#====================表にまとめて、出力====================
def pandas(volatility_term):
param,all_result,flag = aggregate(volatility_term)
# 全てのパラメータによるバックテスト結果をPandasで１つの表にする
df = pd.DataFrame({
"Interval" : param["chart_min"],
"Buy_term" : param["buy_term"],
"Sell_term" : param["sell_term"],
"Judge_price" : param["judge_price"],
"Trade-count" : all_result["count"],
"Win-Rate" : all_result["winRate"],
"Reture-Ave" : all_result["returnRate"],
"DrawDownRate" : all_result["Drawdown"],
"PF" : all_result["ProfitFactor"],
"Gross" : all_result["Gross"]
})
# トレード回数が100に満たない記録は消す
df.drop( df[ df["Trade-count"] < 100].index, inplace=True )
File_output(df,flag)
pandas(volatility_term)前回のコードをいじっているので、変化点を////////で囲ってます。
照れてるわけじゃないよ！///
◆解説
変化点を解説していきまっせ～！
・バックテストの初期値
#====================バックテストの初期設定値====================
lot = 1000 # 1トレードのロット($)
slippage = 0.001 # 手数料やスリッページ（0.075％初期値）
wait = 0 # 待機時間
start = '2019/06/01 09:00' # ローソク足取得開始時刻
get_start = int(datetime.strptime(start, '%Y/%m/%d %H:%M').timestamp()) #タイムスタンプ変換
n = 50 # ローソク足取得リクエスト回数
#/////////////////////////////////////////////////////////////////
stop_range = 1 # 損切りのレンジ幅
volatility_term = 14 # ATRを算出するための足の数
#/////////////////////////////////////////////////////////////////stop_rangeとvolatility_termの項目を追加してます。
・stop_rangeは損切りする際のレンジ幅です。1なら損切り幅がATR*1になります。
・volatility_termはATRを算出するのに過去何本の足を使うかです。
・パラメータ
#====================バックテストのパラメーター設定====================
chart_min_list = [ 120,240 ] # テストに使う時間軸(1 3 5 15 30 60 120 240 360 720 "D" "M" "W")
buy_term_list = np.arange(20,40,5) # テストに使う上値ブレイクアウトの期間
sell_term_list = np.arange(20,40,5) # テストに使う下値ブレイクアウトの期間
judge_price_list = [
{"BUY":"close","SELL":"close"}, # ブレイクアウト判定に終値を使用
{"BUY":"high","SELL":"low"} # ブレイクアウト判定に高値・安値を使用
]buy/sell_term_listをnp.arangeで作成してます。
np.arangeはnp.arange(start,stop,step)です。stepは指定しなかったら1。
今回でいうと”20～40を5間隔で作成”となり、[20,25,30・・・40]というリストを作成してくれます。便利だ。
詳しくはこちら↓
・calculate_volatility( last_data )
#/////////////////////////////////////////////////////////////////////////////
# 期間の平均ボラティリティを計算する関数
def calculate_volatility( last_data ):
high_sum = sum(float(i["high"]) for i in last_data[-1 * volatility_term :])
low_sum = sum(float(i["low"]) for i in last_data[-1 * volatility_term :])
volatility = round((high_sum - low_sum) / volatility_term)
return volatility
#/////////////////////////////////////////////////////////////////////////////ボラティリティを計算する関数を追加しました。
high_sum = sum(float(i["high"]) for i in last_data[-1 * volatility_term :])last_dataの0～volatility_term番目までのhighの合計
volatility = round((high_sum - low_sum) / volatility_term)（高値の合計-安値の合計）/volatility_termで平均を算出
説明めっちゃ適当ですが、要するにATRを算出してます。
・entry_signal( data,last_data,flag,buy_term,sell_term,judge_price )
引数多スギィ！
#====================買い・売り注文====================
def entry_signal( data,last_data,flag,buy_term,sell_term,judge_price ):
signal = donchian( data,last_data,buy_term,sell_term,judge_price )
if signal["side"] == "BUY":
flag["records"]["log"].append("過去{0}足の最高値{1}$を、直近の価格が{2}$でブレイクしました\n".format(buy_term,signal["price"],data[judge_price["BUY"]]))
flag["records"]["log"].append(data["close"] + "$で買いの指値注文を出します\n")
# ここに買い注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "BUY"
flag["order"]["price"] = float(data["close"])
#///////////////////////////////////////////////////////
flag["order"]["ATR"] = calculate_volatility( last_data )
#///////////////////////////////////////////////////////
if signal["side"] == "SELL":
flag["records"]["log"].append("過去{0}足の最高値{1}$を、直近の価格が{2}$でブレイクしました\n".format(sell_term,signal["price"],data[judge_price["SELL"]]))
flag["records"]["log"].append(data["close"] + "$で買いの指値注文を出します\n")
# ここに売り注文のコードを入れる
flag["order"]["exist"] = True
flag["order"]["side"] = "SELL"
flag["order"]["price"] = float(data["close"])
#///////////////////////////////////////////////////////
flag["order"]["ATR"] = calculate_volatility( last_data )
#///////////////////////////////////////////////////////
return flag
#///////////////////////////////////////////////////////
flag["order"]["ATR"] = calculate_volatility( last_data )
#///////////////////////////////////////////////////////flag["order"]に["ATR"]の項目を追加し、
そこにcalculate_volatility( last_data )の戻り値を入れてます。
損切りの判断をするときに使います。
・check_order( 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"]
#/////////////////////////////////////////////
flag["position"]["ATR"] = flag["order"]["ATR"]
#/////////////////////////////////////////////
return flag
#/////////////////////////////////////////////
flag["position"]["ATR"] = flag["order"]["ATR"]
#/////////////////////////////////////////////そのままです。
flag["position"]に["ATR"]の項目を追加し、
flag["order"]["ATR"]を代入します。
・stop_position( data,flag,last_data,chart_min )
今回の目玉！損切りを行う関数です。
#////////////////////////////////////////////////
#====================損切確認====================
def stop_position( data,flag,last_data,chart_min ):
if flag["position"]["side"] == "BUY":
stop_price = float(flag["position"]["price"]) - flag["position"]["ATR"] * stop_range
if float(data["low"]) < stop_price:
flag["records"]["log"].append("{0}$の損切ラインに引っかかりました。\n".format( stop_price ))
flag["records"]["log"].append(str(round(stop_price)) + "$あたりで成行注文を出してポジションを決済します\n")
# 決済の成行注文コードを入れる
records( flag,data,stop_price,"STOP" )
flag["position"]["exist"] = False
flag["position"]["count"] = 0
if flag["position"]["side"] == "SELL":
stop_price = float(flag["position"]["price"]) + flag["position"]["ATR"] * stop_range
if float(data["high"]) > stop_price:
flag["records"]["log"].append("{0}$の損切ラインに引っかかりました。\n".format( stop_price ))
flag["records"]["log"].append(str(round(stop_price)) + "$あたりで成行注文を出してポジションを決済します\n")
# 決済の成行注文コードを入れる
records( flag,data,stop_price,"STOP" )
flag["position"]["exist"] = False
flag["position"]["count"] = 0
return flag
#////////////////////////////////////////////////ポジションを持っているとき、設定している損切りラインに到達していないかを確認し、到達していたら成行決済を行います。
stop_price = float(flag["position"]["price"]) - flag["position"]["ATR"] * stop_range
ポジションの価格から、ATR＊stop_rangeの価格を引いた価格を損切り価格とします。
if float(data["low"]) < stop_price:
flag["records"]["log"].append("{0}$の損切ラインに引っかかりました。\n".format( stop_price ))
flag["records"]["log"].append(str(round(stop_price)) + "$あたりで成行注文を出してポジションを決済します\n")
# 決済の成行注文コードを入れる
records( flag,data,stop_price,"STOP" )
flag["position"]["exist"] = False
flag["position"]["count"] = 0安値が損切りラインより低い場合、成行き決済を行い、flag["position"]の各値をリセットします。
・backtest( flag,buy_term,sell_term,judge_price )
#====================バックテストの集計====================
def backtest( flag,buy_term,sell_term,judge_price ):
# 成績を記録したpandas DataFrameを作成
records = pd.DataFrame({
"Date" : pd.to_datetime(flag["records"]["date"]),#決済日時
"Side" : flag["records"]["side"], #ポジションの側
"Stop" : flag["records"]["stop-count"],
"Rate" : flag["records"]["return"], #獲得レート
"Periods" : flag["records"]["holding-periods"], #ポジション保有期間
"Slippage" : flag["records"]["slippage"] #手数料等
})
# 獲得利益の列を追加
records["Profit"] = records.Rate*lot
#////////////////////////////////////////////////
# 連敗回数をカウントする
consecutive_lose = []
defeats = 0
for p in flag["records"]["return"]:
if p < 0:
defeats += 1
else:
consecutive_lose.append( defeats )
defeats = 0
#////////////////////////////////////////////////
# 総利益の列を追加
records["Gross"] = records.Profit.cumsum()
# ドローダウンの列を追加
records["Drawdown"] = records.Gross.cummax().subtract( abs(records.Gross) )
records["DrawdownRate"] = records.Drawdown / records.Gross.cummax() * 100
print("\nバックテスト結果")
print("==============================")
print("\n------------総合成績--------------")
print("全トレード数 : {}回".format(len(records) ))
print("勝率 : {}％".format(round(len(records[records.Profit>0]) / len(records) * 100,1)))
print("平均リターン : {}％".format(round(records.Rate.mean()*100,2)))
print("平均保有期間 : {}足".format(round(records.Periods.mean(),1) ))
#/////////////////////////////////////////////////////////////////
print("損切りの回数 : {}回".format( records.Stop.sum() ))
print("最大連敗回数 : {}回".format( max(consecutive_lose) ))
#/////////////////////////////////////////////////////////////////
print("最大の勝ちトレード : {}$".format((round(records.Profit.max(),2))))
print("最大の負けトレード : {}$".format((round(records.Profit.min(),2))))
print("最大ドローダウン : {0}$ / {1}％".format(round(-1 * records.Drawdown.max()), round( records.DrawdownRate.loc[records.Drawdown.idxmax()] )))
print("利益合計 : {}$".format((round(records[records.Profit>0].Profit.sum(),2))))
print("損失合計 : {}$".format(round(records[records.Profit<0].Profit.sum(),2),))
print("手数料合計 : {}$".format(-1 * records.Slippage.sum() ))
print("最終損益 : {}$".format((round(records.Profit.sum()-(records.Slippage.sum()) ,2))))
print("==============================")
result = {
"Trade-count" : len(records), #トレード回数
"Win-rate" : round(len(records[records.Profit>0]) / len(records) * 100,1),#勝率
"Return-ave" : round(records.Rate.mean(),2), #平均リターン
"DD-rate-max" : -1 * round( records.DrawdownRate.loc[records.Drawdown.idxmax()] ), #最大ドローダウンレート
"Gross" : records.Profit.sum()-(records.Slippage.sum()), #最終損益
"PF" : round( -1 * (records[records.Profit>0].Profit.sum() / records[records.Profit<0].Profit.sum()) ,2)#プロフィットファクター
}
# plot(records,buy_term,sell_term,judge_price)
return result
#//////////////////////////////////////////////////////////////////////////
"Stop" : flag["records"]["stop-count"], #損切りを行った回数
#//////////////////////////////////////////////////////////////////////////recordsに"Stop"という項目を作成しました。
flag["records"]["stop-count"]の値を入れます。
#////////////////////////////////////////////////
# 連敗回数をカウントする
consecutive_lose = []
defeats = 0
for p in flag["records"]["return"]:
if p < 0:
defeats += 1
else:
consecutive_lose.append( defeats )
defeats = 0
#////////////////////////////////////////////////連敗回数をカウントする部分を作成しました。
①consecutive_loseというリストを作成
②変数：defeatsを作成
③flag["records"]["return"]の値を全て参照し、flag["records"]["return"]がマイナスならdefeatsを+1する
④flag["records"]["return"]がプラスになったらdefeatsの値をリセット
これで、consecutive_loseの中に連敗回数を入れられます。
後でこの中の最大値を参照し、最大連敗回数として表示します。
#/////////////////////////////////////////////////////////////////
print("損切りの回数 : {}回".format( records.Stop.sum() ))
print("最大連敗回数 : {}回".format( max(consecutive_lose) ))
#/////////////////////////////////////////////////////////////////画面表示する部分に、損切り回数・最大連敗回数を追加してます。
解説終わり！
◆実行結果
前回と同じく最適パラメータを探すコードなので、実行結果はほとんど同じです。
前回と今回で、同じパラメータでの実行結果を比較してみます。
パラメータ↓
#====================バックテストのパラメーター設定====================
chart_min_list = [ 120,240 ] # テストに使う時間軸
buy_term_list = np.arange(20,45,5) # テストに使う上値ブレイクアウトの期間
sell_term_list = np.arange(20,45,5) # テストに使う下値ブレイクアウトの期間
judge_price_list = [
{"BUY":"close","SELL":"close"}, # ブレイクアウト判定に終値を使用
{"BUY":"high","SELL":"low"} # ブレイクアウト判定に高値・安値を使用
]
手書きですんません。。。
見づらいかもしれないけど、損切り無しに比べて、ありの方がPF（青列）が大きく上がり、最大ドローダウン（赤列）が小さくなっています。
また、グラフにしてみるとドローダウンの小ささが分かりますな。
これが資金管理のパワーか、、、！
stop_rangeとvolatility_termもパラメータにしてテストをやってみたいので、次回のコードに適用しようかな～
今回はここまで！
この記事が気に入ったらサポートをしてみませんか？