LSTMモデルを用いた株価予測アプリの作成

はじめに

AI学習の成果を残す為の記事です。

学習の成果

1. python(+anaconda)環境構築
2. pythonプログラム実習
3. 各種機械学習モデルの作成
4. flask環境構築
5. LSTMモデルによる株価予測アプリの作成
6. 学習成果をまとめた本記事

作成したアプリの主旨

今回作成したアプリはLSTMモデルを用いた株価予測アプリとなります。
これまでの学習成果をある程度網羅でき、データの入手が容易で、入力に対する出力の変化を目視で確認しやすいという点からこの題材を選びました。
株価予測という観点から学習モデルはLSTMを採択しました。
今回はローカルで動作するアプリを先に作成してから、後でWeb化を行いました。

アプリの作成

1. 必要なライブラリのインポート
2. データセットの抽出
3. データの前処理
4. 訓練データと検証データの分割
5. 訓練データの作成
6. モデル構築
7. 検証データの作成
8. 予測結果のプロット
9. 作成したコード

必要なライブラリのインポート

• flask flaskの利用
• yfinance 株価情報のデータセット取得
• datetime 日付処理
• matplotlib 結果のプロット
• io 画像処理
• numpy 数学処理
• sklearn,keras 機械学習

データセットの抽出

Yahoo Financeを利用して、ユーザ指定の年数分（開始日は1/1）のデータを取得します。
Yahoo Finance APIでユーザ指定の銘柄、開始日、終了日（当日）、intervalに1dを指定する為、終値のみを取得します。（銘柄は数値のみの場合は.tを強制的に付与）

データの前処理

前処理ではデータから終値(Close)のみを取り出しMinMaxScalerで0～1にスケーリングしています。

訓練データと検証データの分割

訓練データと検証データを7:3に分割しました。

訓練データの作成

訓練データは株価を予測する為のデータを直近で絞り込む為、ユーザが指定した日数分を遡った期間のみ取得する形としました。また、データのパディングも行っています。

モデル構築

kerasによるモデル構築を行っています。

検証データの作成

検証データの作成を行い訓練データの加工に合わせた検証用データを作成しています。
また、ここで予測値も算出しています。

予測結果のプロット

予測結果をmatplotlibを用いてグラフ化しています。

作成したコード

# -*- coding: utf-8 -*-

# ---1.必要なライブラリのインポート---
import yfinance as yf
from datetime import datetime
import matplotlib.pyplot as plt

#import os
from keras.models import Sequential
from keras.layers import Dense, LSTM
from keras.models import load_model
from sklearn.metrics import mean_squared_error

import warnings
warnings.filterwarnings('ignore')

# ---標準入力処理関連---
def get_input(prompt, default_value):
    user_input = input(prompt)
    if user_input == '':
        return default_value
    elif default_value is int:
        return int(user_input)
    elif default_value is float :
        return float(user_input)
    else:
        return user_input

from_year = get_input('取得データ年数を入力（デフォルトは3年）: ', 3)
ref_days = get_input('学習日数を入力（デフォルトは60日）: ', 60)
code = get_input('銘柄コードを入力（デフォルトは9861）: ', '9861')
code_dl=code+".t"

# ---2.データセットの抽出---
end_date = datetime.now()  # 直近の日付を計算
start_date = datetime(end_date.year - from_year, 1, 1)  # from_year年前の1月1日を設定
df = yf.download(code_dl, start=start_date, end=end_date, interval="1d")

# ---3.データの前処理---
from sklearn.preprocessing import MinMaxScaler
data = df.filter(["Close"]) ##　Closeコラム（取引終了時の株価）のみ
dataset = data.values
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(dataset)

# ---4.訓練データと検証データの分割---
import numpy as np
training_data_len = int(np.ceil(len(dataset) * 0.7))
train_data = scaled_data[0:int(training_data_len), :]

# ---5.訓練データの作成---
x_train = []
y_train = []
for i in range(ref_days, len(train_data)):
    x_train.append(train_data[i-ref_days:i, 0])
    y_train.append(train_data[i, 0])
max_length = max(len(row) for row in x_train)

x_train_padded = []
for row in x_train:
    if len(row) < max_length:
        row = np.pad(row, (0, max_length - len(row)), 'constant')
    x_train_padded.append(row)

x_train = np.array(x_train_padded)
y_train = np.array(y_train)

x_train = np.reshape(x_train, (x_train.shape[0], x_train.shape[1], 1))

# ---6.モデル構築---
model = Sequential()
model.add(LSTM(128,return_sequences = True, input_shape=(x_train.shape[1], 1)))
model.add(LSTM(64, return_sequences = False))
model.add(Dense(25))
model.add(Dense(1))
model.compile(optimizer='adam',loss='mean_squared_error')
model.fit(x_train, y_train, batch_size = 1, epochs =1)
#savefile = os.path.dirname(__file__) + '/' + "kabuka.h5"
#model.save(savefile)

#savefile = os.path.dirname(__file__) + '/' + "kabuka.h5"
#model = load_model(savefile)

# ---7.検証データの作成---
test_data = scaled_data[training_data_len - ref_days: , :]

x_test = []
y_test = dataset[training_data_len:, :]
for i in range(ref_days, len(test_data)):
    x_test.append(test_data[i-ref_days:i, 0])
