機械学習を用いたバックテスト


機械学習を用いたバックテスト

今回は機械学習の手法を用いてバックテストを行ってみる。


環境


  • CPU Core i7-2600k

  • メモリ DDR3 8GB

  • OS Ubuntu 18.04 LTS 64bit

  • Python 3.6.7

  • scikit-learn 0.20.1


データのダウンロード

今回は2018年のUSDJPYのデータを使うことにする。HistData.comのデータが登録やデモ口座の開設も必要なしにcsvファイルでダウンロードできるので、これを使う。

①以下のアドレスをクリックする。

http://www.histdata.com/download-free-forex-historical-data/?/ascii/1-minute-bar-quotes/usdjpy/2018

②「HISTDATA_COM_ASCII_USDJPY_M12018.zip」をクリックする。

③「HISTDATA_COM_ASCII_USDJPY_M12018.zip を開く」で「ファイルを保存する」を選択する。

④「OK」ボタンをクリックする。

⑤私の環境では「~/ダウンロード」フォルダーに保存される。

⑥端末で以下のコマンドを実行する。

cd ~/ダウンロード; \

unzip HISTDATA_COM_ASCII_USDJPY_M12018.zip


サンプルコード

import numpy as np

import pandas as pd
from collections import OrderedDict
from sklearn import svm

# ローリング・ウィンドウで交差検証を行う関数を定義する。
def rolling_window_cross_val_score(estimator, X, y, n_train, n_test):
n_sample = len(X)
score = 0.0
i = 0
while True:
X_train = X[n_test*i:n_test*i+n_train]
X_test = X[n_test*i+n_train:n_test*(i+1)+n_train]
y_train = y[n_test*i:n_test*i+n_train]
y_test = y[n_test*i+n_train:n_test*(i+1)+n_train]
fit = estimator.fit(X_train, y_train)
score += fit.score(X_test, y_test)
i += 1
if n_test*(i+1)+n_train > n_sample-1:
break
score = score / i
return score

# ダウンロードしたデータは1分足だが、
# 私のPCのスペックではバックテストに時間がかかるので1時間足に変換する。
df = pd.read_csv('~/ダウンロード/DAT_ASCII_USDJPY_M1_2018.csv', sep=';',
names=('Open', 'High', 'Low', 'Close', 'Volume'))
df.index = pd.to_datetime(df.index)
ohlcv_dict = OrderedDict()
ohlcv_dict['Open'] = 'first'
ohlcv_dict['High'] = 'max'
ohlcv_dict['Low'] = 'min'
ohlcv_dict['Close'] = 'last'
ohlcv_dict['Volume'] = 'sum'
df = df.resample('60T', label='left', closed='left').apply(ohlcv_dict)
# 独立変数として前の足の終値が前の足の移動平均より上であれば2、等しければ1、
# 下であれば0とする。
# 移動平均の計算期間は5、10、15、20、25の5種類とした。
# 従属変数として終値が前の足の終値上であれば2、等しければ1、下であれば0とする。
cl = df.Close
cl = cl.dropna()
ma5 = cl.rolling(window=5).mean()
ma10 = cl.rolling(window=10).mean()
ma15 = cl.rolling(window=15).mean()
ma20 = cl.rolling(window=20).mean()
ma25 = cl.rolling(window=25).mean()
x = pd.DataFrame(index=cl.index, columns=['x1', 'x2', 'x3', 'x4', 'x5'])
x.x1[cl.shift()>ma5.shift()] = 2
x.x1[cl.shift()==ma5.shift()] = 1
x.x1[cl.shift()<ma5.shift()] = 0
x.x2[cl.shift()>ma10.shift()] = 2
x.x2[cl.shift()==ma10.shift()] = 1
x.x2[cl.shift()<ma10.shift()] = 0
x.x3[cl.shift()>ma15.shift()] = 2
x.x3[cl.shift()==ma15.shift()] = 1
x.x3[cl.shift()<ma15.shift()] = 0
x.x4[cl.shift()>ma20.shift()] = 2
x.x4[cl.shift()==ma20.shift()] = 1
x.x4[cl.shift()<ma20.shift()] = 0
x.x5[cl.shift()>ma25.shift()] = 2
x.x5[cl.shift()==ma25.shift()] = 1
x.x5[cl.shift()<ma25.shift()] = 0
y = pd.Series(index=cl.index)
y[cl>cl.shift()] = 2
y[cl==cl.shift()] = 1
y[cl<cl.shift()] = 0
x = x[25:]
y = y[25:]
x = np.array(x)
y = np.array(y)
# 分類器としてサポートベクターマシンを使用する。
clf = svm.SVC(gamma='scale')
# 1年を260営業日として訓練期間を半年、
# テスト期間を1ヶ月に設定して正解率を計算する。
# 私のPCだと5分くらいかかる。
n_train = int(260 / 2 * 1440 / 60)
n_test = int(260 / 12 * 1440 / 60)
score = rolling_window_cross_val_score(clf, x, y, n_train, n_test)
print('正解率 = ', score)

正解率 =  0.5211538461538461

正解率は52%くらいかなと予想していたのだが、実際その通りの結果に。

従属変数、独立変数、分類器、訓練期間、テスト期間などを取り替えれば、実際に使えるモデルを作れるかも。