6.3 深度学习时序预测简介 — LSTM¶

学习目标¶

• 理解 LSTM（长短期记忆网络）处理时序数据的原理
• 用 PyTorch 构建并训练 LSTM 价格预测模型
• 理解 DL 在金融场景的局限性与注意事项

本节需要安装 PyTorch（pip install torch）。 初学者可先跳过，在掌握前几节 ML 内容后再回来。

1. 这节究竟解决什么实际问题？¶

6.3 深度学习时序预测简介 — LSTM 的具体任务是：用 LSTM 演示时序输入、训练与验证，但不把拟合曲线视为预测优势。 本节将明确输入数据、计算过程、输出及其在后续研究中的用途。

2. 前置知识¶

掌握特征、标签、训练集/测试集和基本回测概念。 对公式的理解应包括变量定义、单位、时间索引和适用条件。

3. 数据说明¶

特征默认来自固定样本或模拟样本；每个标签必须标明预测期限及其可交易时间。 运行前先检查 shape、日期范围、列名与缺失值。数据看起来能画图，不代表它适合回答研究问题。

4. 学习目标与完成标准¶

完成本节后，你应能用自己的话解释 6.3 深度学习时序预测简介 — LSTM 的输入、计算步骤、输出和局限；尤其能说明：用 LSTM 演示时序输入、训练与验证，但不把拟合曲线视为预测优势。 你还应能独立改动一个参数并解释结果为什么变化，指出至少一个会让结论失真的假设。

5. 核心概念与公式¶

机器学习模型从训练样本估计输入与目标之间的关系。训练集拟合程度不能代表样本外预测能力。

训练时只能看到训练区间；特征时间不得晚于决策时间，标签与测试区间必须隔离。 看到公式时依次确认：每个符号的单位是什么？它属于哪个时点？分母会不会为零？它是描述性统计量、预测量，还是交易规则？

6. 从零实现¶

原始代码按照“准备输入 → 计算中间量 → 得到结果 → 解释输出”的顺序组织。每个单元应检查变量类型、索引、维度和代表性数值。中间变量用于核对公式实现和定位计算错误。

7. 成熟库实现¶

sklearn/PyTorch 加速建模；管道、切分和经济解释仍由研究者负责。 使用库函数前先阅读参数含义、默认值和返回对象；库输出一个数字，并不说明这个数字适合你的交易假设。

8. 两种实现的差异¶

从零实现便于核对公式、调试和识别隐含假设；成熟库适合在计算逻辑已验证后复用。两种实现依赖相同的数据口径与时点假设，使用库函数不会改变输入数据的质量。

9. 预期结果与检查范围¶

你应看到维度正确的 Series/DataFrame、日期单调递增的索引，以及能支持“用 LSTM 演示时序输入、训练与验证，但不把拟合曲线视为预测优势。”这一任务的表格或图形。金融日收益率均值通常接近 0；若出现极大的连续收益、空表、全 NaN 或索引错位，应先排查数据和对齐逻辑，而不是立即相信结果。

10. 常见错误与诊断¶

随机打乱时序、Scaler 在全样本 fit、目标泄漏、调参反复查看测试集、把准确率等同收益。

诊断顺序：先打印类型与 shape，再看 head()/tail()，接着检查日期和 NaN，最后才比较数值。这样能避免在错误的对象上反复改参数。

11. 研究偏差检查¶

时间泄漏、p-hacking、多重检验和样本选择会制造虚假的 Alpha；必须报告朴素基线。 结果至少应与一个简单基线比较，并明确哪些部分只是教学演示。

12. 三层练习与答案¶

• 基础练习：验证一个中间变量、补全一段小计算，并用一句话解释输出。
• 研究练习：改变窗口、样本区间、约束或标的，记录哪些结论仍成立。
• 开放项目：把本节方法嵌入模块项目，报告基线、失败结果和限制。

题目在 ../exercises/06_ml_trading/03_deep_learning_exercises.ipynb；完整推导、可运行参考实现与结果检查在 ../solutions/06_ml_trading/03_deep_learning_solutions.ipynb。请先独立作答，再打开答案。

13. 延伸阅读¶

延伸材料：scikit-learn 模型选择文档。阅读记录应包括研究假设、数据频率、样本范围和结论适用条件。