x_test = np.array(x_test)
x_test = np.reshape(x_test, (x_test.shape[0], x_test.shape[1], 1))
max_length = max(len(row) for row in x_test)

x_test_padded = []
for row in x_test:
    if len(row) < max_length:
        row = np.pad(row, (0, max_length - len(row)), 'constant')
    x_test_padded.append(row)

x_test = np.array(x_test_padded)
x_test = np.reshape(x_test, (x_test.shape[0], x_test.shape[1], 1))

predictions = model.predict(x_test)
predictions = scaler.inverse_transform(predictions)

test_score = np.sqrt(mean_squared_error(y_test, predictions))

train = data[:training_data_len]
valid = data[training_data_len:]
valid['Predictions'] = predictions

# ---8.予測結果のプロット---
plt.figure(figsize=(16, 6))
plt.title('LSTM Model')
plt.xlabel('Date', fontsize=18)
plt.ylabel('Close Price of Lasertec', fontsize=18)
plt.plot(train['Close'])
plt.plot(valid[['Close', 'Predictions']])
plt.legend(['Train', 'Real', 'Prediction'], loc='lower right')
plt.show()

Webアプリ化

ローカルにて動作を確認した後、flaskによるWebアプリ化のための修正を行いました。
メインのページ(入力フォーム)とplotページ(グラフの画像)の2つを設け、標準入出力によるユーザ設定をフォームからの入力に置き換え、グラフ出力を画像に変換し結果を表示するようにしています。
ここでデプロイの際にPythonバージョンによる問題が生じたため、Pythonのバージョンを3.8.16に落としてrequirements.txtもそれに合わせて修正しました。

from flask import Flask, request, render_template, send_file
import yfinance as yf
from datetime import datetime
import matplotlib.pyplot as plt
import io
import numpy as np
from sklearn.preprocessing import MinMaxScaler
from keras.models import Sequential
from keras.layers import Dense, LSTM
# import os

app = Flask(__name__)

@app.route('/')
def index():
    # ---入力フォーム---
    return render_template('index.html', show_plot=False)

@app.route('/plot', methods=['POST'])
def plot():
    # ---プロット結果---

    # ---入力関連の修正---
    from_year = int(request.form['from_year'])
    ref_days = int(request.form['ref_days'])
    code = request.form['code']
    code_dl = code + ".t"

    end_date = datetime.now()
    start_date = datetime(end_date.year - from_year, 1, 1)
    df = yf.download(code_dl, start=start_date, end=end_date, interval="1d")

    # データの前処理
    data = df.filter(["Close"])
    dataset = data.values
    scaler = MinMaxScaler(feature_range=(0, 1))
    scaled_data = scaler.fit_transform(dataset)

    # 訓練データと検証データの分割
    training_data_len = int(np.ceil(len(dataset) * 0.7))
    train_data = scaled_data[0:int(training_data_len), :]

    # 訓練データの作成
    x_train, y_train = [], []
    for i in range(ref_days, len(train_data)):
        x_train.append(train_data[i-ref_days:i, 0])
        y_train.append(train_data[i, 0])

    x_train = np.array(x_train)
    y_train = np.array(y_train)
    x_train = np.reshape(x_train, (x_train.shape[0], x_train.shape[1], 1))

    # LSTMモデル構築
    model = Sequential()
    model.add(LSTM(128, return_sequences=True, input_shape=(x_train.shape[1], 1)))
    model.add(LSTM(64, return_sequences=False))
    model.add(Dense(25))
    model.add(Dense(1))
    model.compile(optimizer='adam', loss='mean_squared_error')

    # モデルの訓練
    model.fit(x_train, y_train, batch_size=1, epochs=1)

    # モデルの保存(ローカルのみ)
    # savefile = os.path.join(os.path.dirname(__file__), "kabuka.h5")
    # model.save(savefile)

    # 検証用データの作成
    test_data = scaled_data[training_data_len - ref_days:, :]
    x_test, y_test = [], dataset[training_data_len:, :]
    for i in range(ref_days, len(test_data)):
        x_test.append(test_data[i-ref_days:i, 0])
    x_test = np.array(x_test)
    x_test = np.reshape(x_test, (x_test.shape[0], x_test.shape[1], 1))

    # 予測値の算出
    predictions = model.predict(x_test)
    predictions = scaler.inverse_transform(predictions)

    # 予測のプロット
    train = data[:training_data_len]
    valid = data[training_data_len:]
    valid['Predictions'] = predictions

    plt.figure(figsize=(16, 6))
    plt.title('LSTM Model Predictions')
    plt.xlabel('Date', fontsize=18)
    plt.ylabel('Close Price', fontsize=18)
    plt.plot(train['Close'])
    plt.plot(valid[['Close', 'Predictions']])
    plt.legend(['Train', 'Real', 'Prediction'], loc='lower right')

    img = io.BytesIO()
    plt.savefig(img, format='png')
    img.seek(0)
    plt.close()  # プロットを閉じる
    return send_file(img, mimetype='image/png')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, debug=False)

実行結果

• メインページ
  

• プロットページ
  

苦労した箇所

デプロイの際のバージョンによる動作不良の原因調査に一番時間が掛かりました。
これは経験していかないと勘所がなかなか分からないように感じました。

今後の改良点

現状は予測チャートを閲覧するのみなので、予測結果をもとに投資シミュレートできるような機能を設けられればなあと思います。