LSTM神经网络股价预测：深度学习在量化交易中的实战指南

import numpy as np
import pandas as pd
import yfinance as yf
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout

# 1. 获取沪深300数据
hs300 = yf.download('000300.SS', start='2015-01-01', end='2024-12-31')

# 2. 构造特征
df = pd.DataFrame({
    'close': hs300['Close'],
    'returns': hs300['Close'].pct_change(),
    'ma5': hs300['Close'].rolling(5).mean(),
    'ma20': hs300['Close'].rolling(20).mean(),
    'volatility': hs300['Close'].pct_change().rolling(20).std(),
    'volume_change': hs300['Volume'].pct_change()
}).dropna()

# 3. 归一化
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(df)

# 4. 构造时间序列样本
def create_sequences(data, seq_length):
    X, y = [], []
    for i in range(seq_length, len(data)):
        X.append(data[i-seq_length:i])
        y.append(data[i, 0])  # 预测收盘价
    return np.array(X), np.array(y)

SEQ_LENGTH = 60  # 使用过去60个交易日预测下一天
X, y = create_sequences(scaled_data, SEQ_LENGTH)

# 5. 划分训练集和测试集
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:]

# 6. 构建LSTM模型
model = Sequential([
    LSTM(50, return_sequences=True, input_shape=(SEQ_LENGTH, X.shape[2])),
    Dropout(0.2),
    LSTM(50, return_sequences=False),
    Dropout(0.2),
    Dense(25),
    Dense(1)
])

model.compile(optimizer='adam', loss='mse')

# 7. 训练模型
history = model.fit(
    X_train, y_train,
    batch_size=32,
    epochs=50,
    validation_split=0.1,
    verbose=1
)

# 8. 预测
predicted = model.predict(X_test)
predicted = scaler.inverse_transform(
    np.concatenate([predicted, np.zeros((len(predicted), scaled_data.shape[1]-1))], axis=1)
)[:, 0]

actual = scaler.inverse_transform(
    np.concatenate([y_test.reshape(-1, 1), np.zeros((len(y_test), scaled_data.shape[1]-1))], axis=1)
)[:, 0]

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

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

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

# 预测明日涨跌
signals = np.where(predicted > actual[:-1], 1, -1)

# 计算策略收益
strategy_returns = signals * df['returns'].iloc[train_size+SEQ_LENGTH:]

cumulative_returns = (1 + strategy_returns).cumprod()
buy_hold_returns = (1 + df['returns'].iloc[train_size+SEQ_LENGTH:]).cumprod()

# 可视化
import matplotlib.pyplot as plt

plt.figure(figsize=(12, 6))
plt.plot(cumulative_returns, label='LSTM Strategy')
plt.plot(buy_hold_returns, label='Buy & Hold')
plt.legend()
plt.show()

# 使用预测概率（回归转分类）
up_prob = model.predict(X_test)

threshold = 0.55  # 只有预测上涨概率>55%才买入
signals = np.where(up_prob > threshold, 1, 0)

# 风险控制：连续止损3次后空仓
max_consecutive_loss = 3
consecutive_loss = 0

for i in range(1, len(signals)):
    if strategy_returns[i-1] < 0:
        consecutive_loss += 1
    else:
        consecutive_loss = 0
    
    if consecutive_loss >= max_consecutive_loss:
        signals[i] = 0  # 空仓