时间序列预测实战：LSTM与Transformer的PyTorch实现与对比

时间序列预测是机器学习里最常被练手、也最容易被问出细节的一类任务。不管做风功率预测、设备故障预警、销量预测，还是时序指标监控，最后都会遇到同一个问题：用 LSTM 还是 Transformer？这次我们直接把两个模型放在一起，从论文核心思路讲到 PyTorch 代码复现，再跑训练、验证、批量预测和接口封装，最后给出一套可以落地到实际项目里的选型建议。文章偏实战，建议打开编辑器跟着敲一遍。

先给结论：LSTM 是循环神经网络的代表，1997 年提出的门控机制解决了一代 RNN 的梯度消失问题；Transformer 是 2017 年《Attention Is All You Need》提出的架构，用自注意力取代循环结构，擅长并行计算和长序列依赖建模。两者在过去几年里几乎成了时间序列预测论文里的标配 baseline，也是面试和期末复习里最容易同时出现的两个模型。本文不偏袒某一个，只做对比验证。

文章会按照“核心能力速览 -> 环境准备 -> 代码复现 -> 训练验证 -> API 封装 -> 性能观察 -> 问题排查”的顺序展开，适合正在入门机器学习、准备论文复现，或者想在真实时序数据上快速搭建预测模型的读者。建议全程保留一份干净的数据集，先把流程跑通，再逐步调整参数。

1. 核心能力速览

需要说明的是，LSTM 和 Transformer 本身都是通用架构，在不同数据集上的表现差异很大。没有哪个模型“绝对更好”，只有“在特定数据规模和任务上更合适”。本文提供的是对比验证框架，最终效果要以你自己的数据集为准。

2. 适用场景与使用边界

LSTM 和 Transformer 在时间序列预测里的定位不同，适用场景可以从下面几个维度来判断。

LSTM 适合中等长度序列、样本量不是特别大的场景。比如 24 点到 48 点的滑窗预测、设备传感器数据、短期电力负荷预测。这类任务里 LSTM 结构简单、收敛速度快、对数据量的要求低于 Transformer。它的劣势也很明显：必须按时间顺序逐步处理，训练速度慢，超长序列上的长期依赖能力不如注意力机制。

Transformer 适合长序列、强依赖场景。比如 96 点以上输入、96 点以上输出的长时预测；或者输入序列内部存在明显的周期性和多尺度依赖。自注意力机制能够同时看到整个序列里所有位置的关系，在长序列上通常比 LSTM 更稳。劣势是模型参数多，在小数据集上容易过拟合，训练时对学习率和正则化更敏感。

使用边界上必须注意几点：

• 涉及金融行情、医疗指标、个人隐私数据时，只能用于合法的研究或内部测试，不能直接用于投资建议、诊断结论等敏感决策。
• 训练数据如果包含第三方采集的样本，需要确认数据授权范围，否则不要公开传播。
• 预测模型存在误差，真实系统里要预留观测纠偏机制，不能盲目信任模型输出。

3. 环境准备与前置条件

本文以 PyTorch 为主要框架，不依赖复杂的外部服务。建议先用 conda 建一个干净的虚拟环境，避免依赖冲突。

3.1 操作系统与 Python 版本

Windows、Linux、macOS 都可以运行。差异点主要在 PyTorch 的 CUDA 支持上，Linux 下 GPU 驱动和 PyTorch 版本匹配最容易处理。Python 版本建议 3.8 到 3.11，不要用太新的版本，部分旧项目依赖不支持。

3.2 GPU 与 CPU 要求

CPU 可以完成本文章所有训练和推理流程，只是训练时间更长。如果使用 GPU，需要先确认显卡驱动版本，再安装对应 CUDA 版本的 PyTorch。显存占用主要由三个因素决定：序列长度、batch_size、模型维度。4G 到 8G 显存足够跑本文的示例配置，实际占用需要在本机测试后确认。

3.3 依赖清单

安装命令如下：

conda create -n ts_forecast python=3.10 -y conda activate ts_forecast pip install torch numpy pandas scikit-learn matplotlib fastapi uvicorn tqdm

