今回からもう少しコンパクトなタイムフレームにしてみました。
import numpy as np
import pandas as pd
import indicators as ind #indicators.pyのインポート
from backtest import Backtest,BacktestReport
dataM1 = pd.read_csv('DAT_ASCII_GBPUSD_M1_201705.csv', sep=';',
names=('Time','Open','High','Low','Close', ''),
index_col='Time', parse_dates=True)
dataM1.index += pd.offsets.Hour(7) #7時間のオフセット
ohlc = ind.TF_ohlc(dataM1, 'H') #１時間足データの作成
前回の最適化では、以下のように二つのパラメータをある範囲で変化させて最適化しました。 このままだと1066通りしかないですが、組み合わせの数は後で変えてみます。
SlowMAperiod = np.arange(10, 61) #長期移動平均期間の範囲
FastMAperiod = np.arange(5, 31) #短期移動平均期間の範囲
単純なランダムサーチだと、パラメータの値を順次ランダムに発生させて、バックテストの評価の最も高いものを採用すればいいのですが、ここでは、遺伝的アルゴリズムにも応用できるよう、１回の世代で20個ほどのパラメータの組み合わせの試行を行い、それを何世代か繰り返す方法を取ってみました。
def Optimize(ohlc, SlowMAperiod, FastMAperiod):
def shift(x, n=1): return np.concatenate((np.zeros(n), x[:-n])) #シフト関数
SlowMA = np.empty([len(SlowMAperiod), len(ohlc)]) #長期移動平均
for i in range(len(SlowMAperiod)):
SlowMA[i] = ind.iMA(ohlc, SlowMAperiod[i])
FastMA = np.empty([len(FastMAperiod), len(ohlc)]) #短期移動平均
for i in range(len(FastMAperiod)):
FastMA[i] = ind.iMA(ohlc, FastMAperiod[i])
M = 20 #個体数
Eval = np.zeros([M, 6]) #評価項目
Param = np.zeros([M, 2], dtype=int) #パラメータ
RandomSearch(Param, Eval[:,0], SlowMAperiod, FastMAperiod) #パラメータ初期化
gens = 0 #世代数
while gens < 100:
for k in range(M):
i = Param[k,0]
j = Param[k,1]
#買いエントリーシグナル
BuyEntry = (FastMA[j] > SlowMA[i]) & (shift(FastMA[j]) <= shift(SlowMA[i]))
#売りエントリーシグナル
SellEntry = (FastMA[j] < SlowMA[i]) & (shift(FastMA[j]) >= shift(SlowMA[i]))
#買いエグジットシグナル
BuyExit = SellEntry.copy()
#売りエグジットシグナル
SellExit = BuyEntry.copy()
#バックテスト
Trade, PL = Backtest(ohlc, BuyEntry, SellEntry, BuyExit, SellExit)
Eval[k] = BacktestReport(Trade, PL)
#世代の交代
Param = RandomSearch(Param, Eval[:,0], SlowMAperiod, FastMAperiod)
gens += 1
Slow = SlowMAperiod[Param[:,0]]
Fast = FastMAperiod[Param[:,1]]
return pd.DataFrame({'Slow':Slow, 'Fast':Fast, 'Profit': Eval[:,0], 'Trades':Eval[:,1],
'Average':Eval[:,2],'PF':Eval[:,3], 'MDD':Eval[:,4], 'RF':Eval[:,5]},
columns=['Slow','Fast','Profit','Trades','Average','PF','MDD','RF'])
前回と異なるのは、各世代でパラメータの組み合わせを保存しておく配列Paramを用意した点です。そして、各世代で評価値（ここでは総損益）が最も大きいパラメータのセットを保存（エリート保存）しておき、残りのパラメータをランダムに生成することを繰り返します。
エリート保存付きでパラメータをランダムに生成する関数は次のように書けます。
#エリート保存付きランダムサーチ
def RandomSearch(Param, Eval, SlowMAperiod, FastMAperiod):
Param = Param[np.argsort(Eval)[::-1]] #ソート
NewParam = np.vstack((np.random.randint(len(SlowMAperiod), size=len(Eval)),
np.random.randint(len(FastMAperiod), size=len(Eval)))).T
NewParam[0] = Param[0] #エリート保存
return NewParam
評価値をソートして0番目のパラメータを再度セットするという方法です。 これを実行すると、以下のようになります。
import warnings;warnings.filterwarnings('ignore')
result = Optimize(ohlc, SlowMAperiod, FastMAperiod)
result.sort_values('Profit', ascending=False)
このように最適解は求まっていますが、20個体を100世代分、つまり2000回近く試行しているので、当然といえば当然です。
今回はパラメータが２個しかないので、それほど組み合わせの数は増えないのですが、
SlowMAperiod = np.arange(10, 161) #長期移動平均期間の範囲
FastMAperiod = np.arange(5, 131) #短期移動平均期間の範囲
のように19,026通りに増やしてみます。結果は
def Optimize(ohlc, SlowMAperiod, FastMAperiod):
def shift(x, n=1): return np.concatenate((np.zeros(n), x[:-n])) #シフト関数
SlowMA = np.empty([len(SlowMAperiod), len(ohlc)]) #長期移動平均
for i in range(len(SlowMAperiod)):
SlowMA[i] = ind.iMA(ohlc, SlowMAperiod[i])
FastMA = np.empty([len(FastMAperiod), len(ohlc)]) #短期移動平均
for i in range(len(FastMAperiod)):
FastMA[i] = ind.iMA(ohlc, FastMAperiod[i])
M = 20 #個体数
Eval = np.zeros([M, 6]) #評価項目
Param = np.zeros([M, 2], dtype=int) #パラメータ
RandomSearch(Param, Eval[:,0], SlowMAperiod, FastMAperiod) #パラメータ初期化
gens = 0 #世代数
while gens < 100:
for k in range(M):
i = Param[k,0]
j = Param[k,1]
#買いエントリーシグナル
BuyEntry = (FastMA[j] > SlowMA[i]) & (shift(FastMA[j]) <= shift(SlowMA[i]))
#売りエントリーシグナル
SellEntry = (FastMA[j] < SlowMA[i]) & (shift(FastMA[j]) >= shift(SlowMA[i]))
#買いエグジットシグナル
BuyExit = SellEntry.copy()
#売りエグジットシグナル
SellExit = BuyEntry.copy()
#バックテスト
Trade, PL = Backtest(ohlc, BuyEntry, SellEntry, BuyExit, SellExit)
Eval[k] = BacktestReport(Trade, PL)
#世代の交代
Param = RandomSearch(Param, Eval[:,0], SlowMAperiod, FastMAperiod)
gens += 1
Slow = SlowMAperiod[Param[:,0]]
Fast = FastMAperiod[Param[:,1]]
return pd.DataFrame({'Slow':Slow, 'Fast':Fast, 'Profit': Eval[:,0], 'Trades':Eval[:,1],
'Average':Eval[:,2],'PF':Eval[:,3], 'MDD':Eval[:,4], 'RF':Eval[:,5]},
columns=['Slow','Fast','Profit','Trades','Average','PF','MDD','RF'])
#エリート保存付きランダムサーチ
def RandomSearch(Param, Eval, SlowMAperiod, FastMAperiod):
Param = Param[np.argsort(Eval)[::-1]] #ソート
NewParam = np.vstack((np.random.randint(len(SlowMAperiod), size=len(Eval)),
np.random.randint(len(FastMAperiod), size=len(Eval)))).T
NewParam[0] = Param[0] #エリート保存
return NewParam
import warnings;warnings.filterwarnings('ignore')
result = Optimize(ohlc, SlowMAperiod, FastMAperiod)
result.sort_values('Profit', ascending=False)
となり、最適解は求まりませんでした。ただ、これを何度か繰り返せば、そのうち、最適解が求まります。
まあ、総当たりでもなんとかできるくらいの組み合わせなら、ランダムサーチでもそれなりの解が求まるということでしょうね。