【Python】ナンピン幅をATRに連動させるFX自動売買ツールの開発⑤【MQL5/MQL4】

この度、ナンピン幅および利確幅をATRに連動させて、破綻リスクを確率的に制御するナンピンマーチン型のFX自動売買ツール（EAまたはBOT）の開発を行いました。

この記事は、前回の記事の続きです。

【Python】ナンピン幅をATRに連動させるFX自動売買ツールの開発④【MQL5/MQL4】

この度、ナンピン幅および利確幅をATRに連動させて、破綻リスクを確率的に制御するナンピンマーチン型のFX自動売買ツール（EAまたはBOT）の開発を行いました。この記事は、前回の記事の続きです。 MQL5で開発しMT5で稼働させるEA、MQ...

MQL5で開発しMT5で稼働させるEA、MQL4で開発しMT4で稼働させるEA、Pythonで開発しMT5のPython APIを利用して稼働させるBOT、とそれぞれ用意しているのですが、この記事ではそのEAまたはBOTの開発における思考過程を少しずつ具体的に記事にしていきたいと思います。

なお、作成したMT5用EAとMT4用EAについては、以下からダウンロード可能です。

404 NOT FOUND | ナンピンマーチンEAの研究

nanpin-martin.com

前回まで
ティックデータに1時間足のフラグを追加
勝率を試算する関数
1. buyポジションの勝率計算
2. sellポジションの勝率計算
ティックデータをインプットに勝率計算
まとめ

前回まで

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 y

sellポジションの勝率計算

@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,
        )

おそらくただ実行すると、とても時間がかかります。