如果是 NVIDIA 显卡环境，建议先到 PyTorch 官网选择对应 CUDA 版本的安装命令，例如：

pip install torch --index-url https://download.pytorch.org/whl/cu118

4. 项目结构与数据准备

实际项目建议把代码、数据、模型、结果分开管理，避免后面跑多个实验时互相覆盖。推荐目录结构如下：

time_series_forecast/ ├── data/ │ ├── train.csv │ └── test.csv ├── models/ │ ├── __init__.py │ ├── lstm_model.py │ └── transformer_model.py ├── utils/ │ ├── data_loader.py │ └── metrics.py ├── train.py ├── predict.py ├── batch_predict.py └── api_server.py

4.1 数据加载与归一化

时间序列预测的第一步是把原始 CSV 变成模型能学习的窗口样本。这里以单变量预测为例，多变量预测需要额外处理特征维度。

import numpy as np import pandas as pd from sklearn.preprocessing import MinMaxScaler def load_data(file_path, target_col): df = pd.read_csv(file_path) values = df[target_col].values.reshape(-1, 1) scaler = MinMaxScaler() scaled = scaler.fit_transform(values) return scaled, scaler def create_sequences(data, seq_len=24, pred_len=1): xs, ys = [], [] for i in range(len(data) - seq_len - pred_len + 1): x = data[i : i + seq_len] y = data[i + seq_len : i + seq_len + pred_len] xs.append(x) ys.append(y) return np.array(xs, dtype=np.float32), np.array(ys, dtype=np.float32)

这里需要注意：seq_len是输入窗口长度，pred_len是预测长度。单步预测时pred_len=1，预测的是序列下一个点；多步预测时pred_len大于 1，模型输出维度也要对应调整。

4.2 训练集和验证集划分

时间序列数据不能像图像分类那样随机洗牌，必须按时间顺序切分，否则会引入未来信息泄漏。

def split_data(scaled, train_ratio=0.8): n_train = int(len(scaled) * train_ratio) train_data = scaled[:n_train] test_data = scaled[n_train:] return train_data, test_data

实际操作中，切分后再调用create_sequences构造窗口样本，并转成 PyTorch 的DataLoader。

from torch.utils.data import DataLoader, TensorDataset import torch def make_loader(data, seq_len=24, pred_len=1, batch_size=32, shuffle=False): xs, ys = create_sequences(data, seq_len, pred_len) dataset = TensorDataset(torch.FloatTensor(xs), torch.FloatTensor(ys)) loader = DataLoader(dataset, batch_size=batch_size, shuffle=shuffle) return loader

5. 模型代码复现

5.1 LSTM 模型定义

LSTM 的核心是门控机制：输入门、遗忘门、输出门和细胞状态。PyTorch 的nn.LSTM已经把门控计算封装好了，我们只需要在最后一层接全连接层输出预测值。

import torch.nn as nn class LSTMPredictor(nn.Module): def __init__(self, input_size=1, hidden_size=64, num_layers=2, output_size=1, dropout=0.1): super().__init__() self.lstm = nn.LSTM( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, batch_first=True, dropout=dropout ) self.fc = nn.Linear(hidden_size, output_size) def forward(self, x): out, _ = self.lstm(x) out = out[:, -1, :] out = self.fc(out) return out

关键点在于batch_first=True，输入张量形状是[batch, seq_len, input_size]。out[:, -1, :]取最后一个时间步的隐藏状态，再接全连接输出预测值。多步预测时，output_size设为pred_len即可，也可以改成逐时间步解码的结构。

5.2 Transformer 模型定义

Transformer 需要两部分：位置编码和编码器层。时间序列不同于 NLP，没有天然的位置顺序，所以必须加上位置编码，让模型感知时间步的前后关系。

import numpy as np import torch import torch.nn as nn class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=5000): super().__init__() pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp( torch.arange(0, d_model, 2, dtype=torch.float) * (-np.log(10000.0) / d_model) ) pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) pe = pe.unsqueeze(0) self.register_buffer('pe', pe) def forward(self, x): return x + self.pe[:, :x.size(1), :]