14. 下一节¶

下一节是 04_dangers_of_overfitting。本节的概念会成为理解它的输入，而不是孤立的名词。

In [3]:

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import yfinance as yf
from sklearn.preprocessing import MinMaxScaler

try:
    import torch
    import torch.nn as nn
    from torch.utils.data import DataLoader, TensorDataset
    print(f'PyTorch 版本: {torch.__version__} ')
    DEVICE = 'cuda' if torch.cuda.is_available() else 'cpu'
    print(f'使用设备: {DEVICE}')
except ImportError:
    print('  请安装 PyTorch: pip install torch')
    raise SystemExit('跳过本节，请先安装 torch')

import numpy as np import pandas as pd import matplotlib.pyplot as plt import yfinance as yf from sklearn.preprocessing import MinMaxScaler try: import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset print(f'PyTorch 版本: {torch.__version__} ') DEVICE = 'cuda' if torch.cuda.is_available() else 'cpu' print(f'使用设备: {DEVICE}') except ImportError: print(' 请安装 PyTorch: pip install torch') raise SystemExit('跳过本节，请先安装 torch')

PyTorch 版本: 2.5.1 
使用设备: cuda

In [ ]:

# 默认使用固定样本，保证课程与 CI 不依赖实时接口。
USE_OFFLINE_DATA = True
if USE_OFFLINE_DATA:
    from pathlib import Path
    import sys
    docs_root = next(parent for parent in [Path.cwd(), *Path.cwd().parents]
                     if (parent / 'utils' / 'offline_data.py').exists())
    sys.path.insert(0, str(docs_root))
    from utils.offline_data import configure_yfinance_offline
    configure_yfinance_offline(yf)
else:
    print('在线下载为扩展步骤；请核对数据字段、时区和下载日期。')

# 默认使用固定样本，保证课程与 CI 不依赖实时接口。 USE_OFFLINE_DATA = True if USE_OFFLINE_DATA: from pathlib import Path import sys docs_root = next(parent for parent in [Path.cwd(), *Path.cwd().parents] if (parent / 'utils' / 'offline_data.py').exists()) sys.path.insert(0, str(docs_root)) from utils.offline_data import configure_yfinance_offline configure_yfinance_offline(yf) else: print('在线下载为扩展步骤；请核对数据字段、时区和下载日期。')

In [4]:

import matplotlib.pyplot as plt

# 1. 设置系统自带的中文字体（这里使用黑体 SimHei）
plt.rcParams['font.sans-serif'] = ['SimHei']  # 如果你想用微软雅黑，可以改成 ['Microsoft YaHei']

# 2. 解决更换字体后，负号（-）显示为方块的问题
plt.rcParams['axes.unicode_minus'] = False

import matplotlib.pyplot as plt # 1. 设置系统自带的中文字体（这里使用黑体 SimHei） plt.rcParams['font.sans-serif'] = ['SimHei'] # 如果你想用微软雅黑，可以改成 ['Microsoft YaHei'] # 2. 解决更换字体后，负号（-）显示为方块的问题 plt.rcParams['axes.unicode_minus'] = False

1. 为什么用 LSTM？¶

普通 RNN 存在梯度消失问题，难以捕捉长期依赖。LSTM 通过门机制解决这一问题：

输入门  → 决定哪些新信息进入记忆
遗忘门  → 决定丢弃哪些旧记忆
输出门  → 决定输出哪些隐藏状态

LSTM 可以表示跨多个时间步的依赖关系。金融序列通常信噪比较低，因此需要使用时序验证、正则化和样本外测试检查过拟合。

2. 数据准备¶

In [5]:

# 下载数据
raw = yf.download('SPY', start='2015-01-01', end='2024-01-01', progress=False)
close = raw['Close'].squeeze().values.reshape(-1, 1)

# 归一化到 [0, 1]
scaler = MinMaxScaler()
close_scaled = scaler.fit_transform(close)

def make_sequences(data, seq_len=60):
    """将时序数据转为 (X, y) 样本对
    X: 过去 seq_len 天的价格序列
    y: 下一天的价格
    """
    X, y = [], []
    for i in range(len(data) - seq_len):
        X.append(data[i: i + seq_len])
        y.append(data[i + seq_len])
    return np.array(X), np.array(y)

