バイナリーオプションを機械学習(CatBoostClassifier)で攻略！

はじめに

基本的な方針は以下のページを参照してください。

機械学習でバイナリーオプションで勝つための指針

バイナリーオプション取引で利用するFX会社

以下の2社を想定しています。

• GMOクリック証券：外為オプション
• ＦＸプライムｂｙＧＭＯ：選べる外為オプション

理由は、上記2社の終了時間が、1時間ずつズレていますので、毎時間トレードすることが可能になるからです。
詳しくは以下を参照してください。

バイナリーオプション（外為オプション）の各社の取引概要・ルールの比較

今回使用したモデル

CatBoostClassifier

• 決定木の勾配ブースティングによるアンサンブル学習して、分類に分けます。
• CatBoostを使用したのは複数のモデルで軽くためしたが、確率が良かったからです。

予想の評価

今回は精度 (Precision) で評価しようと思います。理由は以下のとおりです。

1. ローリングして予想するため、分類したデータに偏りが発生する可能性があります。
2. 正解率 (Accuracy) の場合、偏りが発生すると正確な確率を取得することができません。
3. 精度 (Precision) であれば、正と判断したものの正解率になりますので、偏りが発生しても正解率 (Accuracy) よりも正確なのではないかと考えました。

トレードの方針

上記の予想の評価方法で精度 (Precision) を採用しますので、「高くなる」と「低くなる」のどちらか一方しかトレードすることができません。

今回のバイナリーオプションは、「高くなる」だけを予想してトレードすることにします。

目標

以下で説明していますが、予想精度が６０％以上であれば、バイナリーオプション取引の判断を機械学習に任せる価値があると思います。

機械学習でバイナリーオプションで勝つための指針

バイナリーオプション攻略のアルゴリズムの作成

ライブラリとデータの読み込み

import pandas as pd
import numpy as np

from sklearn.metrics import accuracy_score
from sklearn.metrics import precision_score

drive_dir = 'drive/My Drive/python/'
rates_file = drive_dir + "USD_JPY_M1-2.csv"
df_temp  = pd.read_csv(rates_file)
df_temp.head(3)

今回用意した過去データは、OANDA Japanから取得したもので、1分足の米ドル円になります。

データの前処理

# データをコピー
df = df_temp[df_temp["DateTime"] > "2020-04-01 22:00:00+00:00"].copy()

# 日時データの型を変換
df['DateTime']  = pd.to_datetime(df['DateTime'])

# 時と分の列を作成
df['hour'] = df['DateTime'].dt.hour
df['minute'] = df['DateTime'].dt.minute

# 不要な列を削除
del df['Open']
del df['High']
del df['Low']
del df['Volume']

# 予想用のデータ
OUTPUT_LEN = 60
colms = []
for i in range(0,OUTPUT_LEN + 1):
    colms.append(str(i))
    df[str(i)] = df["Close"].shift(i * -1)
    
# 正解ラベル
df["end"] = df["Close"].shift((OUTPUT_LEN + 30 - 1) * -1)
mask = df[str(OUTPUT_LEN)] < df["end"]
df.loc[mask, "target"] = 1
df.loc[~mask, "target"] = 0

# 不要な行を削除
df = df.drop(df[df.minute != 30].index)

# trade_time = [8,10,12,14,16,18,20,22,0,2]

for i in range(0, 24):
    if i >3 and i < 8:
    # if i not in trade_time:
        df = df.drop(df[df.hour == i].index)

# 不要な列を削除
del df["Close"]
del df["minute"]
del df["end"]

# インデックスをリセット
df = df.reset_index(drop=True)

df.head(10)

予想用のデータ

現在の時間を含め、以降６０個（分）のデータとしています。

正解ラベル

６０個（分）に２９個（分）を追加した行の終値を正解ラベルとしています。
要は５９分の終値をセットできるようにしています。

この５９分の終値と予想データの最後の値を比べて、上がっていれば１（正）とし、それ以外は０（負）としています。

不要な行を削除

分が３０以外のものは削除しています。
今回は、６０分間のデータで３０分後の毎時５９分の終値を予想するためです。

まとめ

• 例えば、８時開始の場合は以下になります。
• 8:30から9:29の60個分を予想データとする。
• 8:30から60個（分）＋30個（分）ー1個をシフトした9:59の終値を正解ラベルの元データとする。
• 9:59の終値と9:29の終値を比較して、上がっていれば正、下がっていれば負とする。

モデルの設定

from catboost import CatBoostClassifier
default_params = {
    'iterations'          : 20,
    'eval_metric'         : 'Precision',
    'l2_leaf_reg'         : 0.3,
    'bagging_temperature' : None,
    'use_best_model'      : False,
    'depth'               : 16,
    'random_strength'     : 1,
    'max_leaves'          : 31,
    'silent'              : True,
    'task_type'         : 'GPU'
}
best_params = {'learning_rate': 0.030416903277478133, 'od_type': 'Iter', 'od_wait': 25}
params = default_params
params.update(best_params)
clf = CatBoostClassifier(**params)

