LSTMで日経平均を予測するPython実装|前日終値に勝てるか検証

AI×自動売買

※本記事のコードは学習・検証目的です。予測モデルの出力をそのまま発注に繋がないでください。投資判断はご自身の責任で行ってください。

先に結論です。LSTMで日経平均の翌日終値を予測すると、グラフは驚くほど実際の価格に重なります。そして、その予測は「前日終値をそのまま答える」だけのナイーブ予測にほぼ確実に負けます。重なって見えるのは、モデルが1日遅れで価格を追いかけているからです。

これは筆者の主観ではありません。20銘柄・2つのデータセットでLSTM・GRU・Transformerを検証した研究では、いずれもナイーブ基準を安定して上回れず、勝率は10〜35%でした。為替とNASDAQ100で8手法を比較した研究でも、ナイーブ予測を超えたモデルはゼロです。S&P500の20年データでは、ARIMAの方がLSTMより1期先予測で優れていました。

それでもLSTMを実装する価値はあります。「自分のモデルがナイーブに負けている」と数字で確認できる人だけが、次の一歩に進めるからです。本記事では、リークを潰した実装、ナイーブ比較、方向的中率の3点を、そのまま動くコードで示します。

📘 外部参考Keras LSTM(公式)TimeSeriesSplit(scikit-learn公式)yfinance(GitHub)

入門記事のコードに必ず入っている2つのリーク

「LSTM 株価予測」で出てくるコードは、ほぼ同じ形をしています。5年分の終値を取得し、MinMaxScalerで0〜1に正規化し、60日窓を作り、8:2で分割する。この順序に問題があります。

手順よくある実装何が漏れるか
正規化全期間でfit_transform将来の最高値・最安値が学習時に既知になる
窓作成分割前に全期間で窓を作る訓練窓とテスト窓が日付をまたいで重なる
評価MSEだけ見るラグ予測でもMSEは小さくなる
比較対象なしナイーブに負けていても気づけない

特に1番目が致命的です。MinMaxScalerは期間全体の最小値と最大値を使ってスケールします。テスト期間に史上最高値があれば、その情報が訓練データの正規化に混入します。「テストにはtransformだけ」と注意書きしている記事でも、実際のコードがfit_transform(close)のまま、という矛盾は珍しくありません。

lookaheadの考え方はディープラーニングに限りません。移動平均クロスでも同じ事故が起きます(Vibe CodingでBOTを作る方法で検査コードを扱っています)。

環境とデータ取得

pip install yfinance pandas numpy scikit-learn tensorflow

Google Colabならtensorflowは導入済みです。ローカルのWindowsでGPUを使う場合は、TensorFlowのバージョン対応に注意してください。CPUでも、本記事の規模なら数分で終わります。環境の作り方はPython環境構築ガイドにまとめています。

import numpy as np
import pandas as pd
import yfinance as yf

SEED = 42
np.random.seed(SEED)

def load_nikkei(start: str = "2015-01-01") -> pd.Series:
    raw = yf.download("^N225", start=start, auto_adjust=True, progress=False)
    if raw.empty:
        raise RuntimeError("no data")
    if isinstance(raw.columns, pd.MultiIndex):
        raw.columns = raw.columns.get_level_values(0)
    close = raw["Close"].dropna()
    close.index = pd.to_datetime(close.index).tz_localize(None)
    return close.sort_index()

close = load_nikkei()
print(close.shape, close.index.min().date(), close.index.max().date())

日経平均のティッカーは^N225です。auto_adjust=Trueは指数では実質的に影響しませんが、個別株に差し替えたときのために揃えておきます。

リークを潰した前処理

順序を変えます。先に日付で分割し、訓練期間だけでスケーラーをfitし、それから窓を作る。この3行の順番が、記事の価値の半分です。

from sklearn.preprocessing import MinMaxScaler

WINDOW = 60

def make_windows(arr: np.ndarray, window: int = WINDOW):
    X, y = [], []
    for i in range(window, len(arr)):
        X.append(arr[i - window:i, 0])
        y.append(arr[i, 0])
    X = np.array(X).reshape(-1, window, 1)
    return X, np.array(y)

def prepare(close: pd.Series, test_ratio: float = 0.2, window: int = WINDOW):
    values = close.values.reshape(-1, 1)
    split = int(len(values) * (1 - test_ratio))

    # 1. 分割が先
    train_raw, test_raw = values[:split], values[split:]

    # 2. fit は訓練だけ
    scaler = MinMaxScaler()
    train_scaled = scaler.fit_transform(train_raw)
    test_scaled = scaler.transform(test_raw)

    # 3. テスト窓の先頭に必要な過去60日だけを訓練末尾から借りる
    bridge = np.vstack([train_scaled[-window:], test_scaled])

    X_train, y_train = make_windows(train_scaled, window)
    X_test, y_test = make_windows(bridge, window)
    test_index = close.index[split:]
    return X_train, y_train, X_test, y_test, scaler, test_index

X_train, y_train, X_test, y_test, scaler, test_index = prepare(close)
print(X_train.shape, X_test.shape, len(test_index))

