PythonでFXのランダムウォークを検証

「MT4でのFXのランダムウォークを検証」で実施したランダムウォークの検証をPythonでも実施しました。

MT4でのバックテストだとMQLで記述しなければならず、いろんなパターンを作っていく上で効率が上がらない為、Pythonでバックテストできるように実装しました。作りはじめから結果が完全一致するまで、1日かかりましたが、逆に1日で一通り揃えて、デバッグできてしまうあたり、Pythonはいいなと思いました。

ちなみにPythonでプログラム作るの初めてです。なので、こう書いたほうがよいのでは?というのはコメントでご指導していただけると嬉しいです。これからいくつか作っていって、腕を上げていきたいと思います。

データの準備

データはMT4のヒストリカルデータをCSVにエクスポートしました。方法は下記を参照ください。
PythonでFXのヒストリカルデータ分析 導入編

アルゴの概要

今回は、ランダムウォークの検証のため、毎時0分に常に買い続けるという適当なアルゴです。
「MT4でのFXのランダムウォークを検証」と同じ条件で同じ結果になるかの確認だけです。

Pythonでのプログラムの作り方

私はクラスをたくさん作っていくのが嫌いです。継承していくほうが複雑でロジックを追いにくくなると考えています。そのため、関数でシンプルに作って行きます。必要があればクロージャを駆使します。
Pythonのクロージャでできること

処理の流れ

  1. データをロード
  2. 全ヒストリカルデータをループし、判断ロジックを適用し、売買フラグだけ立てていく
  3. 売買フラグが立った時刻から、売買および、stoplossやtakeprofitなどシミュレーションして勝敗を記録する

ループを繰り返すと処理速度遅くなりやすいですが、シンプルにするために、フェーズを分けて、フェーズごとにヒストリカルデータをぐるっと回す戦法です。
本当は、上記の1.2.3.をデータを読みながら全部実行していけば、ループは1回で済みます。

コード

1. データをロード

データのロードは、下記のリンクと同じです。
PythonでFXのヒストリカルデータ分析 導入編

# %% データロード
# 1週間
#ticks_all = pd.read_csv('./data/EURUSD_20180312_1W.csv', parse_dates=['t'], index_col='t')
# 1年
ticks_all = pd.read_csv('./data/EURUSD_20170101_1Y.csv', parse_dates=['t'], index_col='t')

2. 売買フラグを立てるロジック

呼び出し側

下記の通りで、ヒストリカルデータ、判断情報、アルゴのオプションを保持したクロージャを作って、全データループしていく感じです。将来的にはインジケータも渡すように作りたいと思っています。

# 判断情報の格納先([dict])
judges=[]

# 売買アルゴの初期化
trader = simple_time_trader(ticks_all, judges=judges, options=objdict(minute=0, pos=1))

# 判断ロジック実行
for i in range(len(ticks_all)):
    trader.apply(i)

売買アルゴ

判断ロジックは毎時0分なので、すごいシンプルです。

objdictはdictを拡張して属性アクセスできるようにしたクラスで、使いがってよいのでぜひ下記も合わせて読んでください。
Pythonのディクショナリに属性とアクセスするには

判断結果はクロージャに渡されたjudgesへ{t=時間,pos=買or売}をlistに詰めていく感じで反映しています。
今回は毎時0分というシンプルな条件なので、{t=時間,pos=買or売}だけにしていますが、今後はここに追加の判断材料、例えばSMAの値とか入れていき、あとの結果分析でフィルタにつかえるようにしていきます。

# 判断ロジック
def simple_time_trader(
        # ヒストリカルデータ()
        ticks=DataFrame(),
        # 判断で使うインジケータ(Series)
        indicators=[],
        # 判断情報の格納先
        judges=[],
        # 判断で使うオプション(dict)
        options=objdict(minute=0, pos=1)):

    def _apply_tick(index):     #index=tickの位置

        # 該当データと時刻
        tick = ticks.iloc[index]
        time = tick.name

        # 判断ロジック
        if time.minute == options.minute:
            judges.append(objdict(t=time, pos=options.pos))

    return objdict(apply=_apply_tick)

3.売買シミュレーション

「2. 売買フラグを立てるロジック」では、どの時刻で、買いか売りかを判断しているだけです。
このシミュレーションで、利確や損切り含めた発注したらどうなるかというシミュレーションを作っていきます。