SEQ_LEN = 60  # 用过去60天预测
X, y = make_sequences(close_scaled, SEQ_LEN)

# 时序划分：80% 训练，20% 测试
split = int(len(X) * 0.8)
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]

# 转为 Tensor
X_train_t = torch.FloatTensor(X_train).to(DEVICE)
X_test_t  = torch.FloatTensor(X_test).to(DEVICE)
y_train_t = torch.FloatTensor(y_train).to(DEVICE)
y_test_t  = torch.FloatTensor(y_test).to(DEVICE)

train_dataset = TensorDataset(X_train_t, y_train_t)
train_loader  = DataLoader(train_dataset, batch_size=32, shuffle=False)

print(f'训练集: X={X_train.shape}, y={y_train.shape}')
print(f'测试集: X={X_test.shape},  y={y_test.shape}')

# 下载数据 raw = yf.download('SPY', start='2015-01-01', end='2024-01-01', progress=False) close = raw['Close'].squeeze().values.reshape(-1, 1) # 归一化到 [0, 1] scaler = MinMaxScaler() close_scaled = scaler.fit_transform(close) def make_sequences(data, seq_len=60): """将时序数据转为 (X, y) 样本对 X: 过去 seq_len 天的价格序列 y: 下一天的价格 """ X, y = [], [] for i in range(len(data) - seq_len): X.append(data[i: i + seq_len]) y.append(data[i + seq_len]) return np.array(X), np.array(y) SEQ_LEN = 60 # 用过去60天预测 X, y = make_sequences(close_scaled, SEQ_LEN) # 时序划分：80% 训练，20% 测试 split = int(len(X) * 0.8) X_train, X_test = X[:split], X[split:] y_train, y_test = y[:split], y[split:] # 转为 Tensor X_train_t = torch.FloatTensor(X_train).to(DEVICE) X_test_t = torch.FloatTensor(X_test).to(DEVICE) y_train_t = torch.FloatTensor(y_train).to(DEVICE) y_test_t = torch.FloatTensor(y_test).to(DEVICE) train_dataset = TensorDataset(X_train_t, y_train_t) train_loader = DataLoader(train_dataset, batch_size=32, shuffle=False) print(f'训练集: X={X_train.shape}, y={y_train.shape}') print(f'测试集: X={X_test.shape}, y={y_test.shape}')

训练集: X=(1763, 60, 1), y=(1763, 1)
测试集: X=(441, 60, 1),  y=(441, 1)

3. 构建 LSTM 模型¶

In [6]:

class LSTMModel(nn.Module):
    def __init__(self, input_size=1, hidden_size=64, num_layers=2,
                 output_size=1, dropout=0.2):
        super().__init__()
        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.dropout = nn.Dropout(dropout)
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        # x: (batch, seq_len, input_size)
        out, _ = self.lstm(x)
        # 取最后一个时间步
        out = self.dropout(out[:, -1, :])
        return self.fc(out)

model = LSTMModel().to(DEVICE)
print(model)
total_params = sum(p.numel() for p in model.parameters())
print(f'\n总参数量: {total_params:,}')

class LSTMModel(nn.Module): def __init__(self, input_size=1, hidden_size=64, num_layers=2, output_size=1, dropout=0.2): super().__init__() 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.dropout = nn.Dropout(dropout) self.fc = nn.Linear(hidden_size, output_size) def forward(self, x): # x: (batch, seq_len, input_size) out, _ = self.lstm(x) # 取最后一个时间步 out = self.dropout(out[:, -1, :]) return self.fc(out) model = LSTMModel().to(DEVICE) print(model) total_params = sum(p.numel() for p in model.parameters()) print(f'\n总参数量: {total_params:,}')

LSTMModel(
  (lstm): LSTM(1, 64, num_layers=2, batch_first=True, dropout=0.2)
  (dropout): Dropout(p=0.2, inplace=False)
  (fc): Linear(in_features=64, out_features=1, bias=True)
)

总参数量: 50,497

4. 训练模型¶

In [7]:

criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5, factor=0.5)

EPOCHS = 50
train_losses, val_losses = [], []

