LSTM で FX 為替予測を Python で実装する：設計から限界まで

# 標準的な構成
import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error
import matplotlib.pyplot as plt

# データ読み込み（OHLCV形式を想定）
df = pd.read_csv('usdjpy_1min.csv', parse_dates=['datetime'], index_col='datetime')

# 終値のみ使用する場合
close = df['close'].values.reshape(-1, 1)

# 正規化（0〜1の範囲にスケーリング）
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_close = scaler.fit_transform(close)

# 訓練/テスト分割（順序を保持すること）
train_size = int(len(scaled_close) * 0.8)
train_data = scaled_close[:train_size]
test_data = scaled_close[train_size:]

def create_sequences(data, sequence_length):
    X, y = [], []
    for i in range(sequence_length, len(data)):
        X.append(data[i - sequence_length:i, 0])
        y.append(data[i, 0])
    return np.array(X), np.array(y)

# 過去60期間を入力として次の1期間を予測
sequence_length = 60
X_train, y_train = create_sequences(train_data, sequence_length)
X_test, y_test = create_sequences(test_data, sequence_length)

# LSTMに渡すための次元変換（samples, timesteps, features）
X_train = X_train.reshape(X_train.shape[0], X_train.shape[1], 1)
X_test = X_test.reshape(X_test.shape[0], X_test.shape[1], 1)

model = Sequential([
    LSTM(units=50, return_sequences=True, 
         input_shape=(sequence_length, 1)),
    Dropout(0.2),  # 過学習防止
    LSTM(units=50, return_sequences=False),
    Dropout(0.2),
    Dense(units=1)  # 次の終値を予測
])

model.compile(optimizer='adam', loss='mean_squared_error')
model.summary()

history = model.fit(
    X_train, y_train,
    epochs=50,
    batch_size=32,
    validation_split=0.1,
    shuffle=False  # 時系列データはシャッフル禁止
)

# 予測と逆正規化
predictions_scaled = model.predict(X_test)
predictions = scaler.inverse_transform(predictions_scaled)
actual = scaler.inverse_transform(y_test.reshape(-1, 1))

# pandasで指標を追加する例
df['MA20'] = df['close'].rolling(20).mean()
df['MA50'] = df['close'].rolling(50).mean()

# ボリンジャーバンドの幅
rolling_std = df['close'].rolling(20).std()
df['BB_upper'] = df['MA20'] + 2 * rolling_std
df['BB_lower'] = df['MA20'] - 2 * rolling_std

# RSI
def calculate_rsi(prices, period=14):
    delta = prices.diff()
    gain = (delta.where(delta > 0, 0)).rolling(period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(period).mean()
    rs = gain / loss
    return 100 - (100 / (1 + rs))

df['RSI'] = calculate_rsi(df['close'])

from sklearn.metrics import mean_squared_error, mean_absolute_error
import numpy as np

mse = mean_squared_error(actual, predictions)
rmse = np.sqrt(mse)
mae = mean_absolute_error(actual, predictions)

# 方向予測精度（より実用的）
direction_actual = np.sign(np.diff(actual.flatten()))
direction_pred = np.sign(np.diff(predictions.flatten()))
directional_accuracy = np.mean(direction_actual == direction_pred)

print(f"RMSE: {rmse:.4f}")
print(f"MAE: {mae:.4f}")
print(f"方向予測精度: {directional_accuracy:.2%}")