Yet another 機械学習で株価を予測する (5)

1. 今回の目的

1. Yet another 機械学習で株価を予測する (1)
2. Yet another 機械学習で株価を予測する (2)
3. Yet another 機械学習で株価を予測する (3)
4. Yet another 機械学習で株価を予測する (4)

　これまで3までで日経平均およびNYSE総合 (NYA)の日足データから翌営業日の日経平均が始値から終値にかけて上昇するか下落するか予想するプログラムを作り、交差検証で検証を行いました。利益を期待できそうという結果が得られたので、3ではパラメーターを振ったり、検証期間を変えてみたりして様子を見ました。程度の差こそあれ比較的長期に渡ってそれなりに安定して動作しそうでした。そこで今回は3で作成し、4でパラメーターを調整したプログラムをおおよそ直近の1年間動作させていた場合、利益(あるいは損失)がどれだけ出たのか検討してみることにします。

2. プログラムの改造 (1)

　学習用の期間と検証用の期間を別々に指定するようにします。パラメーターの指定(表示)部分は

dummy01.py

    ###########################################################
    # 読み込む指標を指定する。
    # 先頭の指標の予測をする。
    indices = ["N225", "NYA"]
    
    # 以下のブロックでパラメーターを指定する
    # 学習に使う期間を指定する。
    start_train = '2005-01-01'
    end_train = '2017-08-31'

    # 検証に使う期間を指定する。
    start_test = '2017-09-01'
    end_test = '2018-08-31'

    (中略)

    ##########################################################

    print("Indices: {}".format(indices))
    print("Train period: {} - {}".format(start_train, end_train))
    print("Test period:  {} - {}".format(start_test, end_test))
    print(" threshold: {:.2f} %".format(threshold * 100.0))
    print(" n_estimators: {}".format(n_estimators))

のようにします。この例の場合、2005年初から2017年8月末までのデータで学習して、2017年9月初から2018年8月末までのデータで検証を行います。実際のデータ分割はテーブルの結合を行った後、

dummy02.py

    df = pd.concat([df_dict[index] for index in indices] + [df_ratio], axis=1, join='inner')
    df_train = df[start_train:end_train]
    df_test = df[start_test:end_test]

のように行います。

3. プログラムの改造 (2)

　今回は毎回の予測結果を取り出したいので、最後にテーブルを作って出力するようにします。そのため、　結果格納用の辞書として__pred__を追加します。

dummy03.py

    acc_train = {'positive': [], 'negative': []}
    acc_test  = {'positive': [], 'negative': []}
    pred      = {'positive': [], 'negative': []}
    gain      = {'positive': [], 'negative': []}

　__pred__への結果の格納は

dummy04.py

        pred[polarity] = clf.predict(X_test)

のようにします。

dummy05.py

    df_result = df_test[['Ratio']].copy()
    df_result['pred+'] = pred['positive']
    df_result['pred-'] = pred['negative']
    df_result['gain+'] = gain['positive']
    df_result['gain-'] = gain['negative']
    df_result['gain'] = df_result['gain+'] + df_result['gain-']
    
    df_result.to_csv('result.csv')

結果用のテーブルはテスト用のテーブルの__Ratio__コラムに予測結果、利益を追加していく形で作ります。上昇側の予想は__pred+__、下落側の予想は__pred-__というコラムに収納します。いずれも1が陽性で0なら陰性です。また上昇側の予想に対する利益は__gain+__に、下落側の予想に対する利益は__gain-__に収納します。また__gain__コラムに両側の予想を合わせた利益を収納します。万が一、上昇と下落の予想が同時に陽性になった場合でも、__gain__は両方の和ですので0になります。pandas__to_csv__関数を使って結果をCSVファイルとして書き出します。

dummy06.py

    df_result = df_test[['Ratio']].copy()
    df_result['pred+'] = pred['positive']
    df_result['pred-'] = pred['negative']
    df_result['gain+'] = gain['positive']
    df_result['gain-'] = gain['negative']
    df_result['gain'] = df_result['gain+'] + df_result['gain-']
    
    df_result.to_csv('result.csv')

dummy07.py

    print('   Positive accumulation: {:.3f} %'.format((df_result['gain+']+1).prod() * 100))
    print('   Negative accumulation: {:.3f} %'.format((df_result['gain-']+1).prod() * 100))
    print('   Total accumulation: {:.3f} %'.format((df_result['gain']+1).prod() * 100))

　長期間運用する場合の利益を計算する場合は、足し算(あるいは算術平均)よりは掛け算で通算するほうが実際に即していると考えられます。そこで通算利益は__accumulation__として、毎日の利益を+1したものを掛け合わせて求めます。

4. 結果

　プログラムを走らせた結果の一例です。__RandomForestClassifier__の生成が毎回ランダムに変わりますので結果もランダムに変わります。実は良さそうな結果を選びだしたのですが、それでも1年間の通算でようやく2 %弱の利益にしかなりませんでした。

Indices: ['N225', 'NYA']
Train period: 2005-01-01 - 2017-08-31
Test period:  2017-09-01 - 2018-08-31
 threshold: 0.10 %
 n_estimators: 500
   positive training accuracy: 1.000
   positive test accuracy: 0.550
   positive mean gain: 0.003 %
   negative training accuracy: 1.000
   negative test accuracy: 0.526
   negative mean gain: 0.005 %

   Positive accumulation: 100.645 %
   Negative accumulation: 101.068 %
   Total accumulation: 101.720 %

　実は検証期間をもう少し長めに取ると(学習期間はそれに応じて短くします)、もう少し良い結果になりました。2011年初からの通算ですが、40 %弱の利益が出ています (これは8年8ヶ月の通算ですので、利回りにすると一年あたり5 %弱になります)。