for epoch in range(EPOCHS):
    # ---- 训练 ----
    model.train()
    batch_losses = []
    for X_batch, y_batch in train_loader:
        optimizer.zero_grad()
        pred = model(X_batch)
        loss = criterion(pred, y_batch)
        loss.backward()
        nn.utils.clip_grad_norm_(model.parameters(), 1.0)  # 梯度裁剪
        optimizer.step()
        batch_losses.append(loss.item())
    train_loss = np.mean(batch_losses)
    train_losses.append(train_loss)

    # ---- 验证 ----
    model.eval()
    with torch.no_grad():
        val_pred = model(X_test_t)
        val_loss = criterion(val_pred, y_test_t).item()
    val_losses.append(val_loss)
    scheduler.step(val_loss)

    if (epoch + 1) % 10 == 0:
        print(f'Epoch [{epoch+1:2d}/{EPOCHS}] '
              f'Train Loss: {train_loss:.6f}  '
              f'Val Loss: {val_loss:.6f}')

# 绘制损失曲线
fig, ax = plt.subplots(figsize=(10, 4))
ax.plot(train_losses, label='训练损失', linewidth=1.5)
ax.plot(val_losses, label='验证损失', linewidth=1.5)
ax.set_title('LSTM 训练过程', fontsize=13)
ax.set_xlabel('Epoch')
ax.set_ylabel('MSE Loss')
ax.legend()
ax.grid(alpha=0.3)
plt.tight_layout()
plt.show()

criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=0.001) scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5, factor=0.5) EPOCHS = 50 train_losses, val_losses = [], [] for epoch in range(EPOCHS): # ---- 训练 ---- model.train() batch_losses = [] for X_batch, y_batch in train_loader: optimizer.zero_grad() pred = model(X_batch) loss = criterion(pred, y_batch) loss.backward() nn.utils.clip_grad_norm_(model.parameters(), 1.0) # 梯度裁剪 optimizer.step() batch_losses.append(loss.item()) train_loss = np.mean(batch_losses) train_losses.append(train_loss) # ---- 验证 ---- model.eval() with torch.no_grad(): val_pred = model(X_test_t) val_loss = criterion(val_pred, y_test_t).item() val_losses.append(val_loss) scheduler.step(val_loss) if (epoch + 1) % 10 == 0: print(f'Epoch [{epoch+1:2d}/{EPOCHS}] ' f'Train Loss: {train_loss:.6f} ' f'Val Loss: {val_loss:.6f}') # 绘制损失曲线 fig, ax = plt.subplots(figsize=(10, 4)) ax.plot(train_losses, label='训练损失', linewidth=1.5) ax.plot(val_losses, label='验证损失', linewidth=1.5) ax.set_title('LSTM 训练过程', fontsize=13) ax.set_xlabel('Epoch') ax.set_ylabel('MSE Loss') ax.legend() ax.grid(alpha=0.3) plt.tight_layout() plt.show()

Epoch [10/50] Train Loss: 0.004590  Val Loss: 0.011013
Epoch [20/50] Train Loss: 0.002400  Val Loss: 0.000541
Epoch [30/50] Train Loss: 0.001631  Val Loss: 0.000501
Epoch [40/50] Train Loss: 0.001542  Val Loss: 0.000744
Epoch [50/50] Train Loss: 0.001455  Val Loss: 0.000522

5. 预测与可视化¶

In [8]:

model.eval()
with torch.no_grad():
    y_pred_scaled = model(X_test_t).cpu().numpy()

# 反归一化
y_pred_actual = scaler.inverse_transform(y_pred_scaled)
y_test_actual = scaler.inverse_transform(y_test)

# 对应日期
test_dates = raw.index[SEQ_LEN + split: SEQ_LEN + split + len(y_test)]

fig, ax = plt.subplots(figsize=(13, 5))
ax.plot(test_dates, y_test_actual, label='真实价格', linewidth=1.5, color='steelblue')
ax.plot(test_dates, y_pred_actual, label='LSTM 预测', linewidth=1.5,
         color='orange', linestyle='--')
ax.set_title('LSTM 预测 vs 真实价格（测试集）', fontsize=13)
ax.set_ylabel('SPY 价格 (USD)')
ax.legend()
ax.grid(alpha=0.3)
plt.tight_layout()
plt.show()

