引言:深度学习进军量化交易#
传统的量化策略多依赖 statistical arbitrage、多因子模型等方法。但随着深度学习的发展,**LSTM(Long Short-Term Memory)**等递归神经网络在时序预测中展现出强大潜力。
本文将带你从零实现一个基于LSTM的股价预测系统,涵盖:
- 数据获取与预处理
- LSTM模型设计与训练
- 回测框架搭建
- 实盘部署注意事项

为什么选择LSTM?#
RNN的困境与LSTM的突破#
传统的循环神经网络(RNN)存在梯度消失/爆炸问题,难以捕捉长期依赖关系。LSTM通过精心设计的门控机制解决了这一问题:
- 遗忘门(Forget Gate):决定丢弃哪些历史信息
- 输入门(Input Gate):决定更新哪些记忆
- 输出门(Output Gate):决定输出哪些信息
数学公式#
LSTM的核心公式:
其中:
- 分别是遗忘门、输入门、输出门
- 是细胞状态(长期记忆)
- 是隐藏状态(短期记忆)

数据准备:获取与预处理#
Step 1: 获取股价数据#
我们使用 tushare 获取A股历史数据:
import tushare as ts
import pandas as pd
import numpy as np
# 设置token(需要在tushare官网注册获取)
ts.set_token('your_token_here')
pro = ts.pro_api()
def get_stock_data(ts_code, start_date, end_date):
"""
获取单只股票的交易数据
"""
df = pro.daily(
ts_code=ts_code,
start_date=start_date,
end_date=end_date,
fields='trade_date,open,high,low,close,vol,amount'
)
# 按日期升序排列
df = df.sort_values('trade_date')
df.reset_index(drop=True, inplace=True)
return df
# 获取贵州茅台的数据
df_maotai = get_stock_data('600519.SH', '20200101', '20250601')
print(f"数据形状: {df_maotai.shape}")
print(df_maotai.head())pythonStep 2: 特征工程#
单纯使用收盘价预测效果有限,我们需要构造更多特征:
def create_features(df):
"""
构造技术指标特征
"""
df = df.copy()
# 1. 移动平均线
df['MA5'] = df['close'].rolling(window=5).mean()
df['MA10'] = df['close'].rolling(window=10).mean()
df['MA20'] = df['close'].rolling(window=20).mean()
# 2. 收益率
df['return_1d'] = df['close'].pct_change(1)
df['return_5d'] = df['close'].pct_change(5)
# 3. 波动率
df['volatility_5d'] = df['return_1d'].rolling(window=5).std()
df['volatility_10d'] = df['return_1d'].rolling(window=10).std()
# 4. 成交量变化
df['vol_change'] = df['vol'].pct_change(1)
# 5. RSI指标
delta = df['close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
# 6. MACD
exp1 = df['close'].ewm(span=12, adjust=False).mean()
exp2 = df['close'].ewm(span=26, adjust=False).mean()
df['MACD'] = exp1 - exp2
df['MACD_signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
# 删除NaN值
df = df.dropna()
return df
# 应用特征工程
df_maotai = create_features(df_maotai)
print(f"特征工程后数据形状: {df_maotai.shape}")pythonStep 3: 数据归一化#
神经网络对数据尺度敏感,必须归一化:
from sklearn.preprocessing import MinMaxScaler
def normalize_data(df, feature_cols, target_col):
"""
归一化特征数据
"""
scaler_X = MinMaxScaler()
scaler_y = MinMaxScaler()
X = df[feature_cols].values
y = df[target_col].values.reshape(-1, 1)
X_scaled = scaler_X.fit_transform(X)
y_scaled = scaler_y.fit_transform(y)
return X_scaled, y_scaled, scaler_X, scaler_y
# 选择特征列
feature_cols = ['open', 'high', 'low', 'close', 'vol', 'amount',
'MA5', 'MA10', 'MA20', 'return_1d', 'return_5d',
'volatility_5d', 'vol_change', 'RSI', 'MACD', 'MACD_signal']
X_scaled, y_scaled, scaler_X, scaler_y = normalize_data(
df_maotai,
feature_cols,
'close'
)python构建LSTM模型#
Step 4: 创建时间序列样本#
LSTM需要时序样本(用过去N天预测未来M天):
def create_sequences(X, y, seq_length, pred_length=1):
"""
将数据处理成LSTM需要的时序格式
参数:
X: 特征数据 (n_samples, n_features)
y: 目标数据 (n_samples, 1)
seq_length: 输入序列长度(用过去多少天)
pred_length: 预测长度(预测未来多少天)
返回:
X_seq: 时序特征 (n_samples - seq_length, seq_length, n_features)
y_seq: 时序目标 (n_samples - seq_length, pred_length)
"""
X_seq, y_seq = [], []
for i in range(len(X) - seq_length - pred_length + 1):
X_seq.append(X[i:i + seq_length])
y_seq.append(y[i + seq_length:i + seq_length + pred_length])
return np.array(X_seq), np.array(y_seq)
# 构造时序样本
seq_length = 20 # 用过去20天预测
pred_length = 1 # 预测未来1天
X_seq, y_seq = create_sequences(X_scaled, y_scaled, seq_length, pred_length)
y_seq = y_seq.reshape(y_seq.shape[0], -1) # 展平
print(f"时序样本形状 - X: {X_seq.shape}, y: {y_seq.shape}")pythonStep 5: 划分训练集、验证集、测试集#
def train_val_test_split(X, y, train_ratio=0.7, val_ratio=0.15):
"""
按时间顺序划分数据集(不能随机划分!)
"""
n = len(X)
train_end = int(n * train_ratio)
val_end = int(n * (train_ratio + val_ratio))
X_train, y_train = X[:train_end], y[:train_end]
X_val, y_val = X[train_end:val_end], y[train_end:val_end]
X_test, y_test = X[val_end:], y[val_end:]
return X_train, y_train, X_val, y_val, X_test, y_test
X_train, y_train, X_val, y_val, X_test, y_test = train_val_test_split(
X_seq, y_seq
)
print(f"训练集: {X_train.shape}, 验证集: {X_val.shape}, 测试集: {X_test.shape}")pythonStep 6: 构建LSTM模型(PyTorch版)#
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
class LSTMModel(nn.Module):
def __init__(self, input_size, hidden_size, num_layers, output_size, dropout=0.2):
super(LSTMModel, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0
)
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# x shape: (batch_size, seq_length, input_size)
# 初始化隐藏状态
h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
# LSTM层
out, _ = self.lstm(x, (h0, c0))
# 只取最后一个时间步的输出
out = out[:, -1, :]
# 全连接层
out = self.fc(out)
return out
# 初始化模型
input_size = X_train.shape[2] # 特征数量
hidden_size = 64
num_layers = 2
output_size = pred_length
model = LSTMModel(input_size, hidden_size, num_layers, output_size)
print(model)pythonStep 7: 训练模型#
def train_model(model, X_train, y_train, X_val, y_val, epochs=100, batch_size=32, lr=0.001):
"""
训练LSTM模型
"""
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f"使用设备: {device}")
model = model.to(device)
# 转换为PyTorch张量
X_train_tensor = torch.FloatTensor(X_train).to(device)
y_train_tensor = torch.FloatTensor(y_train).to(device)
X_val_tensor = torch.FloatTensor(X_val).to(device)
y_val_tensor = torch.FloatTensor(y_val).to(device)
# 定义损失函数和优化器
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=lr)
# 训练历史记录
train_losses = []
val_losses = []
for epoch in range(epochs):
# 训练模式
model.train()
train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
train_loss = 0.0
for batch_X, batch_y in train_loader:
optimizer.zero_grad()
outputs = model(batch_X)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
train_loss += loss.item()
train_loss /= len(train_loader)
train_losses.append(train_loss)
# 验证模式
model.eval()
with torch.no_grad():
val_outputs = model(X_val_tensor)
val_loss = criterion(val_outputs, y_val_tensor)
val_losses.append(val_loss.item())
if (epoch + 1) % 20 == 0:
print(f"Epoch [{epoch+1}/{epochs}], Train Loss: {train_loss:.6f}, Val Loss: {val_loss.item():.6f}")
return model, train_losses, val_losses
# 训练模型
model, train_losses, val_losses = train_model(
model, X_train, y_train, X_val, y_val,
epochs=100, batch_size=32, lr=0.001
)python
模型评估与回测#
Step 8: 在测试集上评估#
def evaluate_model(model, X_test, y_test, scaler_y):
"""
在测试集上评估模型
"""
model.eval()
device = next(model.parameters()).device
X_test_tensor = torch.FloatTensor(X_test).to(device)
with torch.no_grad():
y_pred_scaled = model(X_test_tensor).cpu().numpy()
# 反归一化
y_test_inv = scaler_y.inverse_transform(y_test)
y_pred_inv = scaler_y.inverse_transform(y_pred_scaled)
# 计算评估指标
from sklearn.metrics import mean_absolute_error, mean_squared_error
import numpy as np
mae = mean_absolute_error(y_test_inv, y_pred_inv)
rmse = np.sqrt(mean_squared_error(y_test_inv, y_pred_inv))
mape = np.mean(np.abs((y_test_inv - y_pred_inv) / y_test_inv)) * 100
print(f"MAE: {mae:.4f}")
print(f"RMSE: {rmse:.4f}")
print(f"MAPE: {mape:.2f}%")
return y_test_inv, y_pred_inv
y_test_inv, y_pred_inv = evaluate_model(model, X_test, y_test, scaler_y)pythonStep 9: 可视化预测结果#
def plot_predictions(y_true, y_pred, title='LSTM Stock Price Prediction'):
"""
可视化预测结果
"""
plt.figure(figsize=(14, 6))
plt.plot(y_true, label='True Price', linewidth=2)
plt.plot(y_pred, label='Predicted Price', linewidth=2, linestyle='--')
plt.xlabel('Time', fontsize=12)
plt.ylabel('Stock Price', fontsize=12)
plt.title(title, fontsize=14, fontweight='bold')
plt.legend(fontsize=12)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('/Users/halo/workspace/astro-blog/public/images/lstm-stock-prediction-practical/prediction_result.jpg',
dpi=300,
bbox_inches='tight',
facecolor='white')
plt.close()
print("✓ 保存预测结果图")
plot_predictions(y_test_inv, y_pred_inv)python
Step 10: 回测策略#
光有预测精度还不够,关键是能否赚钱!
def backtest_strategy(y_true, y_pred, initial_capital=100000, transaction_cost=0.001):
"""
回测简单交易策略
策略逻辑:
- 预测上涨超过1% → 买入
- 预测下跌超过1% → 卖出
- 否则 → 持有
"""
positions = [] # 持仓记录: 1=多头, -1=空头, 0=空仓
capital = initial_capital
shares = 0
portfolio_values = []
for i in range(1, len(y_pred)):
pred_return = (y_pred[i] - y_true[i-1]) / y_true[i-1]
# 交易信号
if pred_return > 0.01 and positions == []: # 买入信号
shares = capital // y_true[i]
capital -= shares * y_true[i] * (1 + transaction_cost)
positions.append(1)
elif pred_return < -0.01 and positions == [1]: # 卖出信号
capital += shares * y_true[i] * (1 - transaction_cost)
shares = 0
positions = []
# 计算当前组合价值
portfolio_value = capital + shares * y_true[i]
portfolio_values.append(portfolio_value)
# 计算策略收益
final_value = portfolio_values[-1]
total_return = (final_value - initial_capital) / initial_capital * 100
# 计算基准收益(买入持有)
benchmark_return = (y_true[-1] - y_true[0]) / y_true[0] * 100
print(f"策略总收益: {total_return:.2f}%")
print(f"基准收益(买入持有): {benchmark_return:.2f}%")
print(f"超额收益: {total_return - benchmark_return:.2f}%")
return portfolio_values
portfolio_values = backtest_strategy(y_test_inv, y_pred_inv)python优化与调参#
技巧1: 使用更长的序列长度#
# 尝试不同的序列长度
seq_lengths = [10, 20, 30, 40, 50]
results = {}
for seq_len in seq_lengths:
print(f"\n训练模型 (seq_length={seq_len})...")
X_seq, y_seq = create_sequences(X_scaled, y_scaled, seq_len, pred_length=1)
y_seq = y_seq.reshape(y_seq.shape[0], -1)
X_train, y_train, X_val, y_val, X_test, y_test = train_val_test_split(X_seq, y_seq)
model = LSTMModel(input_size, hidden_size, num_layers, output_size)
model, _, _ = train_model(model, X_train, y_train, X_val, y_val, epochs=50, batch_size=32, lr=0.001)
y_test_inv, y_pred_inv = evaluate_model(model, X_test, y_test, scaler_y)
# 记录结果
results[seq_len] = {'mae': mae, 'rmse': rmse, 'mape': mape}
# 可视化不同序列长度的效果
import matplotlib.pyplot as plt
seq_lengths = list(results.keys())
maes = [results[s]['mae'] for s in seq_lengths]
plt.figure(figsize=(10, 6))
plt.plot(seq_lengths, maes, 'bo-', linewidth=2, markersize=8)
plt.xlabel('Sequence Length', fontsize=12)
plt.ylabel('MAE', fontsize=12)
plt.title('Effect of Sequence Length on Model Performance', fontsize=14, fontweight='bold')
plt.grid(True, alpha=0.3)
plt.savefig('/Users/halo/workspace/astro-blog/public/images/lstm-stock-prediction-practical/seq_length_effect.jpg',
dpi=300,
bbox_inches='tight',
facecolor='white')
plt.close()python
技巧2: 加入注意力机制(Attention)#
class LSTMAttentionModel(nn.Module):
def __init__(self, input_size, hidden_size, num_layers, output_size, dropout=0.2):
super(LSTMAttentionModel, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0
)
# 注意力层
self.attention = nn.Linear(hidden_size, 1)
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# LSTM层
lstm_out, _ = self.lstm(x) # lstm_out shape: (batch, seq_len, hidden)
# 注意力权重
attention_weights = torch.softmax(self.attention(lstm_out), dim=1) # (batch, seq_len, 1)
attention_weights = attention_weights.transpose(1, 2) # (batch, 1, seq_len)
# 加权求和
context = torch.bmm(attention_weights, lstm_out).squeeze(1) # (batch, hidden)
# 全连接层
out = self.fc(context)
return out
# 使用注意力模型
model_attention = LSTMAttentionModel(input_size, hidden_size, num_layers, output_size)python技巧3: 贝叶斯优化超参数#
from bayes_opt import BayesianOptimization
def lstm_cv(hidden_size, num_layers, dropout, lr):
"""
交叉验证评估函数(用于贝叶斯优化)
"""
hidden_size = int(hidden_size)
num_layers = int(num_layers)
# 训练模型(简化版,只用一部分数据)
model = LSTMModel(input_size, hidden_size, num_layers, output_size, dropout)
model, _, val_losses = train_model(
model, X_train[:500], y_train[:500], X_val, y_val,
epochs=30, batch_size=32, lr=lr
)
# 返回验证集上的负损失(贝叶斯优化是最大化目标)
return -val_losses[-1]
# 定义参数空间
pbounds = {
'hidden_size': (32, 128),
'num_layers': (1, 3),
'dropout': (0.1, 0.5),
'lr': (1e-4, 1e-2)
}
# 贝叶斯优化
optimizer = BayesianOptimization(
f=lstm_cv,
pbounds=pbounds,
random_state=42
)
optimizer.maximize(
init_points=5,
n_iter=20
)
print("最优超参数:")
print(optimizer.max)python实盘部署注意事项#
陷阱1: 过拟合#
深度学习的过拟合风险极高!防范方法:
- 早停(Early Stopping)
- 正则化(Dropout、L2正则)
- 交叉验证(Time Series Split)
- 样本外测试(Out-of-Sample)
from sklearn.model_selection import TimeSeriesSplit
def time_series_cv(X, y, n_splits=5):
"""
时间序列交叉验证
"""
tscv = TimeSeriesSplit(n_splits=n_splits)
cv_scores = []
for train_idx, val_idx in tscv.split(X):
X_train_cv, X_val_cv = X[train_idx], X[val_idx]
y_train_cv, y_val_cv = y[train_idx], y[val_idx]
model = LSTMModel(input_size, hidden_size, num_layers, output_size)
model, _, _ = train_model(
model, X_train_cv, y_train_cv, X_val_cv, y_val_cv,
epochs=50, batch_size=32, lr=0.001
)
# 评估
y_pred_cv = model(torch.FloatTensor(X_val_cv)).detach().numpy()
mse = mean_squared_error(y_val_cv, y_pred_cv)
cv_scores.append(mse)
print(f"CV MSE: {np.mean(cv_scores):.6f} (+/- {np.std(cv_scores):.6f})")
return cv_scores
cv_scores = time_series_cv(X_seq, y_seq)python陷阱2: 数据泄露#
严禁在训练集中使用未来数据!常见泄露场景:
❌ 错误:用全量数据归一化
# 错误做法
scaler.fit(X) # 泄露测试集信息!
X_train_scaled = scaler.transform(X_train)
X_test_scaled = scaler.transform(X_test)python✅ 正确:仅用训练集拟合
# 正确做法
scaler.fit(X_train) # 只用训练集
X_train_scaled = scaler.transform(X_train)
X_test_scaled = scaler.transform(X_test)python陷阱3: 非平稳性#
股价序列是非平稳的(均值和方差随时间变化),直接预测价格效果差。
解决方案:预测收益率或价格变化
# 预测收益率而不是价格
df['return'] = df['close'].pct_change()
y = df['return'].values
# 或者预测价格变化方向(分类问题)
df['target'] = (df['close'].shift(-1) > df['close']).astype(int)python陷阱4: 交易成本#
回测时必须考虑交易成本(佣金、滑点、印花税)
def backtest_with_cost(y_true, y_pred, transaction_cost=0.002):
"""
考虑交易成本的回测
transaction_cost包括:
- 佣金: 0.0003
- 印花税: 0.001 (卖出时)
- 滑点: 0.0007 (假设)
"""
# ... (回测逻辑同上)python完整实战代码#
端到端Pipeline#
class LSTMStockPredictor:
"""
LSTM股价预测完整流程封装
"""
def __init__(self, seq_length=20, hidden_size=64, num_layers=2, pred_length=1):
self.seq_length = seq_length
self.hidden_size = hidden_size
self.num_layers = num_layers
self.pred_length = pred_length
self.scaler_X = MinMaxScaler()
self.scaler_y = MinMaxScaler()
self.model = None
def prepare_data(self, df, feature_cols, target_col):
"""
数据预处理
"""
# 构造特征
df = create_features(df)
# 归一化
X = df[feature_cols].values
y = df[target_col].values.reshape(-1, 1)
X_scaled = self.scaler_X.fit_transform(X)
y_scaled = self.scaler_y.fit_transform(y)
# 创建时序样本
X_seq, y_seq = create_sequences(X_scaled, y_scaled, self.seq_length, self.pred_length)
y_seq = y_seq.reshape(y_seq.shape[0], -1)
return X_seq, y_seq
def train(self, X, y, epochs=100, batch_size=32, lr=0.001):
"""
训练模型
"""
X_train, y_train, X_val, y_val, X_test, y_test = train_val_test_split(X, y)
input_size = X_train.shape[2]
output_size = self.pred_length
self.model = LSTMModel(input_size, self.hidden_size, self.num_layers, output_size)
self.model, _, _ = train_model(
self.model, X_train, y_train, X_val, y_val,
epochs=epochs, batch_size=batch_size, lr=lr
)
return X_test, y_test
def predict(self, X):
"""
预测
"""
self.model.eval()
device = next(self.model.parameters()).device
X_tensor = torch.FloatTensor(X).to(device)
with torch.no_grad():
y_pred_scaled = self.model(X_tensor).cpu().numpy()
y_pred = self.scaler_y.inverse_transform(y_pred_scaled)
return y_pred
# 使用示例
predictor = LSTMStockPredictor(seq_length=20, hidden_size=64, num_layers=2)
X, y = predictor.prepare_data(df_maotai, feature_cols, 'close')
X_test, y_test = predictor.train(X, y, epochs=100)
y_pred = predictor.predict(X_test)python总结与展望#
本文总结#
✅ 从零实现了基于LSTM的股价预测系统
✅ 涵盖了数据获取、特征工程、模型训练、回测全流程
✅ 介绍了超参数优化、注意力机制等进阶技巧
✅ 强调了实盘部署的常见陷阱
LSTM的局限性#
尽管LSTM强大,但仍有局限:
- 计算成本高(训练慢)
- 需要大量数据(小样本容易过拟合)
- ** interpretability差**(黑盒模型)
- 对噪声敏感(市场噪声多)
未来方向#
- Transformer模型:捕捉更长距离依赖
- 图神经网络:利用股票间关联
- 强化学习:端到端策略优化
- 多模态融合:结合新闻、社交媒体数据
实战建议#
- 不要盲目追AI:传统统计方法仍有效
- 组合模型:LSTM + 线性模型
- 风险管理第一:再好的预测也要止损
- 持续监控:市场结构变化,模型要更新
完整代码仓库: [GitHub链接]
下期预告: 《Transformer在量化交易中的应用:超越LSTM的时序模型》
觉得有用?欢迎关注我的量化专栏,每周更新实战干货!别忘了点赞+收藏~