bridgeの処理が要点です。テスト期間の初日を予測するには直前60日が必要ですが、その60日は訓練期間の値です。訓練の値を「入力として」使うのは問題ありません。禁止なのは、テスト期間の値を訓練の正規化パラメータやパラメータ更新に使うことです。ここを混同して「テスト期間を一切触らない」と全期間fitに戻ってしまう人が多いです。

スケーラーのfit範囲で結果はどれだけ変わるか

def scaler_gap(close: pd.Series, test_ratio: float = 0.2) -> pd.DataFrame:
    values = close.values.reshape(-1, 1)
    split = int(len(values) * (1 - test_ratio))
    full = MinMaxScaler().fit(values)
    train_only = MinMaxScaler().fit(values[:split])
    return pd.DataFrame([
        {"fit": "all_period", "min": float(full.data_min_[0]), "max": float(full.data_max_[0])},
        {"fit": "train_only", "min": float(train_only.data_min_[0]), "max": float(train_only.data_max_[0])},
    ])

print(scaler_gap(close))

上昇相場の期間を含むと、2行のmaxは大きく違います。差が大きいほど、全期間fitのモデルは「将来この水準まで上がる」というヒントを先にもらっていることになります。テスト成績が良く見えるのは当然です。

モデルとコールバック

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout, Input
from tensorflow.keras.callbacks import EarlyStopping

tf.random.set_seed(SEED)

def build_model(window: int = WINDOW) -> Sequential:
    model = Sequential([
        Input(shape=(window, 1)),
        LSTM(64, return_sequences=True),
        Dropout(0.2),
        LSTM(32),
        Dropout(0.2),
        Dense(1),
    ])
    model.compile(optimizer="adam", loss="mse", metrics=["mae"])
    return model

model = build_model()
es = EarlyStopping(monitor="val_loss", patience=5, restore_best_weights=True)
hist = model.fit(
    X_train, y_train,
    epochs=100,
    batch_size=32,
    validation_split=0.1,  # 訓練の後ろ10%。shuffleしない
    shuffle=False,
    callbacks=[es],
    verbose=0,
)
print("stopped at epoch", len(hist.history["loss"]))

shuffle=Falseを明示します。時系列でシャッフルすると、検証データが訓練データより過去になり、また別のリークになります。Inputレイヤーを使うのは、input_shape引数がKerasの新しめのバージョンで警告になるためです。

ナイーブ予測と並べて評価する

ここが本記事の中心です。予測値を逆変換し、「前日終値をそのまま予測値とする」ナイーブと同じ土俵で比較します。

def evaluate(model, X_test, y_test, scaler, close: pd.Series, test_index) -> pd.DataFrame:
    pred_scaled = model.predict(X_test, verbose=0)
    pred = scaler.inverse_transform(pred_scaled).ravel()
    actual = scaler.inverse_transform(y_test.reshape(-1, 1)).ravel()

    # ナイーブ: 前日終値
    prev = close.shift(1).loc[test_index].values[:len(actual)]

    def metrics(name, yhat):
        err = yhat - actual
        return {
            "model": name,
            "rmse": round(float(np.sqrt((err ** 2).mean())), 2),
            "mae": round(float(np.abs(err).mean()), 2),
            "mape%": round(float((np.abs(err) / actual).mean() * 100), 3),
        }

    return pd.DataFrame([metrics("LSTM", pred), metrics("naive_prev_close", prev)])

print(evaluate(model, X_test, y_test, scaler, close, test_index))

多くの環境で、naive側のRMSEがLSTMと同等かそれ以下になります。もしLSTMが大きく勝っているなら、前処理のどこかでリークが残っている可能性を先に疑ってください。冒頭で挙げた研究群の結論と逆の結果が、手元のノートブックだけで出ることは考えにくいです。

方向的中率を見る(ここで幻想が消える)

価格の絶対値ではなく、上がるか下がるかで測ります。基準は「常に上と予測する」ルールです。上昇相場では、この単純ルールが52%前後を取ります。

def direction_accuracy(model, X_test, y_test, scaler, close, test_index) -> dict:
    pred = scaler.inverse_transform(model.predict(X_test, verbose=0)).ravel()
    actual = scaler.inverse_transform(y_test.reshape(-1, 1)).ravel()
    prev = close.shift(1).loc[test_index].values[:len(actual)]

    real_up = actual > prev
    pred_up = pred > prev          # モデルが前日より上と言ったか
    always_up = np.ones_like(real_up, dtype=bool)

    return {
        "n": int(len(real_up)),
        "lstm_direction": round(float((pred_up == real_up).mean()), 3),
        "always_up": round(float((always_up == real_up).mean()), 3),
    }

print(direction_accuracy(model, X_test, y_test, scaler, close, test_index))

12銘柄15年分を比較した研究では、LSTMの方向的中率は平均47.92%で、「常に上」の52.74%を下回りました。手元でも似た数字になったら、それが正常です。回帰の見た目の良さと、売買に使える方向性は別物です。

ラグ予測かどうかを判定する

予測系列を1日ずらすと実測とほぼ一致するなら、モデルは「昨日の値を言っているだけ」です。相関で機械的に判定します。