# RMSE
rmse = np.sqrt(np.mean((y_pred_actual - y_test_actual) ** 2))
print(f'RMSE: {rmse:.4f} USD')
print('  说明：价格误差与交易目标不同，还需要检查方向、成本和基线。')

model.eval() with torch.no_grad(): y_pred_scaled = model(X_test_t).cpu().numpy() # 反归一化 y_pred_actual = scaler.inverse_transform(y_pred_scaled) y_test_actual = scaler.inverse_transform(y_test) # 对应日期 test_dates = raw.index[SEQ_LEN + split: SEQ_LEN + split + len(y_test)] fig, ax = plt.subplots(figsize=(13, 5)) ax.plot(test_dates, y_test_actual, label='真实价格', linewidth=1.5, color='steelblue') ax.plot(test_dates, y_pred_actual, label='LSTM 预测', linewidth=1.5, color='orange', linestyle='--') ax.set_title('LSTM 预测 vs 真实价格（测试集）', fontsize=13) ax.set_ylabel('SPY 价格 (USD)') ax.legend() ax.grid(alpha=0.3) plt.tight_layout() plt.show() # RMSE rmse = np.sqrt(np.mean((y_pred_actual - y_test_actual) ** 2)) print(f'RMSE: {rmse:.4f} USD') print(' 说明：价格误差与交易目标不同，还需要检查方向、成本和基线。')

RMSE: 7.0842 USD
  说明：价格误差与交易目标不同，还需要检查方向、成本和基线。

6. 方向准确率（更实用的指标）¶

In [9]:

# 真实方向 vs 预测方向
actual_dir = np.sign(np.diff(y_test_actual.ravel()))
pred_dir   = np.sign(np.diff(y_pred_actual.ravel()))

dir_accuracy = (actual_dir == pred_dir).mean()
print(f'方向准确率: {dir_accuracy:.2%}')
print(f'基准（随机猜测）: 50%')

if dir_accuracy > 0.52:
    print('\n样本中的方向准确率略高于基线；需要在更多资产和时段验证。')
else:
    print('\n方向准确率接近随机；当前结果不支持模型具有可用的方向预测能力。')
    print('较低的价格预测误差不代表模型能够产生可交易的方向信号。')

# 真实方向 vs 预测方向 actual_dir = np.sign(np.diff(y_test_actual.ravel())) pred_dir = np.sign(np.diff(y_pred_actual.ravel())) dir_accuracy = (actual_dir == pred_dir).mean() print(f'方向准确率: {dir_accuracy:.2%}') print(f'基准（随机猜测）: 50%') if dir_accuracy > 0.52: print('\n样本中的方向准确率略高于基线；需要在更多资产和时段验证。') else: print('\n方向准确率接近随机；当前结果不支持模型具有可用的方向预测能力。') print('较低的价格预测误差不代表模型能够产生可交易的方向信号。')

方向准确率: 50.45%
基准（随机猜测）: 50%

方向准确率接近随机；当前结果不支持模型具有可用的方向预测能力。
较低的价格预测误差不代表模型能够产生可交易的方向信号。

7. 局限性总结¶

问题	说明
趋势跟踪惯性	LSTM 容易学到「预测今天≈昨天」，价格图好看但无实用
非平稳性	市场机制随着时间改变，过去的模式不保证未来有效
数据量不足	相对于 NLP/CV，金融日线数据量极少（~2500条/10年）
过拟合风险	参数量 >> 样本量时，容易记忆噪声而非规律

实践建议：在金融场景，XGBoost + 精选因子 通常比 LSTM 更稳定。 LSTM 的优势更多体现在高频交易（tick数据） 和**另类数据（NLP情绪）**领域。

练习¶

---

练习¶

1. 将 LSTM 的输入特征从「仅价格」扩展到「价格 + RSI + MACD + 成交量」，对比模型预测效果是否改善。
2. 尝试用 Transformer（如 torch.nn.Transformer）替换 LSTM 层，观察训练速度和效果的差异。
3. 实施 Walk-Forward Validation：将数据分为多段，每段分别训练并在下一段测试，汇总整体夏普比率。
4. 将 SEQ_LEN 从 60 改为 20 / 120，观察预测精度的变化。

---

下一节 → 05_cross_validation.ipynb

In [ ]: