【Keras, fx】RNNを使って為替の予測をしてみる

機械学習の一手法であるリカレントニューラルネットワークを使ってFXの予測をしてみます。

具体的に本記事でやったことは、

OANDAPYを使って為替データの取得
取得したデータを学習用データに加工
Kerasを使ってモデルの作成、学習、評価

になります。

基本事項
1. OANDA APIとOANDAPY
  1. OANDAとは
  2. oandapyとは
2. Kerasとリカレントニューラルネットワーク
  1. Kerasとは
  2. RNN(リカレントニューラルネットワーク）とは
実装(OANDAPYとKerasを使って)
評価
感想
参考文献

基本事項

OANDA APIとOANDAPY

OANDAとは

OANDAはアメリカのfx会社です。
OANDA Japan

pythonから使えるapiを提供しているので、
pythonでfx関連のことをしようとする人はよく使うと思います。
今回はoandaからデータを取得して、ローソク足チャートをプロットするので
oandaの口座を持っている人前提の記事になります。

apiを使うためにはoandaで口座を作る必要があります。
口座を作るだけなら無料できます。以下のリンクからoandaのページにいき口座を作成してから以下をお読みください。

oandapyとは

oandapyはOANDA APIをpythonから簡単に利用できるようにしたライブラリです。

https://github.com/oanda/oandapy

本記事ではoandapyを使って実装していきます。

インストール方法はシンプルで

pip install git+https://github.com/oanda/oandapy.git

Kerasとリカレントニューラルネットワーク

Kerasとは

Kerasは、ニューラルネットワークを非常にシンプルに構築できるライブラリです。

https://keras.io/ja/

TensorFlow等で書くとかなり長くなってしまうコードがKerasを使うことでシンプルなコードとなります。

RNN(リカレントニューラルネットワーク）とは

RNNとは再帰型ニューラルネットワークとも呼ばれ、
時系列の関係がある入力にも対応できるニューラルネットワークです。

以下はイメージ図、

図からわかるとおり、
現在の入力${\bf x}$１つ前の出力${\bf h}$を用いることで、入力間の関係性も学習させようというアプローチです。

ちなみに図の右のようにすることをアンロールするといいます。
図の中の$h_t$は記憶セルやセルと呼び、１つ前の状態の内容(内部状態)を保存します。
$h_t$は数式で表現すると次のようになります。

$$\begin{aligned}
h_t = f(h_{t-1}, x_t)
\end{aligned}$$

実装(OANDAPYとKerasを使って)

必要なモジュールのimportと定数を設定します。

import os
import oandapy
import pandas as pd
import numpy as np
from keras.models import Sequential
from keras.layers import Dense, Activation, SimpleRNN
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler
from datetime import datetime

# 定数
TARGET = "USD_JPY"  # 通貨ペア
TIME_CONF = "M30"     # 時間足
COUNT = 5000           # count
NB_ITR = 3          # count * NB_ITR 分データを取得

EPOCH = 100      # 学習エポック
TRAIN_RATE = 0.7  # 全データ中の学習用データの割合
BATCH_SIZE = 32

定数からわかる通り、今回はUSD/JPYの30足のデータを用います。

1.データの取得

では、OANDA APIからデータを取得します。
以前の記事で挙げたように5000件以上のデータを取得するテクニックを用いました。

OANDA API V20で5000件以上のデータを取得する

oanda = oandapy.API(environment="live",
                    access_token="your access token")

# 学習・検証用データの取得
end = None
df = None
for i in range(NB_ITR):
    if i == 0 and end is None:
        hist = oanda.get_history(
            instrument=TARGET, granularity=TIME_CONF, count=COUNT)
    else:
        hist = oanda.get_history(
            instrument=TARGET, granularity=TIME_CONF, end=end, count=COUNT)

    hist = hist["candles"]
    end = hist[0]['time']    # [古い→新しい] なので、res[0]を取得
    df_new = pd.DataFrame(hist)
    if df is None:
        df = df_new
    else:
        df = pd.concat([df_new, df])

df = df.set_index(df["time"])

2行目のaccess_tokenは自分のaccess tokenに変更してください。
あとは以前の記事と同じコードになります。

これで、今回使うデータを取得できました。

2.データの成形

次にRNNに学習させるためにデータを加工していきます。
時刻tからinput_len前までのデータを見て時刻t+1の為替のデータを予測させます。

なので、データ先程取得したデータも[t1, t2, t3, t4, t5]→t6となる組み合わせを作成します。

# データの加工
y = np.array(df["closeAsk"].tolist())
sc = MinMaxScaler(feature_range=(0, 1))   # 標準化
y = y.reshape(-1, 1)
y = sc.fit_transform(y)
x = np.arange(len(y))

input_len = 5
X, Y = [], []
for i, _ in enumerate(x):
    if (i+input_len+1 >= len(x)):
        break
    X.append(y[i:i+input_len])
    Y.append(y[i+input_len+1][0])

MinMaxScalerで0〜1に値を変換しています。

加工したデータを学習用とテスト用のデータに分割します。

split_index = int(len(X)*TRAIN_RATE)
# 学習・テスト用データに分割
train_x = X[:split_index]
train_y = Y[:split_index]
test_x = X[split_index:]
test_y = Y[split_index:]

train_x = np.array(train_x).reshape(len(train_x), -1, 1)
test_x = np.array(test_x).reshape(len(test_x), -1, 1)
train_y = np.array(train_y).reshape(len(train_y))

これで、学習とテストに使うデータを作成できました。

3.モデルの構築と学習

RNNのモデルの構築と学習をしていきます。

まずRNNモデルを生成します。

# モデルの構築と学習
input_shape = (len(train_x[0]), 1)
model = Sequential()
model.add(SimpleRNN(100, return_sequences=False, input_shape=input_shape))
model.add(Dense(1))
model.add(Activation("linear"))
model.compile(loss="mse", optimizer="adam")

そして、学習させます。

model.fit(train_x, train_y, batch_size=BATCH_SIZE, epochs=EPOCH, verbose=1)

これで学習することができました。

評価

では、学習し終えたデータを使って評価していきましょう。
テスト用に残しておいたデータを使って評価してみます。

まず、test_xデータでの誤差を見てみます。
evaluateメソッドを使って評価できます。

# モデルの評価
score = model.evaluate(test_x, test_y)
print(score) #=> 9.805023211211208e-05

思ったより誤差が小さくなっています・・・
ほんとに大丈夫なのか？

次に折れ線をプロットして、予測がどんな感じになっているかを見てみます。

まず、正解データのプロットを行います。

# ラベル等の設定
label_x = df["time"].apply(lambda x: datetime.strptime(x, "%Y-%m-%dT%H:%M:%S.%fZ"))
new_format = "%Y-%m-%dT%H:%M"
label_x = label_x.apply(lambda x: x.strftime(new_format)).tolist()
label_x = label_x[len(y)-len(test_y):]

test_y = sc.inverse_transform(np.array(test_y).reshape(-1,1))
test_y = test_y.reshape(len(test_y))

fig, ax = plt.subplots()
ax.plot(label_x, test_y, label="real")

次に予測値を出力し、プロットします。

predict_y = model.predict(test_x)   # 予測値の抽出
predict_y = sc.inverse_transform(predict_y)     # 元の値に戻す

ax.plot(label_x, predict_y, label="predict")

labels = np.array(ax.get_xticklabels())
ticks = np.arange(0,len(labels),2000)
ax.get_xticks()
ax.xaxis.set_ticks(ticks=ticks)
ax.legend()
plt.show()

以下、出力