时间序列预测是机器学习里最常被练手、也最容易被问出细节的一类任务。不管做风功率预测、设备故障预警、销量预测还是时序指标监控最后都会遇到同一个问题用 LSTM 还是 Transformer这次我们直接把两个模型放在一起从论文核心思路讲到 PyTorch 代码复现再跑训练、验证、批量预测和接口封装最后给出一套可以落地到实际项目里的选型建议。文章偏实战建议打开编辑器跟着敲一遍。先给结论LSTM 是循环神经网络的代表1997 年提出的门控机制解决了一代 RNN 的梯度消失问题Transformer 是 2017 年《Attention Is All You Need》提出的架构用自注意力取代循环结构擅长并行计算和长序列依赖建模。两者在过去几年里几乎成了时间序列预测论文里的标配 baseline也是面试和期末复习里最容易同时出现的两个模型。本文不偏袒某一个只做对比验证。文章会按照“核心能力速览 - 环境准备 - 代码复现 - 训练验证 - API 封装 - 性能观察 - 问题排查”的顺序展开适合正在入门机器学习、准备论文复现或者想在真实时序数据上快速搭建预测模型的读者。建议全程保留一份干净的数据集先把流程跑通再逐步调整参数。1. 核心能力速览能力项说明项目类型时间序列预测模型代码复现与实战涉及模型LSTM、Transformer主要功能单步预测、多步预测、批量预测、模型对比、API 封装输入数据CSV/DataFrame 形式的单变量或多变量时间序列输出形式预测值、可视化曲线、预测结果 CSV运行环境Python 3.8PyTorch 1.12建议安装 CUDA 版 PyTorch硬件要求CPU 可运行GPU 可加速显存需求与 seq_len、batch_size 相关支持批量任务支持可滑窗批量预测也可跑批量接口接口能力使用 FastAPI 封装支持 POST 请求调参适合场景教学练手、论文复现、工业时序预测、数据竞赛 baseline需要说明的是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 依赖清单依赖库作用torch模型构建、训练、推理numpy数值计算和数据转换pandasCSV 数据读取与整理scikit-learn数据归一化、评估指标matplotlib预测结果可视化fastapi uvicorn接口服务封装tqdm训练进度条显示安装命令如下conda create -n ts_forecast python3.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/cu1184. 项目结构与数据准备实际项目建议把代码、数据、模型、结果分开管理避免后面跑多个实验时互相覆盖。推荐目录结构如下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.py4.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_len24, pred_len1): 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, dtypenp.float32), np.array(ys, dtypenp.float32)这里需要注意seq_len是输入窗口长度pred_len是预测长度。单步预测时pred_len1预测的是序列下一个点多步预测时pred_len大于 1模型输出维度也要对应调整。4.2 训练集和验证集划分时间序列数据不能像图像分类那样随机洗牌必须按时间顺序切分否则会引入未来信息泄漏。def split_data(scaled, train_ratio0.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_len24, pred_len1, batch_size32, shuffleFalse): xs, ys create_sequences(data, seq_len, pred_len) dataset TensorDataset(torch.FloatTensor(xs), torch.FloatTensor(ys)) loader DataLoader(dataset, batch_sizebatch_size, shuffleshuffle) return loader5. 模型代码复现5.1 LSTM 模型定义LSTM 的核心是门控机制输入门、遗忘门、输出门和细胞状态。PyTorch 的nn.LSTM已经把门控计算封装好了我们只需要在最后一层接全连接层输出预测值。import torch.nn as nn class LSTMPredictor(nn.Module): def __init__(self, input_size1, hidden_size64, num_layers2, output_size1, dropout0.1): super().__init__() self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue, dropoutdropout ) 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_firstTrue输入张量形状是[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_len5000): super().__init__() pe torch.zeros(max_len, d_model) position torch.arange(0, max_len, dtypetorch.float).unsqueeze(1) div_term torch.exp( torch.arange(0, d_model, 2, dtypetorch.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_size1, d_model64, nhead4, num_layers2, output_size1, dropout0.1): super().__init__() self.embedding nn.Linear(input_size, d_model) self.pos_encoding PositionalEncoding(d_model) encoder_layer nn.TransformerEncoderLayer( d_modeld_model, nheadnhead, batch_firstTrue, dropoutdropout ) self.transformer_encoder nn.TransformerEncoder( encoder_layer, num_layersnum_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_model32或64num_layers1或2避免过拟合。6. 训练流程与效果验证6.1 训练函数训练过程统一使用 MSE 损失和 Adam 优化器。为了方便复现设置固定随机种子。import torch.optim as optim def set_seed(seed42): 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, epochs50, lr1e-3, devicecpu): criterion nn.MSELoss() optimizer optim.Adam(model.parameters(), lrlr) 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(fEpoch {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(fBest 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, devicecpu): 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_len12, devicecpu): 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和相同训练轮数下对比两个模型。建议先固定参数参数LSTMTransformerseq_len2424pred_len11hidden_size64d_model64num_layers22nhead-4epochs5050lr1e-31e-3从训练经验看Transformer 通常在训练早期收敛更快但在小数据上更容易出现验证集 loss 反弹即过拟合。LSTM 收敛相对慢但往往更稳定。具体表现需要看训练日志和评估指标曲线。6.5 结果可视化训练完成后把真实值和预测值画在同一张图里能直观判断滞后程度和整体拟合偏差。import matplotlib.pyplot as plt def plot_result(y_true, y_pred, save_pathresult.png): plt.figure(figsize(12, 5)) plt.plot(y_true, labelTrue) plt.plot(y_pred, labelPred) plt.legend() plt.grid(True) plt.savefig(save_path, dpi120) plt.close()7. 接口 API 与批量任务训练好的模型要接到业务系统里最直接的方式是封装成 HTTP 接口。这里用 FastAPI 封装两个核心接口单次预测和批量预测。7.1 接口服务from fastapi import FastAPI from pydantic import BaseModel import numpy as np import torch app FastAPI(titleTime 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_len24, pred_len12, devicecpu): 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, typestr, requiredTrue) parser.add_argument(--output, typestr, requiredTrue) parser.add_argument(--seq_len, typeint, default24) parser.add_argument(--pred_len, typeint, default12) parser.add_argument(--model_path, typestr, requiredTrue) args parser.parse_args() # 加载模型和 scaler 的代码需要按实际项目补充 print(fInput: {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(fPeak 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. 常见问题与排查方法问题现象可能原因排查方式解决方案训练 Loss 为 NaN学习率过大、数据未归一化或包含 NaN检查数据空值查看前几轮 Loss降低学习率增加归一化步骤清洗异常值Transformer 过拟合严重训练数据太少、模型参数量过大对比训练 Loss 和验证 Loss减小 d_model/layers增加 dropout减少训练轮数LSTM 训练特别慢seq_len 过长、数据量大、CPU 推理检查训练日志耗时减小 batch_size使用 GPU或截断序列长度预测曲线滞后严重使用了迭代预测的累计误差比较单步预测和多步预测误差改用多输出结构或加入额外特征修正端口 8000 被占用uvicorn 端口冲突查看端口占用进程换一个端口启动例如 8001API 返回 500 错误模型未加载或输入长度不匹配查看服务端日志在请求处理前加载模型校验输入维度验证集误差低于训练集Dropout 在训练时生效、验证时关闭检查模型模式切换确保 model.train() 和 model.eval() 配对使用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 对比后续所有改造都有明确参照。