この度、ナンピン幅および利確幅をATRに連動させて、破綻リスクを確率的に制御するナンピンマーチン型のFX自動売買ツール（EAまたはBOT）の開発を行いました。
この記事は、前回の記事の続きです。
MQL5で開発しMT5で稼働させるEA、MQL4で開発しMT4で稼働させるEA、Pythonで開発しMT5のPython APIを利用して稼働させるBOT、とそれぞれ用意しているのですが、この記事ではそのEAまたはBOTの開発における思考過程を少しずつ具体的に記事にしていきたいと思います。
前回まで
1分足を使って勝率を計算し、ATRとDMIを考慮せずに勝率を計算した場合と、考慮して勝率を計算した場合で、どのように変化するかを確認しました。
| 利確幅 | DMI | buy 勝率 | sell 勝率 | buy+sell 勝率 |
| 12.00固定 | なし | 52.9% | 45.7% | 49.3% |
| ATR×3.2 | なし | 50.8% | 47.2% | 49.0% |
| 12.00固定 | あり | 54.8% | 47.6% | 51.1% |
| ATR×3.2 | あり | 54.1% | 50.4% | 52.2% |
ベースロジックが思い浮かんだら、このように1分足で勝率を試算してみるというのは有用です。これが良かったら実装しても必ず良い結果が得られるとは限りませんが、少なくともこの結果が悪ければそれ以上ブラッシュアップしてもあまり良いEAにはならなそうということは言えるかと思います。
そして今回は、ティックデータを使ってより詳細なバックテストを行う方法をご紹介します。
ティックデータに1時間足のフラグを追加
前回までの勝率シミュレーションでは、価格変動は1分足を用いて、売買判断は1時間足を用いました。
今回は、価格変動はティックデータを用いて、売買判断は1時間足を用いる方法に変更します。
df_H1にbar_timeを追加
df_H1にbar_timeを追加する方法は、前回までと全く同じです。
start_time = datetime(2017, 1, 1, 0, 0, 0, 0)
df = df_H1
df = df.dropna()
df = df.reset_index()
df['start_time'] = start_time
delta = (df['time'] - df['start_time']).dt
df['bar_time'] = (df['time'] - df['start_time'])//timedelta(minutes=60)
df = df.set_index('time')
df_H1 = df出来上がったdf_H1は以下の通りです。
一つ目のレコードの2017-01-06 09:00:00は、2017-01-01 00:00:00から129時間後なので、bar_timeには129が入っています。以下、1時間（1レコード）毎に1ずつ増えていくような形です。
ティックデータにbar_timeを追加
ティックデータにも同様にbar_timeを追加します。
YYYYMM = '202212'
start_time = datetime(2017, 1, 1, 0, 0, 0, 0)
df = pd.read_pickle('/content/drive/My Drive/backtest/GOLD#_tick_'+YYYYMM+'.pkl')
df = df.reset_index()
df['start_time'] = start_time
delta = (df['time'] - df['start_time']).dt
df['bar_time_H1'] = (df['time'] - df['start_time'])//timedelta(minutes=60)
df = df.set_index('time')
df_tick = df出来上がったdf_tickは以下の通りです。
ティックデータに、1時間足のテクニカル指標を結合
pandasのdfのまま追加するロジックの方がわかりやすかったりしますが、それだと処理が遅くなるので、numbaを用いて処理が速くなるようなロジックにしています。少しわかりにくいかもしれませんが、載せておきます。
@numba.njit
def calc_tick_i(
tick_len,
tick_bid,
tick_bar_H1,
len_bar_H1,
bar_time_H1,
bar_DMI_H1_flg,
bar_ATR_H1,
):
tick_DMI_H1_flg = tick_bar_H1.copy()
tick_DMI_H1_flg[:] = np.nan
tick_ATR_H1 = tick_bid.copy()
tick_ATR_H1[:] = np.nan
if tick_len > 0:
for j in range(len_bar_H1):
if bar_time_H1[j] == tick_bar_H1[0]:
H1_j = j
break
for i in range(tick_len):
for j in range(H1_j,len_bar_H1):
if bar_time_H1[j] == tick_bar_H1[i]:
H1_j = j
tick_DMI_H1_flg[i] = bar_DMI_H1_flg[j-1]
tick_ATR_H1[i] = bar_ATR_H1[j-1]
break
return tick_DMI_H1_flg,tick_ATR_H1そして、このcalc_tick_i関数を用いて、以下の処理を行います。
df = df_tick
df = df.reset_index()
if(len(df)>0):
df['DMI_H1_flg'],df['ATR_H1']= calc_tick_i(
tick_len=len(df),
tick_bid=df['bid'].values,
tick_bar_H1=df['bar_time_H1'].values,
len_bar_H1=len(df_H1),
bar_time_H1=df_H1['bar_time'].values,
bar_DMI_H1_flg=df_H1['DMI_flg'].values,
bar_ATR_H1=df_H1['ATR'].values,
)
df = df.set_index('time')
df_tick_ATR = df結果、以下のようなデータフレーム（df_tick_ATR）が出来上がります。
処理の流れを簡単に解説しておくと、df_tickのtimeから、2017年1月1日の0時を起点とした1時間単位の整数（bar_time_H1）を作成して、df_H1のbar_timeとマッチング、マッチングしたレコードの1つ前のレコードのDMI_flgとATRを取得しています。
勝率を試算する関数
buyポジションの勝率計算
@numba.njit
def calc_buy_tick(bid=None,ask=None,spread=None,lc_range=None,tp_range=None,spread_limit=None):
y = bid.copy()
y[:] = np.nan
for i in range(bid.size):
if spread[i] >= spread_limit:
y[i]=0
else:
for j in range(i + 1, bid.size):
if bid[j] < ask[i] - lc_range[i]:
y[i] = bid[j] - ask[i]
break
if bid[j] > ask[i] + tp_range[i]:
y[i] = bid[j] - ask[i]
break
return ysellポジションの勝率計算
@numba.njit
def calc_sell_tick(bid=None,ask=None,spread=None,lc_range=None,tp_range=None,spread_limit=None):
y = bid.copy()
y[:] = np.nan
for i in range(bid.size):
if spread[i] >= spread_limit:
y[i]=0
else:
for j in range(i + 1, bid.size):
if ask[j] > bid[i] + lc_range[i]:
y[i] = bid[i] - ask[j]
break
if ask[j] < bid[i] - tp_range[i]:
y[i] = bid[i] - ask[j]
break
return yティックデータをインプットに勝率計算
df[‘buy_y’],df[‘sell_y’]の追加
用意したdf_tick_ATRとcalc_buy_result関数、calc_sell_result関数を使って、以下のようにdf[‘buy_y’]とdf[‘sell_y’]を追加します。
from tqdm.notebook import tqdm
spread_limit = 0.15
df = df_tick_ATR
for i in tqdm(range(len(df))):
df['buy_y'] = calc_buy_tick(
bid=df['bid'].values,
ask=df['ask'].values,
spread=df['spread'].values,
lc_range = 3.0*df['ATR_H1'].values,
tp_range = 3.2*df['ATR_H1'].values,
spread_limit=spread_limit,
)
df['sell_y'] = calc_sell_tick(
bid=df['bid'].values,
ask=df['ask'].values,
spread=df['spread'].values,
lc_range = 3.0*df['ATR_H1'].values,
tp_range = 3.2*df['ATR_H1'].values,
spread_limit=spread_limit,
)おそらくただ実行すると、とても時間がかかります。
実行レコードの限定
ひとまずdf = df_tick_ATRの部分を、
df = df_tick_ATR[:8000]と修正して実行してみてください。これは実行する範囲を8000レコードに限定しています。
進捗状況をプログレスバーで確認
Pythonでfor文の処理の進捗状況をプログレスバーで確認するには、tqdmライブラリが有効です。
from tqdm.notebook import tqdm上記インポートの後、
for i in tqdm(range(len(df))):という形で、range部分をtqdm()で括るだけで、以下のような進捗を確認できるバーが表示されるようになります。
まとめ
今回は、ティックデータに1時間足のテクニカル指標を追加して、df[‘buy_y’]とdf[‘sell_y’]を追加する方法をご紹介しました。出力されたyを合計すれば想定損益計算が可能ですし、y>0とy<0で分けてレコード数をカウントすれば、勝率計算も可能です。
ただ、ティックデータは2022年12月分だけでも261万レコードと非常に大きなデータなので、ループ処理にはとても時間がかかります。一気に処理しようとするとColabの接続が途中で切れてしまったりするかもしれません。
それを回避する方法としては、実行途中に10,000レコード毎とか適当な区切りで、結果ファイルを都度Drive上に保存しておいて後で連結するといった方法が考えられます。
本格的に勝率を計算するためには、上記のような処理を入れる必要がありますが、それはまた別の記事にさせていただきます。
例えば、機械学習を取り入れようとすると、インプットデータ（学習データ）は多ければ多いほど良いはずなので、そのあたりの処理（大量データ処理）を効率的に行うことも重要になってきます。
次回は、同様のティックデータを用いて、より実践的なバックテスト、つまり、仮想取引履歴を作成していくような仕組みをご紹介したいと思います。
