PyAlgoTrade 処理を分散してシミュレーション

　いままで、細かいところを見ているだけで一通り動くところはチュートリアルで試しただけなので、改めて一式動くようにしてみた。
　まず、パラメータの組み合わせを変えて試すと試行回数が莫大になる。 例えば、移動平均のパラメータだけで考えた場合、１本の移動平均線だと平均期間が2本～100本なら98回試す、エントリーとクローズに別の期間を使うのであれば98x98=9604 回 になる。さらに長期移動平均と短期移動平均を組み合わせると９２０億回になる。組み合わせを掛け算すると爆発的にテスト回数は増えるのだ。 それを、少しでも早く計算するために、複数のコアを持つマシンで処理を分ける方法と、複数のマシンを使って大規模にテストする方法が用意されている。
　複数のマシンは簡単に用意できないので、まずはコアがたくさんあるマシンを使おう


用意するファイルは３つ
・トレードアルゴリズムを書いた mytrade.py
・並列実行させるためのlocalworker.py
・テストに使うデータ fxdata.csv

まず、トレードアルゴリズムを書いたmytrade.pyが以下になる。ポジションを持っていないとき、単純移動平均より上ならロング、下ならショートでポジションを持つ。
ポジションを追っているときは、別の期間での単純移動平均で判定して手放す。
手持ち資金の９０％で買う

__main__の下にメインプログラムを書いておくことでこのプログラム単体でも実行できる。こうしておけばデバッグが簡単になる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import matplotlib
matplotlib.use('Agg')
from pyalgotrade import strategy
from pyalgotrade.technical import ma
from pyalgotrade.technical import cross
from pyalgotrade import plotter
from pyalgotrade.tools import yahoofinance
from pyalgotrade.stratanalyzer import sharpe
from pyalgotrade import bar
from pyalgotrade.barfeed import csvfeed

class MyTrade(strategy.BacktestingStrategy):
    def __init__(self, feed, instrument, entrySMA, exitSMA):
        strategy.BacktestingStrategy.__init__(self, feed)
        self.__instrument = instrument
        # We'll use adjusted close values, if available, instead of regular close values.
        if feed.barsHaveAdjClose():
            self.setUseAdjustedValues(True)
        self.__priceDS = feed[instrument].getPriceDataSeries()
        self.__entrySMA = ma.SMA(self.__priceDS, entrySMA)
        self.__exitSMA = ma.SMA(self.__priceDS, exitSMA)
        self.__longPos = None
        self.__shortPos = None
    #マーケットインするときに使用する単純移動平均
    def getEntrySMA(self):
        return self.__entrySMA

    #マーケットアウトするときに使用する単純移動平均
    def getExitSMA(self):
        return self.__exitSMA

    def onEnterCanceled(self, position): #ポジションを持っていたら売り、買いともに清算
        if self.__longPos == position:
            self.__longPos = None
        elif self.__shortPos == position:
            self.__shortPos = None
        else:
            assert(False)

    def onExitOk(self, position): #上と同内容
        if self.__longPos == position:
            self.__longPos = None
        elif self.__shortPos == position:
            self.__shortPos = None
        else:
            assert(False)

    def onExitCanceled(self, position): #全決済できなければ決済関数を呼ぶ
        # If the exit was canceled, re-submit it.
        position.exitMarket()

    def onBars(self, bars):
        # Wait for enough bars to be available to calculate SMA
        #　各種指標が出るまで待つ
        if self.__exitSMA[-1] is None or self.__entrySMA[-1] is None:
            return

        bar = bars[self.__instrument]
        if self.__longPos is not None:#買いポジがないとき
            if self.exitLongSignal():#終了シグナルあり
                self.__longPos.exitMarket()#買いポジ全決済
        elif self.__shortPos is not None:#売りポジないとき
            if self.exitShortSignal():#終了シグナルあり
                self.__shortPos.exitMarket()#売りポジ全決済strategy
        else:#ノーポジのとき
            if self.enterLongSignal(bar):#買い条件成立時
                #全予算の90%で買う
                shares = int(self.getBroker().getCash() * 0.9 / bars[self.__instrument].getPrice())#
                self.__longPos = self.enterLong(self.__instrument, shares, True)
            elif self.enterShortSignal(bar):#売り条件成立時
                #全予算の90%を使って売る
                shares = int(self.getBroker().getCash() * 0.9 / bars[self.__instrument].getPrice())
                self.__shortPos = self.enterShort(self.__instrument, shares, True)

    def enterLongSignal(self, bar):#買いから入る条件
        #単純移動平均がプラス
        return bar.getPrice() > self.__entrySMA[-1]

    def exitLongSignal(self):#ロングポジションを解消する条件
        #現在価格が単純移動平均とクロスした
        return cross.cross_above(self.__priceDS, self.__exitSMA) and not self.__longPos.exitActive()

    def enterShortSignal(self, bar):
        return bar.getPrice() < self.__entrySMA[-1]

    def exitShortSignal(self):
        return cross.cross_below(self.__priceDS, self.__exitSMA) and not self.__shortPos.exitActive()

