基于LSTM的股价预测模型实战：从数据预处理到策略回测

import yfinance as yf
import pandas as pd

# 下载股票数据
ticker = "AAPL"
data = yf.download(ticker, start="2015-01-01", end="2026-01-01")

# 计算技术指标
def add_technical_indicators(df):
    # 移动平均
    df['MA_5'] = df['Close'].rolling(5).mean()
    df['MA_20'] = df['Close'].rolling(20).mean()
    
    # RSI
    delta = df['Close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    rs = gain / loss
    df['RSI'] = 100 - (100 / (1 + rs))
    
    # Bollinger Bands
    df['BB_middle'] = df['Close'].rolling(20).mean()
    bb_std = df['Close'].rolling(20).std()
    df['BB_upper'] = df['BB_middle'] + (bb_std * 2)
    df['BB_lower'] = df['BB_middle'] - (bb_std * 2)
    
    return df

data = add_technical_indicators(data)

# 添加滞后特征
def create_lag_features(df, lags=10):
    for lag in range(1, lags + 1):
        df[f'Close_lag_{lag}'] = df['Close'].shift(lag)
        df[f'Volume_lag_{lag}'] = df['Volume'].shift(lag)
        df[f'Return_lag_{lag}'] = df['Close'].pct_change().shift(lag)
    return df

# 添加滚动统计特征
def add_rolling_features(df, windows=[5, 10, 20]):
    for window in windows:
        df[f'Return_mean_{window}'] = df['Close'].pct_change().rolling(window).mean()
        df[f'Return_std_{window}'] = df['Close'].pct_change().rolling(window).std()
        df[f'Volume_mean_{window}'] = df['Volume'].rolling(window).mean()
    return df

data = create_lag_features(data)
data = add_rolling_features(data)

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout, BatchNormalization
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau

def build_lstm_model(input_shape):
    model = Sequential([
        # 第一层LSTM
        LSTM(128, return_sequences=True, input_shape=input_shape),
        BatchNormalization(),
        Dropout(0.3),
        
        # 第二层LSTM
        LSTM(64, return_sequences=True),
        BatchNormalization(),
        Dropout(0.3),
        
        # 第三层LSTM
        LSTM(32, return_sequences=False),
        BatchNormalization(),
        Dropout(0.3),
        
        # 全连接层
        Dense(64, activation='relu'),
        Dense(32, activation='relu'),
        Dense(1, activation='linear')  # 预测下一期收益率
    ])
    
    optimizer = Adam(learning_rate=0.001)
    model.compile(optimizer=optimizer, loss='mse', metrics=['mae'])
    
    return model

# 构建模型
input_shape = (lookback_days, n_features)
model = build_lstm_model(input_shape)
model.summary()

# 时间序列交叉验证
from sklearn.model_selection import TimeSeriesSplit

tscv = TimeSeriesSplit(n_splits=5)

for train_idx, val_idx in tscv.split(scaled_data):
    X_train, X_val = scaled_data[train_idx], scaled_data[val_idx]
    y_train, y_val = labels[train_idx], labels[val_idx]
    
    # 早停法
    early_stopping = EarlyStopping(
        monitor='val_loss',
        patience=10,
        restore_best_weights=True
    )
    
    # 学习率衰减
    reduce_lr = ReduceLROnPlateau(
        monitor='val_loss',
        factor=0.5,
        patience=5,
        min_lr=1e-6
    )
    
    history = model.fit(
        X_train, y_train,
        validation_data=(X_val, y_val),
        epochs=100,
        batch_size=32,
        callbacks=[early_stopping, reduce_lr],
        verbose=1
    )

import backtrader as bt

class LSTMSignalStrategy(bt.Strategy):
    params = (('threshold', 0.001),)
    
    def __init__(self):
        self.lstm_predictions = []  # LSTM预测结果
        self.current_bar = 0
        
    def next(self):
        if self.current_bar >= len(self.lstm_predictions):
            return
            
        prediction = self.lstm_predictions[self.current_bar]
        signal = generate_signals([prediction], self.params.threshold)[0]
        
        if signal == 1 and not self.position:
            self.buy(size=100)
        elif signal == -1 and self.position:
            self.close()
            
        self.current_bar += 1

# 执行回测
cerebro = bt.Cerebro()
cerebro.addstrategy(LSTMSignalStrategy, threshold=0.001)
# ... 添加数据、设置佣金等
results = cerebro.run()

def evaluate_strategy(returns, benchmark_returns):
    """计算策略绩效指标"""
    metrics = {
        '年化收益率': returns.mean() * 252,
        '年化波动率': returns.std() * np.sqrt(252),
        '夏普比率': returns.mean() / returns.std() * np.sqrt(252),
        '最大回撤': (1 - (1 + returns).cumprod() / (1 + returns).cumprod().expanding().max()).max(),
        '胜率': (returns > 0).sum() / len(returns),
        '信息比率': (returns - benchmark_returns).mean() / (returns - benchmark_returns).std() * np.sqrt(252)
    }
    return metrics

# 计算绩效
strategy_returns = calculate_strategy_returns(signals, actual_returns)
benchmark_returns = data['Close'].pct_change()

metrics = evaluate_strategy(strategy_returns, benchmark_returns)
print(metrics)