LSTM神经网络在量化交易中的应用：从理论到实战

import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler

# 1. 加载数据
df = pd.read_csv('stock_data.csv', index_col=0, parse_dates=True)
prices = df['close'].values.reshape(-1, 1)

# 2. 数据归一化（LSTM对尺度敏感）
scaler = MinMaxScaler(feature_range=(0, 1))
prices_scaled = scaler.fit_transform(prices)

# 3. 构建时间序列样本
def create_dataset(data, look_back=60):
    X, y = [], []
    for i in range(len(data) - look_back):
        X.append(data[i:(i + look_back), 0])
        y.append(data[i + look_back, 0])
    return np.array(X), np.array(y)

look_back = 60  # 用过去60天预测下一天
X, y = create_dataset(prices_scaled, look_back)

# 4. 划分训练集/测试集
train_size = int(len(X) * 0.8)
X_train, X_test = X[:train_size], X[train_size:]
y_train, y_test = y[:train_size], y[train_size:]

# 5. 调整输入形状 (samples, time_steps, 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))

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout

model = Sequential([
    # 第一层LSTM：返回完整序列
    LSTM(50, return_sequences=True, input_shape=(look_back, 1)),
    Dropout(0.2),  # 防止过拟合

    # 第二层LSTM：只返回最后时刻的输出
    LSTM(50, return_sequences=False),
    Dropout(0.2),

    # 全连接层：输出预测值
    Dense(25, activation='relu'),
    Dense(1, activation='linear')  # 回归任务用linear
])

# 编译模型
model.compile(optimizer='adam', loss='mse', metrics=['mae'])
model.summary()

# 预测
y_pred_scaled = model.predict(X_test)

# 反归一化
y_pred = scaler.inverse_transform(y_pred_scaled)
y_true = scaler.inverse_transform(y_test.reshape(-1, 1))

# 评估指标
from sklearn.metrics import mean_squared_error, mean_absolute_error
import math

rmse = math.sqrt(mean_squared_error(y_true, y_pred))
mae = mean_absolute_error(y_true, y_pred)
mape = np.mean(np.abs((y_true - y_pred) / y_true)) * 100

print(f"RMSE: {rmse:.2f}")
print(f"MAE: {mae:.2f}")
print(f"MAPE: {mape:.2f}%")

# 可视化
import matplotlib.pyplot as plt

plt.figure(figsize=(12, 6))
plt.plot(y_true, label='True Price')
plt.plot(y_pred, label='Predicted Price')
plt.legend()
plt.show()

# 特征工程
features = pd.DataFrame({
    'close': df['close'],
    'volume': df['volume'],
    'ma5': df['close'].rolling(5).mean(),
    'ma20': df['close'].rolling(20).mean(),
    'rsi': compute_rsi(df['close'], 14),
    'macd': compute_macd(df['close'])
})

# 归一化
scaler = MinMaxScaler()
features_scaled = scaler.fit_transform(features)

# 调整输入形状 (samples, time_steps, features)
X = features_scaled.reshape((n_samples, look_back, n_features))

# 1. 滚动训练（Walk-Forward Optimization）
for end_date in rolling_dates:
    # 用过去1年数据训练
    X_train, y_train = get_data(train_start, end_date)
    model.fit(X_train, y_train)

    # 预测未来1个月
    X_test = get_data(end_date, end_date + 30)
    predictions = model.predict(X_test)

    # 用真实值重新训练
    model.fit(X_test, y_true)

# 2. 在线学习（Online Learning）
for new_data in stream_data:
    model.train_on_batch(new_data.x, new_data.y)  # 增量训练