class TransformerPredictor(nn.Module): def __init__(self, input_size=1, d_model=64, nhead=4, num_layers=2, output_size=1, dropout=0.1): super().__init__() self.embedding = nn.Linear(input_size, d_model) self.pos_encoding = PositionalEncoding(d_model) encoder_layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, batch_first=True, dropout=dropout ) self.transformer_encoder = nn.TransformerEncoder( encoder_layer, num_layers=num_layers ) self.fc = nn.Linear(d_model, output_size) def forward(self, x): x = self.embedding(x) x = self.pos_encoding(x) x = self.transformer_encoder(x) x = x[:, -1, :] x = self.fc(x) return x

这里使用的是 Transformer 的编码器部分做预测。d_model是嵌入维度，nhead是多头注意力头数，num_layers是编码器层数。小数据量场景建议d_model=32或64，num_layers=1或2，避免过拟合。

6. 训练流程与效果验证

6.1 训练函数

训练过程统一使用 MSE 损失和 Adam 优化器。为了方便复现，设置固定随机种子。

import torch.optim as optim def set_seed(seed=42): np.random.seed(seed) torch.manual_seed(seed) if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed) def train_epoch(model, loader, criterion, optimizer, device): model.train() total_loss = 0.0 total_samples = 0 for xb, yb in loader: xb = xb.to(device) yb = yb.to(device) optimizer.zero_grad() pred = model(xb) loss = criterion(pred, yb) loss.backward() optimizer.step() total_loss += loss.item() * xb.size(0) total_samples += xb.size(0) return total_loss / total_samples def evaluate(model, loader, criterion, device): model.eval() total_loss = 0.0 total_samples = 0 with torch.no_grad(): for xb, yb in loader: xb = xb.to(device) yb = yb.to(device) pred = model(xb) loss = criterion(pred, yb) total_loss += loss.item() * xb.size(0) total_samples += xb.size(0) return total_loss / total_samples def train_model(model, train_loader, val_loader, epochs=50, lr=1e-3, device='cpu'): criterion = nn.MSELoss() optimizer = optim.Adam(model.parameters(), lr=lr) best_val = float('inf') for epoch in range(1, epochs + 1): train_loss = train_epoch(model, train_loader, criterion, optimizer, device) val_loss = evaluate(model, val_loader, criterion, device) if epoch % 5 == 0 or epoch == 1: print(f"Epoch {epoch:3d} | Train Loss: {train_loss:.6f} | Val Loss: {val_loss:.6f}") if val_loss < best_val: best_val = val_loss torch.save(model.state_dict(), "best_model.pt") print(f"Best Val Loss: {best_val:.6f}")

训练时可以使用nvidia-smi观察显存变化，也可以使用torch.cuda.max_memory_allocated()获取当前进程分配的最大显存：

if torch.cuda.is_available(): print(torch.cuda.max_memory_allocated(device) / 1024 ** 2, "MB")

6.2 单步预测测试

单步预测是验证模型是否“学进去”的第一步。训练集和测试集都按时间顺序构造样本，训练完成后直接用测试集最后一个窗口做预测。

def predict_next(model, scaler, history, device='cpu'): model.eval() arr = np.array(history).reshape(1, -1, 1).astype(np.float32) tensor = torch.FloatTensor(arr).to(device) with torch.no_grad(): pred = model(tensor) pred = pred.cpu().numpy().reshape(-1, 1) return scaler.inverse_transform(pred)

判断标准：把预测值反归一化后，与真实值对比，观察误差量级是否可接受。建议在测试集上计算 MSE、MAE、MAPE 三个指标。

from sklearn.metrics import mean_squared_error, mean_absolute_error def calc_metrics(y_true, y_pred): mse = 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 + 1e-8))) * 100 return {"MSE": mse, "MAE": mae, "MAPE": mape}

6.3 多步预测测试