if __name__ == "__main__":
    instrument = "USDJPY"
    entrySMA = 59
    exitSMA = 17

    feed = csvfeed.GenericBarFeed(bar.Frequency.DAY)
    feed.addBarsFromCSV("USDJPY","fxdata.csv")

    strat = MyTrade(feed, instrument, entrySMA, exitSMA)

    plt = plotter.StrategyPlotter(strat, True, True, True)
    plt.getInstrumentSubplot(instrument).addDataSeries("Entry SMA", strat.getEntrySMA())
    plt.getInstrumentSubplot(instrument).addDataSeries("Exit SMA", strat.getExitSMA())

    strat.run()

    fig=plt.buildFigure(None,None)
    fig.savefig("output.png")

並列実行させるためのlocalworker.pyは以下のソースになる。
アルゴリズムが変わったときは、local.run の実行プログラムを書き換える。パラメータが変わったときはparameters_generater()関数を変更する。
準備ができたらpython localworker.py で実行する。このトレードアルゴリズムでは調整パラメータが２個しかなく55x55回の試行なのですぐに終わる。

#!/usr/bin/python

import itertools
from pyalgotrade.optimizer import local
from pyalgotrade.barfeed import yahoofeed
from pyalgotrade import bar
from pyalgotrade.barfeed import csvfeed
import mytrade


def parameters_generator():
    instrument = ["USDJPY"]
    entrySMA = range(5, 60)
    exitSMA = range(5, 60)
    return itertools.product(instrument, entrySMA, exitSMA)


# The if __name__ == '__main__' part is necessary if running on Windows.
if __name__ == '__main__':
    # Load the feed from the CSV files.
    feed = csvfeed.GenericBarFeed(bar.Frequency.DAY)
    feed.addBarsFromCSV("USDJPY","fxdata.csv")

    local.run(mytrade.MyTrade, feed, parameters_generator())

データはGMOクリック証券から取得したドル円の１日分のデータを実行可能なCSVに変換したもの。出来高は10000000固定

Date Time,Open,High,Low,Close,Volume
2015-10-12 07:00:00,120.191,120.220,120.186,120.206,10000000
2015-10-12 07:01:00,120.207,120.234,120.204,120.221,10000000
2015-10-12 07:02:00,120.221,120.222,120.187,120.190,10000000
2015-10-12 07:03:00,120.189,120.198,120.176,120.176,10000000
2015-10-12 07:04:00,120.176,120.189,120.174,120.179,10000000
2015-10-12 07:05:00,120.179,120.184,120.176,120.184,10000000

実行結果ではentryが59本、exitが17本がベストであった。
これらのパラメータをmytrade.py に設定して単独実行、python mytrade.py を実行すると結果がoutput.pnｇに吐き出される。


さすが往復ビンタ恐ろしい数の売買をしていて笑える。