def lag_check(pred: np.ndarray, actual: np.ndarray) -> dict:
    s_pred = pd.Series(pred)
    s_act = pd.Series(actual)
    return {
        "corr_same_day": round(float(s_pred.corr(s_act)), 4),
        "corr_pred_vs_prev_actual": round(float(s_pred.corr(s_act.shift(1))), 4),
    }

corr_pred_vs_prev_actualcorr_same_dayと同じか高いなら、ラグ予測です。グラフを重ねて「よく当たっている」と書いてある記事の多くは、この検査をしていません。

それでも改善したい人向けの順序

ナイーブに負けたあと、やみくもに層を深くしても改善しません。効き目のある順に並べます。

  1. 予測対象を価格からリターンに変える。価格は非定常です。pct_changeや対数差分を目的変数にするだけで、ラグ予測の見せかけが消えます。
  2. 回帰から分類へ。翌日が上か下かの2値にし、精度ではなくPrecision/Recallで見る。閾値を動かせるので運用に繋げやすい。
  3. 特徴量を足す。出来高、ドル円、SOX、VIX。単一系列のOHLCVだけでは信号が足りない、というのが冒頭の研究の結論でした。
  4. ウォークフォワード検証。1回の8:2分割では期間依存が消えません。TimeSeriesSplitで複数窓を回す。
  5. アンサンブル。LightGBMなど木系と混ぜる。深層学習単独に固執しない。
from sklearn.model_selection import TimeSeriesSplit

def walk_forward(close: pd.Series, n_splits: int = 5) -> pd.DataFrame:
    values = close.values.reshape(-1, 1)
    tscv = TimeSeriesSplit(n_splits=n_splits)
    rows = []
    for fold, (tr_idx, te_idx) in enumerate(tscv.split(values), start=1):
        scaler = MinMaxScaler().fit(values[tr_idx])
        tr = scaler.transform(values[tr_idx])
        te = scaler.transform(values[te_idx])
        bridge = np.vstack([tr[-WINDOW:], te])
        Xtr, ytr = make_windows(tr)
        Xte, yte = make_windows(bridge)
        if len(Xtr) < 200 or len(Xte) < 20:
            continue
        m = build_model()
        m.fit(Xtr, ytr, epochs=30, batch_size=32, shuffle=False, verbose=0,
              callbacks=[EarlyStopping(monitor="loss", patience=4,
                                       restore_best_weights=True)])
        p = scaler.inverse_transform(m.predict(Xte, verbose=0)).ravel()
        a = scaler.inverse_transform(yte.reshape(-1, 1)).ravel()
        prev = np.concatenate([[a[0]], a[:-1]])
        rows.append({
            "fold": fold,
            "lstm_rmse": round(float(np.sqrt(((p - a) ** 2).mean())), 2),
            "naive_rmse": round(float(np.sqrt(((prev - a) ** 2).mean())), 2),
        })
    return pd.DataFrame(rows)

foldごとにlstm_rmsenaive_rmseを並べ、LSTMが勝ったfoldを数えてください。5分割で1〜2勝なら、それは偶然の範囲です。全勝しない限り「予測できた」とは言えません。

よくある質問

グラフはぴったり重なるのに、なぜ使えないのか

日経平均は前日比0.5〜1%程度しか動かない日が大半です。前日の値を答えるだけで、グラフは重なります。重なりは精度の証明ではありません。差分(リターン)でプロットし直すと、予測がほぼ平坦なノイズになるのが分かります。

エポックを増やせば当たるようになるか

なりません。訓練誤差は下がりますが、検証誤差は途中から上がります。EarlyStoppingを入れているのはそのためです。ノイズを記憶しているだけです。

Transformerに変えれば勝てるか

先述の研究では、LSTM・GRU・Transformerのいずれもナイーブに勝てませんでした。アーキテクチャの問題ではなく、日次OHLCVという入力に含まれる情報量の問題です。勝ち筋があるとすれば、他の人が持っていないデータ側です。

この実装は何の役に立つのか

再現可能な検証パイプラインが手に入ります。データ取得、分割、正規化、学習、ナイーブ比較。この型があれば、次に別のモデルや特徴量を試すとき、公平に比較できます。株分析を学ぶ本3選で挙げた「型を先に作る」考え方と同じです。

まとめ:勝てないと分かるところまでが実装

LSTMで日経平均の翌日終値を予測するコードは、30行で書けます。問題は、そのコードが自分に嘘をつくことです。全期間スケーリングと分割前の窓作成で成績が上がり、MSEだけを見ればうまくいったように見えます。

やることは3つです。分割してからスケーラーをfitする。ナイーブ(前日終値)と同じ表に並べる。方向的中率を「常に上」と比較する。この3つを通したうえで負けたなら、それは失敗ではなく、正しい測定です。

次に進むなら、目的変数をリターンに変えるか、特徴量を増やす方向です。単純ルールの検証は移動平均クロスのバックテスト、コスト込みの考え方はBacktesting.py入門を先に押さえておくと、モデルの良し悪しを売買ベースで判断できるようになります。

タイトルとURLをコピーしました