PythonでのFXのランダムウォークを検証で作成した、「2017年の1年間分のヒストリカルデータを使って、毎時0分に常に買いのオーダーを入れたら勝率5割になるか？」と検証したプログラムを高速化しました。
高速化にあたって、手法をPythonの高速化のまとめ Ver.1にまとめています。
結果、PythonでのFXのランダムウォークを検証では3分かかっていた1年間分のシミュレーションが、0.1秒で終わるようになりました。なんと、1,500倍はやくなったことになります。こんなに効果があるので、逆に言えば、Pyhtonではちゃんと考えて書かないと、無駄に遅いプログラムを書いてしまいやすいといえます。
過去バージョンのプログラムでの計測
まず、PythonでのFXのランダムウォークを検証のプログラムの計測結果は、
- 売買フラグを立てるロジックは33秒くらい
- 売買シミュレーションは150秒くらい
だったので、合計3分かかっていました。
高速化のポイント
Pythonの高速化のまとめ Ver.1から抜粋。
- Forなどループは書かない。
- シンプルなオブジェクトで扱う
- どうしてもループするのであればnumbaでコンパイル
高速化：売買フラグを立てるロジック
汎用化を考えて、クロージャを作り、ヒストリカルデータをループしていくように作ってましが、汎用化は考えないようにします。とにかくループを回さなくても売買タイミングを計算できるように、各アルゴリズムを実装時に工夫していきます。
例えば、「PythonでのFXのランダムウォークを検証」では毎時0分に、常に買いのポジションを取るだけなので、下記のようにループを回さずともフィルタで計算できます。
buys = ticks.index.minute == 0 #array([ True, False, False, ..., False, False, False])
ticksはDataFrameでインデックスに時刻(datetime64)のシリーズDateTimeIndexになっているので、minuteで分のndarrayが取り出せます。==0で、True or Flaseの配列になり、Trueの位置が分が0ということになります。
このステートメントであれば、0.03秒で処理が終わりました。ループしないことで、33秒が0.03で、約1000倍早くなりました。
今回は毎時0分というシンプルなアルゴなので、上記のようなシンプルなステートメントで実装できたかもしれませんが、移動平均のゴールデンクロス・デッドクロスなどの他のアルゴリズムも上記のようにループを書かずに実装していく工夫をしていきます。とにかく、高速化に向けてはループを書かないよう工夫することが思っています。
もはや汎用化ではないくらいの汎用化ですが、決めごとは、買いや売りのタイミングをBooleanのndarrayで作成するというだけです。後続の注文のシミュレーションではnumbaを使う予定なので、シンプルなオブジェクトで保持するようにしました。
高速化：売買シミュレーション
複雑になってしまったので、ソースは最後に貼ります。
ポイントは、どうしてもループとIF分が必要だったので、numbaでコンパイルする部分を作りました。numbaに外出しするために、DataFrameからndarrayに変換して、それを受け取ってループしていく処理を外出ししています。
結果、0.09秒で処理できるようになりました。もともと150秒かかっていたので、約1500倍速くなりました。
長いですが、参考までにソースを貼り付けておきます。_fast_ordering_simulationがnumbaでコンパイルする関数です。
import numpy as np
from pandas import DataFrame
from numba import jit
"""
"""
def ordering_simulation(ticks, buy_flags, sell_flags, order_options):
# ヒストリカルデータから判定用の最高値、安値を取り出す
ticks_high = ticks.high.values.round(order_options.max_digit)
ticks_low = ticks.low.values.round(order_options.max_digit)
# ヒストリカルデータと売買フラグから注文履歴の初期化
buy_order = _create_order(ticks.open[buy_flags], 1, order_options)
sell_order = _create_order(ticks.open[sell_flags], -1, order_options)
order_history = buy_order.append(sell_order).sort_index()
## close_time, close_price, profitをここから作成
# シミュレーションに必要な情報作成
## 注文の買い(1)or売り(-1)のフラグ
order_pos = order_history.index.get_level_values('pos').values
## 注文の買い(1)or売り(-1)のフラグ
order_open_index = np.arange(len(ticks))[buy_flags | sell_flags]
# シミュレーションの結果を格納するndarrayを作成
## 注文の決済価格
order_close_price = np.zeros(len(order_history), dtype=np.float64)
## 注文の決済の配列の位置
order_close_index = np.full(len(order_history), -1, dtype=np.int64)
# シミュレーション
_fast_ordering_simulation(
ticks_low, ticks_high,
order_history.stoploss.values, order_history.takeprofit.values,
order_open_index,
order_pos,
order_close_index,
order_close_price)
## close_time
order_history['close_time'] = ticks.index[order_close_index]
## close_price
order_history['close_price'] = order_close_price
## close_priceの補正
order_history[order_close_index == -1].close_price = ticks.iloc[-1].close
## profit
order_history['profit'] = ((order_history.close_price - order_history.open_price) * order_pos).round(order_options.max_digit)
return order_history
"""
numbaによる注文シミュレーションの高速化ロジック
"""
@jit('void(float64[:],float64[:],float64[:],float64[:],int32[:],int64[:],int64[:],float64[:])', nopython=True)
def _fast_ordering_simulation(
ticks_low, ticks_high, # [in]ヒストリカルデータの高値、安値
order_stoploss, order_takeprofit, # [in]注文の上限値、下限値
order_open_index, # [in]注文開始のヒストリカルデータの位置
order_pos, # [in]注文の買い(1) or 売り(-1)
order_close_index, # [out]注文決済のヒストリカルデータの位置
order_close): # [out]注文決済の価格
# ヒストリカルデータの数
size_of_ticks = len(ticks_high)
# 注文の数(代表でインデックスのサイズを使ってるだけ)
size_of_order = len(order_open_index)
# 注文でループ
for index_of_order in range(size_of_order):
# ヒストリカルデータを注文位置から後ろにループして決済チェック
for index_of_ticks in range(order_open_index[index_of_order], size_of_ticks):
#買いオーダー
if order_pos[index_of_order] == 1:
# 逆指値 = 損切り
if order_stoploss[index_of_order] >= ticks_low[index_of_ticks]:
order_close_index[index_of_order] = index_of_ticks
order_close[index_of_order] = order_stoploss[index_of_order]
break
# 指値 = 利確
if order_takeprofit[index_of_order] <= ticks_high[index_of_ticks]:
order_close_index[index_of_order] = index_of_ticks
order_close[index_of_order] = order_takeprofit[index_of_order]
break
#売りオーダー
elif order_pos[index_of_order] == -1:
# 逆指値 = 損切り
if order_stoploss[index_of_order] <= ticks_high[index_of_ticks]:
order_close_index[index_of_order] = index_of_ticks
order_close[index_of_order] = order_stoploss[index_of_order]
break
# 指値 = 利確
if order_takeprofit[index_of_order] >= ticks_low[index_of_ticks]:
order_close_index[index_of_order] = index_of_ticks
order_close[index_of_order] = order_takeprofit[index_of_order]
break
"""
注文履歴のデータフレームを作る
"""
def _create_order(ticks_open, pos, options):
order_history = DataFrame(ticks_open)
# インデックス
## open_time
order_history.index.names = ['open_time']
## pos
order_history['pos'] = pos
order_history.set_index('pos', append=True, inplace=True)
# データ
## open_price
order_history['open_price'] = (order_history.open + options.spread * pos).round(options.max_digit)
## stoploss
order_history['stoploss'] = (order_history.open_price - options.stoploss_w * pos).round(options.max_digit)
## takeprofit
order_history['takeprofit'] = (order_history.open_price + options.takeprofit_w * pos).round(options.max_digit)
## close_time, close_price, profitは後で作成
return order_history
検証まとめ記事へのリンク
システムトレードの他のネタを纏めています。参考になれば嬉しいです。
↓↓↓検証まとめページはこちら↓↓↓
【保存版】PythonでのFXのシステムトレード検証
