多次元(多要素)の時系列データ(東証プライムに上場している全企業の株価の推移データ)から、特定の日の【日本郵政】の株価(終値)を予測するための基本的概念は、下記のように考えます。
①多次元(多要素)の時系列データの期間は7営業日分、15営業日分、30営業日分の
3つのデータセットを用意する。
→ここで扱う株価は終値とし、株価掲載サイトで随時更新/公開される。
2023年11月下旬時点で、上場企業数が2000社を超えているので、データの取得は
半自動化(Power Automateで平日は毎日終値を取得)した。
Power AutomateはDesktop型を選択すれば無料で使えるRPAツールです。(windowsのみ)
②上場企業には【カテゴリ】の概念(金融/水産業など)が含まれるが、
本研究ではカテゴリ間の関連は考慮しないこととし、約2000社の株価推移の関係を
LSTMでモデル化し、学習する。(つまり、業種関係なく、各企業毎で重み付けを学習させる)
③データセットの80％の株価推移データを学習用データとし、残りの20％の株価推移データを
テストデータ(検証用データ)として、最終的に、日本郵政の株価の終値の現実との乖離率を
評価する。
このような概念で、LSTMモデルを用いて多変量解析の実験をします。
以下は、具体的なPythonでの実装コードです。
＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝
【実装】
まず、pythonスクリプトの全体を下記に示し、その後に解説を行います。
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from tensorflow.compat.v1 import ConfigProto
from tensorflow.compat.v1 import InteractiveSession
# GPU configuration
config = ConfigProto()
config.gpu_options.allow_growth = True
session = InteractiveSession(config=config)
# データの読み込み
data = pd.read_csv("stock_data.csv", header=0)
# データの前処理
data = data.pivot(index='Date', columns='Name', values='Close')
data = data.dropna()
# データの正規化
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(data)
# トレーニングデータとテストデータの分割
train_size = int(len(scaled_data) * 0.8)
train_data = scaled_data[:train_size, :]
test_data = scaled_data[train_size:, :]
# データの整形
def create_dataset(dataset, time_step=1):
dataX, dataY = [], []
for i in range(len(dataset)-time_step-1):
a = dataset[i:(i+time_step), :]
dataX.append(a)
dataY.append(dataset[i + time_step, :])
return np.array(dataX), np.array(dataY)
time_step = 100
X_train, y_train = create_dataset(train_data, time_step)
X_test, y_test = create_dataset(test_data, time_step)
# LSTMモデルの構築
model = Sequential()
model.add(LSTM(units=50, return_sequences=True, input_shape=(X_train.shape[1], X_train.shape[2])))
model.add(LSTM(units=50, return_sequences=True))
model.add(LSTM(units=50))
model.add(Dense(units=data.shape[1]))
model.compile(optimizer='adam', loss='mean_squared_error')
model.fit(X_train, y_train, epochs=1, batch_size=1)
# テストデータでの予測
inputs = scaled_data[len(data) - len(test_data) - time_step:, :]
X_test = np.array([inputs[i-time_step:i, :] for i in range(time_step, len(inputs))])
predicted_stock_prices = model.predict(X_test)
predicted_stock_prices = scaler.inverse_transform(predicted_stock_prices)
# テストデータと予測データの視覚化
company_names = data.columns
test_data_df = pd.DataFrame(test_data, columns=company_names)
predicted_data_df = pd.DataFrame(predicted_stock_prices, columns=company_names)
# A社の予測結果を表示
a_company_name = 'A_company'
a_company_data = pd.DataFrame({'Date': data.index[train_size + time_step + 1:],
'Actual_Close': test_data_df[a_company_name],
'Predicted_Close': predicted_data_df[a_company_name]})
a_company_specific_date = 'A_company_specific_date'
a_company_data_on_date = a_company_data[a_company_data['Date'] == a_company_specific_date]
print(a_company_data_on_date[['Date', 'Actual_Close', 'Predicted_Close']])
次に、上記のスクリプトについて説明していきます。
1.ライブラリのインポート
import numpy as np import pandas as pd from sklearn.preprocessing import MinMaxScaler from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout from tensorflow.compat.v1 import ConfigProto from tensorflow.compat.v1 import InteractiveSession
必要なライブラリをインポートしています。NumPyは数値計算、Pandasはデータ操作、Scikit-learnはデータの正規化、TensorFlowは機械学習のためのライブラリです。
2.GPUの設定
config = ConfigProto() config.gpu_options.allow_growth = True session = InteractiveSession(config=config)
3.データの読み込み:
data = pd.read_csv("stock_data.csv", header=0)
CSVファイルから株価の時系列データを読み込んでいます。
“stock_data.csv”は下記のように整形していることが前提です。
Date,Name,Close
2023-01-01,A_company,100.00
2023-01-01,B_company,50.00
2023-01-02,A_company,105.00
2023-01-02,B_company,52.00
…
Date: 取引日付
Name: 企業名
Close: 終値
4.データの前処理:
data = data.pivot(index='Date', columns='Name', values='Close') data = data.dropna()
データを整形し、NaN値のある行を削除しています。データは「Date」をインデックス、各企業の株価をカラムとして持つようになります。
5.データの正規化:
scaler = MinMaxScaler(feature_range=(0, 1)) scaled_data = scaler.fit_transform(data)
Min-Maxスケーリングを使用して、データを0から1の範囲に正規化しています。
6.トレーニングデータとテストデータの分割:
train_size = int(len(scaled_data) * 0.8) train_data = scaled_data[:train_size, :] test_data = scaled_data[train_size:, :]
データセットをトレーニングデータとテストデータに分割しています。
7.データの整形:
time_step = 100 X_train, y_train = create_dataset(train_data, time_step) X_test, y_test = create_dataset(test_data, time_step)
時系列データをLSTMモデルに適した形に整形しています。create_dataset関数は、過去の株価データを使って現在の株価を予測するためのデータセットを作成します。
8.LSTMモデルの構築:
model = Sequential() model.add(LSTM(units=50, return_sequences=True, input_shape=(X_train.shape[1], X_train.shape[2]))) model.add(LSTM(units=50, return_sequences=True)) model.add(LSTM(units=50)) model.add(Dense(units=data.shape[1])) model.compile(optimizer='adam', loss='mean_squared_error') model.fit(X_train, y_train, epochs=1, batch_size=1)
LSTMモデルを構築しています。3つのLSTMレイヤーと1つの全結合（Dense）レイヤーがあり、平均二乗誤差を損失関数として使用しています。
9.テストデータでの予測:
inputs = scaled_data[len(data) - len(test_data) - time_step:, :] X_test = np.array([inputs[i-time_step:i, :] for i in range(time_step, len(inputs))]) predicted_stock_prices = model.predict(X_test) predicted_stock_prices = scaler.inverse_transform(predicted_stock_prices)
学習済みモデルを使用してテストデータの未来の株価を予測しています。そして、逆変換を行い、予測された株価を元のスケールに戻しています。
10.テストデータと予測データの視覚化:
company_names = data.columns
test_data_df = pd.DataFrame(test_data, columns=company_names)
predicted_data_df = pd.DataFrame(predicted_stock_prices, columns=company_names)
a_company_name = 'A_company'
a_company_data = pd.DataFrame({'Date': data.index[train_size + time_step + 1:],
'Actual_Close': test_data_df[a_company_name],
'Predicted_Close': predicted_data_df[a_company_name]})
a_company_specific_date = 'A_company_specific_date'
a_company_data_on_date = a_company_data[a_company_data['Date'] == a_company_specific_date]
print(a_company_data_on_date[['Date', 'Actual_Close', 'Predicted_Close']])
テストデータと予測データをデータフレームに変換し、特定の企業（A社）に関する予測結果を表示します。