売買アルゴの検討をしたいだけで、勝率だけを追っていくのでロット数や資金の管理はしていません。やりたければ、この結果をもとに追っていけば資金も別で計算できるので。

呼び出し側

次で説明する「シミュレーション処理」を呼ぶだけです。

# 売買実行
simple_order(ticks_all, judges)

シミュレーション処理

この処理は、共通になっていきます。判断アルゴを変えても同じ処理になっていきます。

結果は「2. 売買フラグを立てるロジック」で作成した{t=時間,pos=買or売}へ情報を追加していく形をとっています。

どうしても厳密にやりたくて、浮動少数の丸め誤差の対応を入れてます。本当はDecimalを使ってやるべきかもしれませんが、myroundという有効桁数で切り上げ切り下げする関数作って対応しています。

# 注文実行
def simple_order(
        # ヒストリカルデータ
        ticks=DataFrame(),
        # 判断情報([dict(t,pos)])
        judges=[],
        # 注文で使うオプション(dict)
        options=objdict(stoploss_w=0.0001*10, takeprofit_w=0.0001*10, max_digit=5, spread=0.0001*0.2)):

    # 比較判定で使う関数作成
    myround = lambda x : round(x, options.max_digit)

    # 判断情報でループ
    for j in judges:

        # 注文情報を作る
        tick = ticks.loc[j.t]
        j.open_time  = tick.name
        #スプレッドを考慮
        j.open_price = myround((tick.open+options.spread) if j.pos == 1  else (tick.open-options.spread))
        j.stoploss   = myround(j.open_price - options.stoploss_w * j.pos)
        j.takeprofit = myround(j.open_price + options.takeprofit_w * j.pos)
        # これから作るのはclose_time, close_price, profit

        for index, check_tick in ticks.loc[j.open_time:].iterrows():
            # stoplossのチェック
            if myround(check_tick.low) <= j.stoploss:
                j.close_time = check_tick.name
                j.close_price = j.stoploss
                j.profit = myround(j.close_price - j.open_price)
                break
            # takeprofitのチェック
            if j.takeprofit <= myround(check_tick.high):
                j.close_time = check_tick.name
                j.close_price = j.takeprofit
                j.profit = myround(j.close_price - j.open_price)
                break

検証結果

勝利数を確認します。

# DataFrameに型変換して、NAを除去とインデックス指定
order_history = DataFrame(judges)
order_history = order_history.dropna(subset=['close_time'])
order_history.set_index('t', inplace=True)

# 勝率計算
prof = order_history.profit
sum(prof > 0) / len(prof) #0.5031930571475356が返る

judgeにディクショナリを要素に持つリストを格納してあるので、そのままDataFrameに変換できます。

NAを除去してますが、これは最後のほうのデータで利確、損切り条件に到達しなかった注文が決済時刻close_timeがNaで残るのでそれを除去するためです。

勝率は「MT4でのFXのランダムウォークを検証」の結果とほぼ一緒でした。ほぼというのは、上記のNaを除去したところで差異がでました。MT4のバックテストでは、未決済のものは最後のティックで条件に関係なく決済されてためです。

これで、今後はPythonでバックテスト作っていく自身が付きました。

あとはハマった点をメモしておきます。

検証でハマった点

丸め誤差

価格が少数なので、pandas上でfloat64で扱っています。計算機の10進精度の限界で、通常の演算では影響しない誤差ですが、比較(A>=B)で条件が成立したり、しなかったりしました。本当はDecimalで対応すべきかもしれませんが、とりあえずroundで対応しました。

スプレッド

MT4ではスプレッド0にできず、0.2pipで実験してました。まぁ、考慮漏れなだけですが、最初はスプレッドなしで実験して結果が不一致になり悩みました。少数だと6,000件あまりで十数件だったので、スプレッドだと気づかず、不一致箇所を追うのに苦労しました。

今後

次は、この6000件あまりの取引結果の分析をPythonでどうやれるか、試していきたいと思っています。これこそPythonの活躍する場所かもしれませんね。

あと、このプログラムからForループを取り除くための検討ネタも面白そうですね。取り除くにはどうしたらよいか、取り除いたらどれくらい処理速度があがるのか、に真剣に取り組むことは、スキルアップにつながるので。

検証まとめ記事へのリンク

システムトレードの他のネタを纏めています。参考になれば嬉しいです。

↓↓↓検証まとめページはこちら↓↓↓
【保存版】PythonでのFXのシステムトレード検証

シェアする

  • このエントリーをはてなブックマークに追加

フォローする