Indices: ['N225', 'NYA']
Train period: 2005-01-01 - 2010-12-31
Test period:  2011-01-01 - 2018-08-31
 threshold: 0.10 %
 n_estimators: 500
   positive training accuracy: 1.000
   positive test accuracy: 0.534
   positive mean gain: 0.005 %
   negative training accuracy: 1.000
   negative test accuracy: 0.545
   negative mean gain: 0.016 %

   Positive accumulation: 106.061 %
   Negative accumulation: 131.604 %
   Total accumulation: 139.673 %

　残念ながら今回の結果でははっきりと利益を期待できる結果にはなりませんでしたが、入力データを変更したりパラメーターを調整すると安定に(毎年利益が出るように)動作してくれる可能性があります。興味のある方はチャレンジしてください。また実際に投資に利用する場合は、あくまでも自己責任でお願いします。

5. ソースコード

Qiita5.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright by troilus (2018)

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier

def read_table(index='N225'):
    csv = index + '.csv'
    df = pd.read_csv(csv, na_values=["null"])
    df["Date"] = pd.to_datetime(df["Date"])
    df = df.set_index("Date")
    df = df[["Open", "High", "Low", "Close"]]
    return df.dropna()

def normalize_table(table, index='N225'):
    suffix_list = ["Open", "High", "Low", "Close"]
    table["Open"] /= table["Close"]
    table["High"] /= table["Close"]
    table["Low"] /= table["Close"]
    table["Close"] = 1.0
    table = table.resample("D").ffill()
    table = table[suffix_list].shift(1)
    table.columns = [index+"_"+suffix for suffix in suffix_list]
    return table


if __name__ == '__main__':
    ###########################################################
    # 読み込む指標を指定する。
    # 先頭の指標の予測をする。
    indices = ["N225", "NYA"]
    
    # 以下のブロックでパラメーターを指定する
    # 学習に使う期間を指定する。
    start_train = '2005-01-01'
    end_train = '2010-12-31'

    # 検証に使う期間を指定する。
    start_test = '2011-01-01'
    end_test = '2018-08-31'

    # 上昇または下落を判定するしきい値
    threshold = 0.001

    # RandomForestのn_estimator
    n_estimators = 500
    ##########################################################

    print("Indices: {}".format(indices))
    print("Train period: {} - {}".format(start_train, end_train))
    print("Test period:  {} - {}".format(start_test, end_test))
    print(" threshold: {:.2f} %".format(threshold * 100.0))
    print(" n_estimators: {}".format(n_estimators))
    
    df_dict = {}
    for index in indices:
        df_dict[index] = read_table(index)
    df_ratio = df_dict[indices[0]]['Close']/df_dict[indices[0]]['Open']
    df_ratio = df_ratio.to_frame("Ratio")
    
    for index in indices:
        df_dict[index] = normalize_table(df_dict[index], index=index)
    df = pd.concat([df_dict[index] for index in indices] + [df_ratio], axis=1, join='inner')
    df_train = df[start_train:end_train]
    df_test = df[start_test:end_test]

    columns_input = [index + suffix for index in indices\
                        for suffix in ["_Open", "_High", "_Low", "_Close"]]

    acc_train = {'positive': [], 'negative': []}
    acc_test  = {'positive': [], 'negative': []}
    pred      = {'positive': [], 'negative': []}
    gain      = {'positive': [], 'negative': []}

    X_train, r_train = df_train[columns_input], df_train["Ratio"]
    X_test,  r_test  = df_test[columns_input], df_test["Ratio"]
    X_train, X_test = X_train.values, X_test.values

    clf = RandomForestClassifier(n_estimators=n_estimators, n_jobs=-1)
    for polarity in ["positive", "negative"]:
        if polarity == "positive":
            y_train = [1 if r - 1.0 >= threshold else 0 for r in r_train]
            y_test = [1 if r - 1.0 >= threshold else 0 for r in r_test]
        else:
            y_train = [1 if 1.0 - r >= threshold else 0 for r in r_train]
            y_test = [1 if 1.0 - r >= threshold else 0 for r in r_test]
            
        clf.fit(X_train, y_train)
        acc_train[polarity] = clf.score(X_train, y_train)
        acc_test[polarity] = clf.score(X_test, y_test)
        pred[polarity] = clf.predict(X_test)
        if polarity == "positive":
            gain[polarity] = pred['positive'] * (r_test - 1)
        else:
            gain[polarity] = pred['negative'] * (1 - r_test)

    for polarity in ["positive", "negative"]:
        print('   {} training accuracy: {:.3f}'.format(polarity, np.array(acc_train[polarity]).mean()))
        print('   {} test accuracy: {:.3f}'.format(polarity, acc_test[polarity]))
        print('   {} mean gain: {:.3f} %'.format(polarity, np.array(gain[polarity]).mean()*100))

    df_result = df_test[['Ratio']].copy()
    df_result['pred+'] = pred['positive']
    df_result['pred-'] = pred['negative']
    df_result['gain+'] = gain['positive']
    df_result['gain-'] = gain['negative']
    df_result['gain'] = df_result['gain+'] + df_result['gain-']
    
    df_result.to_csv('result.csv')
    print('')
    print('   Positive accumulation: {:.3f} %'.format((df_result['gain+']+1).prod() * 100.0))
    print('   Negative accumulation: {:.3f} %'.format((df_result['gain-']+1).prod() * 100.0))
    print('   Total accumulation: {:.3f} %'.format((df_result['gain']+1).prod() * 100.0))