多步预测有两种常用方式：迭代预测和多输出预测。迭代预测把上一步的输出作为下一步输入，简单但误差会累积；多输出预测一次性输出未来多个点，结构更直接。以下是迭代预测的示例：

def predict_multi_step(model, scaler, history, pred_len=12, device='cpu'): model.eval() input_seq = [float(x) for x in history] predictions = [] for _ in range(pred_len): arr = np.array(input_seq[-24:]).reshape(1, -1, 1).astype(np.float32) tensor = torch.FloatTensor(arr).to(device) with torch.no_grad(): next_val = model(tensor).cpu().numpy()[0, 0] predictions.append(next_val) input_seq.append(next_val) pred_np = np.array(predictions).reshape(-1, 1) return scaler.inverse_transform(pred_np)

多步预测最能检验模型的稳定性。如果预测曲线在几步之后开始明显偏离真实规律，说明模型没有学到足够的时序依赖，或者输入特征太少。

6.4 LSTM 与 Transformer 对比测试

在相同数据、相同seq_len和相同训练轮数下对比两个模型。建议先固定参数：

从训练经验看，Transformer 通常在训练早期收敛更快，但在小数据上更容易出现验证集 loss 反弹，即过拟合。LSTM 收敛相对慢，但往往更稳定。具体表现需要看训练日志和评估指标曲线。

6.5 结果可视化

训练完成后，把真实值和预测值画在同一张图里，能直观判断滞后程度和整体拟合偏差。

import matplotlib.pyplot as plt def plot_result(y_true, y_pred, save_path="result.png"): plt.figure(figsize=(12, 5)) plt.plot(y_true, label="True") plt.plot(y_pred, label="Pred") plt.legend() plt.grid(True) plt.savefig(save_path, dpi=120) plt.close()

7. 接口 API 与批量任务

训练好的模型要接到业务系统里，最直接的方式是封装成 HTTP 接口。这里用 FastAPI 封装两个核心接口：单次预测和批量预测。

7.1 接口服务

from fastapi import FastAPI from pydantic import BaseModel import numpy as np import torch app = FastAPI(title="Time Series Prediction API") lstm_model = None transformer_model = None scaler = None class PredictRequest(BaseModel): data: list[float] seq_len: int = 24 model_type: str = "lstm" class BatchPredictRequest(BaseModel): input_csv: str output_csv: str seq_len: int = 24 pred_len: int = 12 model_type: str = "transformer" @app.post("/predict") def predict(req: PredictRequest): model = lstm_model if req.model_type == "lstm" else transformer_model input_seq = req.data[-req.seq_len:] arr = np.array(input_seq).reshape(1, -1, 1).astype(np.float32) tensor = torch.FloatTensor(arr) with torch.no_grad(): pred = model(tensor).squeeze().numpy().tolist() return {"model_type": req.model_type, "prediction": pred}

启动接口服务：

uvicorn api_server:app --host 127.0.0.1 --port 8000

调用示例：

curl -X POST http://127.0.0.1:8000/predict \ -H "Content-Type: application/json" \ -d '{"data": [0.1, 0.2, 0.3, 0.4, 0.5], "seq_len": 5, "model_type": "lstm"}'

7.2 批量预测脚本

批量预测适合离线处理一整份 CSV，例如对测试集每 24 个点预测下 12 个点，然后把结果写回 CSV。

import argparse import pandas as pd import torch def batch_predict(model, scaler, df, target_col, seq_len=24, pred_len=12, device='cpu'): model.eval() data = scaler.transform(df[target_col].values.reshape(-1, 1)) results = [] for i in range(0, len(data) - seq_len - pred_len + 1, pred_len): input_seq = data[i : i + seq_len] real_seq = data[i + seq_len : i + seq_len + pred_len] arr = input_seq.reshape(1, -1, 1).astype(np.float32) tensor = torch.FloatTensor(arr).to(device) with torch.no_grad(): pred = model(tensor).cpu().numpy().reshape(-1, 1) row = { "start_idx": i, "prediction": scaler.inverse_transform(pred).flatten().tolist(), "actual": scaler.inverse_transform(real_seq).flatten().tolist() } results.append(row) return pd.DataFrame(results) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--input", type=str, required=True) parser.add_argument("--output", type=str, required=True) parser.add_argument("--seq_len", type=int, default=24) parser.add_argument("--pred_len", type=int, default=12) parser.add_argument("--model_path", type=str, required=True) args = parser.parse_args() # 加载模型和 scaler 的代码需要按实际项目补充 print(f"Input: {args.input}, Output: {args.output}")