余りちゃんと調整していません。適当です。以前、FXで使った比較的調子が良かった値をパラメーターにセットしています。

評価用の関数

def scores(y_pred_proba,y_data):

    # しきい値 0.5 で 0, 1 に丸める
    y_pred          = np.where(y_pred_proba > 0.5, 1, 0)
    y_pred_pd       = pd.DataFrame({'y_pred': y_pred})
    y_pred_count    = y_pred_pd[y_pred_pd.y_pred > 0].y_pred.count()

    # 精度を検証する
    a_score = accuracy_score(y_data, y_pred)
    p_score = precision_score(y_data, y_pred)
    r_score = recall_score(y_data, y_pred)
    f_score = f1_score(y_data, y_pred)
    
    acc = 'y_pred_count\t:{}'.format(y_pred_count) + '\n'
    acc += 'Accuracy\t:{}'.format(a_score) + '\n'
    acc += 'Precision\t:{}'.format(p_score) + '\n'
    acc += 'Recall\t\t:{}'.format(r_score) + '\n'
    acc += 'f1\t\t:{}'.format(f_score) + '\n'

    print(acc)

    return p_score

ゴミがありますが無視してください。
今回は、分類ですのでしきい値の調整は不要ですが、あってもインプットが０か１ですので論理的にバグりません。

ローリング開始

# ローリング
rows      = df.shape[0]
df_temp2 = pd.DataFrame()

for index, row in df.iterrows():
    if row["hour"] == 8:
        train_start = index
        train_end   = train_start + 400
        test_start  = train_end
        test_end    = test_start + 21

        if rows < test_end:
            break

        print(train_start, train_end)
        print(test_start, test_end)

        # モデル用データ
        df_temp = df[train_start:train_end].copy()
        y_train     = df_temp['target'].values
        del df_temp['target']
        del df_temp['DateTime']
        X_train     = df_temp.values

        # テスト用データ
        df_temp = df[test_start:test_end].copy()
        y_test      = df_temp['target'].values
        del df_temp['target']
        del df_temp['DateTime']
        X_test      = df_temp.values
        
        # 学習する
        clf.fit(X_train, y_train)
        #---------------------------------------------------------------------------------------
        # 予想する
        y_pred      = clf.predict(X_test) 
        print(len(y_pred), len(y_test))
        # 精度を検証する
        pred = scores(y_pred, y_test)
        print(index, pred)
        #--------------------------------------------------------------------------------------- 
        # 結果データを作成
        df_pred = pd.DataFrame(X_test)
        df_pred["y_test"] = y_test
        df_pred["y_pred"] = y_pred

        mask_win  = (df_pred["y_pred"] == 1) & (df_pred["y_test"] == 1)
        mask_lose = (df_pred["y_pred"] == 1) & (df_pred["y_test"] == 0)

        df_pred.loc[mask_win, "game"] = 1
        df_pred.loc[mask_lose, "game"] = 0
        
        df_temp2[index] = df_pred["game"].copy()
print(df_temp2)

# 結果を集計
df_temp3 = df_temp2.T.copy()
for i in range(0,20):
    print(i, df_temp3[df_temp3[i].isna() == False][i].mean())

主な内容は以下の通りです。

1. 日々のローリング
   1. 訓練/テストデータへ分割する。
   2. 学習する。
   3. 予想する。
   4. 精度を検証する＊今回は利用していません。
   5. 予想した結果をデータフレームに保存する。
2. 結果を集計

日々のローリング

毎日８時にモデルを作成して、１日分である２０個を予想します。

訓練/テストデータへ分割

• 件数
  • 訓練データは約１ヶ月分（２０日分）の４００件です。
  • テストデータは１日分にあたる２０件です。
• 内容
  • 時間帯
  • 終値

予想した結果をデータフレームに保存

正解データと予想データを付け合わせして、正解していれば１を、誤っていれば０を、それ以外はNaNのままにしています。

結果の集計

1. 上記のような結果の軸を反転させます。
2. 列ごと（時間帯別）にNaN以外（isna() == False）の値の平均を取得しています。

予想の結果

時間帯別に結果を出力しました。

全然、駄目ですね。。。むしろ逆張りしたほうがいいかも。

改善案

改善できそうな箇所は以下のとおりです。

• CatBoostClassifierのパラメーターを見直す。optunaなどを利用する。
• 訓練の期間を調整する。
• ３０分後を予想しているが、調整する。
• 予想するデータを終値ではなく、もう少し有効なデータにする。
  • 曜日も大きな要素かも。

終値だけでパターンを見つけてほしいのだけど。。。無理ですかね。
でも人間で見つけられないパターンを見つけるのがAIのはずなんですが。

あと、期間は短くしたほうが、パターンが少なるなる可能性もあり、逆にパターンがあっても結果と結びつかなくなるという可能性もあります。

極端な話、３０分足で、次は陽線か陰線なのかを当てるようなものですから、結局はコイントスになるのでしょうか。

もう少し、修行を積みます。