批量任务注意点：滑窗步长和pred_len保持一致，避免重复预测同一段区间；大批量数据要控制内存，逐批写入结果文件，不要一次性把所有预测结果堆在内存里。

8. 资源占用与性能观察

资源占用是实际部署中最容易踩坑的部分。对时间序列预测来说，影响性能的主要因素不是图像分类那种超大显存需求，而是序列长度和 batch_size 的组合效应。

显存占用观察方法：

nvidia-smi

训练代码里也可以主动记录最大显存：

if torch.cuda.is_available(): peak_memory = torch.cuda.max_memory_allocated(device) / 1024 ** 2 print(f"Peak GPU Memory: {peak_memory:.2f} MB")

如果显存不足，优先降低batch_size，其次是降低hidden_size或d_model。Transformer 的注意力计算量和序列长度呈二次关系，所以当seq_len特别长时，显存上涨会非常明显。LSTM 的显存占用随seq_len线性增长，但训练时间是逐步展开的，速度更慢。

CPU 推理和 GPU 推理的差异需要实测对比。一般来说，小模型、短序列在 CPU 上也能跑得很快；一旦seq_len超过几百或者batch_size变大，GPU 加速的优势才会明显体现。部署时可以根据业务延迟要求选择推理设备。

影响训练速度的因素从大到小排列：样本总量、epochs、模型层数、序列长度、batch_size。第一次跑实验建议先用小参数配置，确认流程通了再放大。

9. 常见问题与排查方法

10. 最佳实践与使用建议

第一次实验不要追求精度，先把最小流程跑通。建议固定种子，用一个几百条数据的小 CSV，把 LSTM 和 Transformer 都训练 20 到 50 轮，确认代码无误后再换大数据集。

工程化层面有几个建议：

• 模型权重、归一化 scaler、配置参数统一保存，预测和部署时使用相同配置加载。
• 训练时间序列模型时不要打乱样本顺序，按时间切分训练集和验证集。
• 每次实验记录数据路径、模型参数、loss 曲线和最终指标，方便后续对比。
• 接口服务要限制访问范围，生产环境不要暴露到公网，避免被恶意调用。
• 批量任务要加日志和失败重试机制，防止中途异常导致结果不完整。
• 涉及人脸、声音、金融、医疗等敏感数据，务必确认授权和合规边界，不公开传播训练数据。

模型选型上，小数据优先试 LSTM，数据量大且序列足够长时再试 Transformer。也可以把 LSTM 的隐藏状态作为额外特征输入到 Transformer，或者把 Transformer 编码器输出接到 LSTM 上形成混合结构，这类变体在论文中很常见，但要从真实需求出发，不要为了堆结构而堆结构。

11. 总结与下一步

这次内容把 LSTM 和 Transformer 从论文核心思路到 PyTorch 代码复现、训练验证、接口封装完整走了一遍。两个模型最值得先验证的点是：在你的数据集上，单步预测的误差量级是多少，多步预测能稳定预测几步。最容易踩的坑是数据泄漏和维度不匹配，其次是 Transformer 在小数据上过拟合。建议先跑通 LSTM，再在相同数据上换 Transformer，保存实验记录后对比效果。

下一步可以尝试的方向：改造 Transformer 的时间特征嵌入，加入外部变量和周期特征；使用因果卷积或状态空间模型做对比；在多步预测任务里尝试 Informer、Autoformer 等长序列变体。选型之前，先用自己的数据完成一轮 baseline 对比，后续所有改造都有明确参照。