简介：本资源是一份面向Python初学者与数据科学入门者的LSTM时间序列预测实战教程，聚焦股票价格、销售趋势、气象预报等典型场景，系统讲解如何用深度学习解决时序建模问题。压缩包共15个文件（391KB），含6个核心Python脚本（如train_lstm.py、train_lstm_multivariate.py实现单变量/多变量预测）、4张关键流程图与结构示意图（jpg）、2份Markdown说明文档（中英文README）、2个真实时序数据集（CSV）及1个.gitignore配置文件，覆盖数据预处理、滑动窗口构造、标准化、模型搭建、训练验证与反向还原全流程。已有39844人学习下载，资源结构清晰、代码即开即用，附带多变量预测扩展实现与输入数组/CSV双模式测试脚本，便于读者理解LSTM输入格式设计逻辑，并快速迁移至自有业务数据。

1. 时间序列预测不是“把数据喂给LSTM就完事”：一个能跑通、能调参、能上线的Python实现闭环

你是不是也试过：下载了网上一堆标着“LSTM时间序列预测”的Python脚本，一跑就报错ValueError: Input 0 is incompatible with layer lstm_1: expected ndim=3, found ndim=2？或者训练完loss掉得飞快，但预测曲线却平得像尺子量过？这不是你代码写错了——是绝大多数公开资源根本没告诉你：LSTM对输入形状、滑动窗口长度、归一化方式、状态重置逻辑这四点极其敏感，漏掉任意一个，模型就从“智能预测”退化成“高级平均器”。这份资源不是教学Demo，而是一个经过某高校实验室在电力负荷、工业传感器、气象数据三类真实时序任务中反复验证的最小可运行闭环：从原始CSV读入→滑动窗口构造→Min-Max+差分双层归一化→LSTM状态管理→多步滚动预测→反归一化还原。它不依赖任何云服务或私有API，纯本地PyTorch/TensorFlow双后端支持，所有参数都带中文注释，连lookback_window=24这种值背后对应“用过去24小时预测下一小时”都写清楚。适合正在做课程设计、毕设、或需要快速验证业务指标趋势的工程师——别再被玄学调参拖垮进度了。

2. LSTM为什么必须用滑动窗口？从数学本质到代码实现的三层拆解

2.1 时间序列的“记忆断层”问题：为什么不能直接喂原始序列？

LSTM的本质是处理有序依赖关系，但它无法理解“第100个点和第1个点之间隔着99个时间步”这种全局时序语义。如果你把整段1000点的序列直接reshape成(1000, 1)丢进LSTM，模型会错误地认为第1点和第1000点是相邻时刻（因为batch维度被压缩了）。真实场景中，我们关心的是“基于最近N个观测值，预测未来M个值”，这个N就是滑动窗口长度（lookback window）。它不是超参数调优的起点，而是问题定义的第一步。比如预测服务器CPU使用率，若采样间隔是5分钟，那么lookback_window=12意味着用过去1小时的数据预测下一时刻——这个值必须由业务周期决定，而非随意设为64或128。

提示：窗口长度选太小（如lookback=5），模型学不到长期模式；选太大（如lookback=500），不仅显存爆炸，还会引入大量无关噪声。经验法则是：取业务周期的整数倍，再加20%冗余。例如风电功率日周期为288点（5分钟/点），则lookback=350比lookback=256更合理。

2.2 构造滑动窗口的两种实现：NumPy切片 vs. Pandas.shift()

核心逻辑是：对长度为T的序列y，生成(T-lookback, lookback, 1)的三维数组X和(T-lookback, 1)的y_true。下面给出两种生产环境常用写法，区别在于内存占用与可读性：

import numpy as np import pandas as pd def create_dataset_numpy(data, lookback): """ NumPy版：内存高效，适合大数据集 data: (T,) 一维数组 return: X (T-lookback, lookback, 1), y (T-lookback, 1) """ X, y = [], [] for i in range(lookback, len(data)): # 取前lookback个点作为输入，当前点作为标签 X.append(data[i-lookback:i].reshape(-1, 1)) y.append(data[i]) return np.array(X), np.array(y).reshape(-1, 1) # 示例：1000点序列，窗口=50 → 输出X.shape=(950, 50, 1) data = np.random.randn(1000) X_np, y_np = create_dataset_numpy(data, lookback=50) print(f"NumPy版输出形状: X={X_np.shape}, y={y_np.shape}")

这段代码的关键在.reshape(-1, 1)：它强制将每个窗口内的50个标量转为(50, 1)列向量，确保LSTM输入是三维张量。如果漏掉.reshape(-1, 1)，X会是(950, 50)二维数组，后续model.fit()必然报错expected ndim=3。

def create_dataset_pandas(data, lookback): """ Pandas版：可读性强，支持多列特征扩展 data: pd.Series 或 pd.DataFrame """ df = pd.DataFrame(data) # 创建滞后特征：df['t-1'], df['t-2'], ..., df['t-lookback'] for i in range(1, lookback + 1): df[f't-{i}'] = df[0].shift(i) df.dropna(inplace=True) # 删除含NaN的行（前lookback行） # 输入列：t-1 到 t-lookback；标签列：t（即原序列当前值） feature_cols = [f't-{i}' for i in range(1, lookback + 1)] X_pd = df[feature_cols].values.reshape(-1, lookback, 1) y_pd = df[0].values.reshape(-1, 1) return X_pd, y_pd # 示例：等效于NumPy版，但后续加温度、湿度特征只需改df[0]为df[['cpu','temp','hum']] X_pd, y_pd = create_dataset_pandas(pd.Series(data), lookback=50) print(f"Pandas版输出形状: X={X_pd.shape}, y={y_pd.shape}")

Pandas版的优势在于shift()天然支持多变量。比如你要同时用CPU、内存、磁盘IO预测负载，只需传入pd.DataFrame({'cpu':cpu_data, 'mem':mem_data, 'disk':disk_data})，shift()会自动为每列生成滞后项，最后feature_cols包含所有变量的滞后特征，这才是工业级时序建模的起点。

2.3 窗口长度与预测步长的耦合关系：为什么单步预测不能直接外推？

很多初学者以为训练好单步LSTM（预测t+1）后，只要把预测值不断喂回模型就能实现多步预测（t+2,t+3...）。这是危险的！因为：

• 每次预测都携带误差，误差会指数级累积（类似蝴蝶效应）；
• 模型从未见过“自己预测的值”作为输入，分布偏移严重。

正确做法是训练时就构造多步目标。例如预测未来3小时（每5分钟一采样，共36步），则y不再是(T-lookback, 1)，而是(T-lookback, 36)：

def create_multistep_dataset(data, lookback, forecast_steps): """ 多步预测窗口构造 data: (T,) 数组 forecast_steps: 要预测的步数，如36（3小时） return: X (T-lookback-forecast_steps+1, lookback, 1), y (T-lookback-forecast_steps+1, forecast_steps) """ X, y = [], [] # 注意：最后一个有效窗口起始位置是 T-lookback-forecast_steps for i in range(lookback, len(data) - forecast_steps + 1): X.append(data[i-lookback:i].reshape(-1, 1)) # 取接下来forecast_steps个点作为多步标签 y.append(data[i:i+forecast_steps]) return np.array(X), np.array(y) X_multi, y_multi = create_multistep_dataset(data, lookback=50, forecast_steps=36) print(f"多步版输出: X={X_multi.shape}, y={y_multi.shape}") # X=(915,50,1), y=(915,36)

这里len(data) - forecast_steps + 1是关键边界——如果忽略+1，最后一行数据会被截断。这个细节在Kaggle竞赛中常导致线下验证和线上提交结果不一致，血泪经验。

3. 归一化不是“套个MinMaxScaler就完事”：差分+缩放双保险的工程实践

3.1 为什么单靠MinMaxScaler会失效？看一个真实翻车案例

某公司用LSTM预测产线良率，原始数据范围是[92.3, 99.7]。他们直接用MinMaxScaler(feature_range=(0,1))归一化，训练loss降到0.001，但部署后预测值全在0.95±0.01区间震荡，完全无法捕捉92.3→93.1这种关键下跌。原因在于：MinMaxScaler只解决量纲问题，不解决趋势漂移。当序列存在缓慢上升/下降趋势（如设备老化导致良率逐年下降），归一化后的数据仍保留该趋势，而LSTM对长期趋势的建模能力极弱，容易过拟合短期波动。

3.2 差分（Differencing）：让序列变平稳的物理意义

差分的本质是计算变化量而非绝对值。对序列y_t，一阶差分Δy_t = y_t - y_{t-1}。它把“预测良率值”转化为“预测良率变化量”，大幅降低趋势干扰。更重要的是，差分具有明确物理意义：Δy_t > 0表示良率提升，Δy_t < 0表示恶化，这对产线告警系统至关重要。

def apply_differencing(data, order=1): """ 对序列进行order阶差分 data: (T,) 数组 return: diff_data (T-order,) 数组，以及用于还原的首order个原始值 """ diff_data = data.copy() history = [] # 存储每次差分前的首元素，用于逆差分 for _ in range(order): history.append(diff_data[0]) diff_data = np.diff(diff_data) # 一阶差分 return diff_data, history # 原始数据（模拟缓慢上升趋势） raw_data = np.linspace(92.0, 99.0, 1000) + np.random.normal(0, 0.2, 1000) diff_data, history = apply_differencing(raw_data, order=1) print(f"原始数据范围: [{raw_data.min():.2f}, {raw_data.max():.2f}]") print(f"差分后范围: [{diff_data.min():.2f}, {diff_data.max():.2f}]") # 范围缩小10倍以上

注意：np.diff()会减少长度，order=1时输出长度为T-1。history列表存储了差分过程中丢失的首元素，这是逆差分（还原预测值）的必要条件。

3.3 差分+MinMaxScaler的串联流程与代码实现

正确顺序是：先差分 → 再归一化。如果反过来，归一化后的数据再差分，会放大归一化引入的数值误差。

from sklearn.preprocessing import MinMaxScaler def preprocess_pipeline(data, lookback, forecast_steps, diff_order=1, scaler_range=(0,1)): """ 完整预处理流水线 """ # 步骤1：差分 diff_data, history = apply_differencing(data, order=diff_order) # 步骤2：归一化（仅对差分后数据） scaler = MinMaxScaler(feature_range=scaler_range) scaled_diff = scaler.fit_transform(diff_data.reshape(-1, 1)).flatten() # 步骤3：构造滑动窗口（对scaled_diff操作） X, y = create_multistep_dataset(scaled_diff, lookback, forecast_steps) return X, y, scaler, history, diff_order # 执行预处理 X_proc, y_proc, scaler, history, diff_order = preprocess_pipeline( raw_data, lookback=50, forecast_steps=36, diff_order=1 ) print(f"预处理后X形状: {X_proc.shape}, y形状: {y_proc.shape}") print(f"归一化器范围: {scaler.data_min_[0]:.3f} ~ {scaler.data_max_[0]:.3f}")

这里scaler.fit_transform().flatten()确保输出是一维数组，避免后续create_multistep_dataset出错。scaler对象必须保存下来，因为预测后要先用scaler.inverse_transform()还原差分值，再用history做逆差分。

4. LSTM模型构建与训练：状态管理、损失函数、早停策略的硬核配置

4.1 PyTorch版LSTM：为什么必须手动管理hidden state？

TensorFlow/Keras的return_sequences=True隐藏了状态管理细节，但在PyTorch中，LSTM的hidden state和cell state必须显式传递。否则模型会默认每次forward都重置状态，导致无法学习长期依赖。

import torch import torch.nn as nn class LSTMModel(nn.Module): def __init__(self, input_size=1, hidden_size=50, num_layers=2, output_size=1, dropout=0.2): super().__init__() self.hidden_size = hidden_size self.num_layers = num_layers self.lstm = nn.LSTM(input_size, hidden_size, 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, hidden=None): # x: (batch, seq_len, features) batch_size = x.size(0) if hidden is None: # 初始化hidden state：(num_layers, batch, hidden_size) h0 = torch.zeros(self.num_layers, batch_size, self.hidden_size) c0 = torch.zeros(self.num_layers, batch_size, self.hidden_size) hidden = (h0, c0) lstm_out, hidden = self.lstm(x, hidden) # lstm_out: (batch, seq_len, hidden_size) # 取最后一个时间步的输出（seq_len维度末尾） last_output = lstm_out[:, -1, :] # (batch, hidden_size) out = self.fc(last_output) # (batch, output_size) return out, hidden # 实例化模型 model = LSTMModel(input_size=1, hidden_size=64, num_layers=2, output_size=36) print(f"模型参数量: {sum(p.numel() for p in model.parameters())}")

关键点：

• hidden参数允许外部传入状态，实现跨batch状态延续（如在线学习）；
• lstm_out[:, -1, :]取最后一个时间步，因为我们要预测未来值，而非整个序列；
• output_size=36直接对应多步预测的36个点，避免循环预测。

4.2 损失函数选择：MAE vs. MSE vs. Huber Loss的场景适配

• MSE（均方误差）：对异常值敏感，适合数据干净、需惩罚大误差的场景（如金融风控）；
• MAE（平均绝对误差）：鲁棒性强，适合含脉冲噪声的工业传感器数据；
• Huber Loss：MSE与MAE的结合体，当误差小于δ时用MSE，大于δ时用MAE，δ=0.5是常用起点。

import torch.nn.functional as F def huber_loss(pred, target, delta=0.5): """PyTorch Huber Loss实现""" error = pred - target abs_error = torch.abs(error) quadratic = torch.clamp(abs_error, max=delta) linear = abs_error - quadratic return torch.mean(0.5 * quadratic ** 2 + delta * linear) # 训练循环中使用 criterion = lambda pred, target: huber_loss(pred, target, delta=0.3) optimizer = torch.optim.Adam(model.parameters(), lr=0.001) # 示例：模拟一次前向传播 dummy_x = torch.randn(32, 50, 1) # batch=32, seq=50, feat=1 dummy_y = torch.randn(32, 36) # batch=32, forecast=36 pred, _ = model(dummy_x) loss = criterion(pred, dummy_y) print(f"Huber Loss值: {loss.item():.4f}")

注意：Huber Loss的delta值需根据归一化后数据的典型误差范围设定。若scaler将数据缩放到[0,1]，delta=0.1意味着对小于10%的误差用平方惩罚，大于10%的用线性惩罚。

4.3 早停（Early Stopping）与学习率衰减：防止过拟合的双重保险

早停不是简单监控val_loss，而是要设置耐心值（patience）和最小增量（min_delta）。否则模型可能在val_loss微小波动时就停止，错过最佳点。

class EarlyStopping: def __init__(self, patience=10, min_delta=0.001, verbose=False): self.patience = patience self.min_delta = min_delta self.verbose = verbose self.counter = 0 self.best_score = None self.early_stop = False def __call__(self, val_loss): score = -val_loss # 因为我们要最大化score（最小化loss） if self.best_score is None: self.best_score = score elif score < self.best_score + self.min_delta: self.counter += 1 if self.verbose: print(f'EarlyStopping counter: {self.counter} out of {self.patience}') if self.counter >= self.patience: self.early_stop = True else: self.best_score = score self.counter = 0 # 使用示例 early_stopping = EarlyStopping(patience=15, min_delta=0.0005, verbose=True) scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau( optimizer, mode='min', factor=0.5, patience=5, verbose=True ) # 在训练循环中 for epoch in range(100): # ... train_step ... val_loss = validate(model, val_loader) scheduler.step(val_loss) # 学习率衰减 early_stopping(val_loss) # 早停检查 if early_stopping.early_stop: print("Early stopping triggered") break

ReduceLROnPlateau比固定学习率衰减更智能：只有当val_loss连续5个epoch不下降时，才将lr乘以0.5。这避免了在训练初期盲目降lr。

5. 预测结果还原与评估：从归一化值到业务指标的完整映射

5.1 逆差分（Inverse Differencing）：三步还原法

预测值是归一化后的差分值，需按顺序执行：

1. 逆归一化：用训练时保存的scaler还原为差分尺度；
2. 逆差分：用history列表逐步累加还原为原始尺度；
3. 对齐时间轴：注意差分导致的索引偏移。

def inverse_transform_prediction(scaler, pred_scaled, history, diff_order=1): """ 将归一化预测值还原为原始尺度 pred_scaled: (N, forecast_steps) 归一化后的预测差分值 history: 差分时保存的首元素列表 """ # 步骤1：逆归一化 → 得到差分值 pred_diff = scaler.inverse_transform(pred_scaled) # 步骤2：逆差分（核心！） # 初始化还原数组，长度=forecast_steps + diff_order pred_original = np.zeros(pred_diff.shape[0] + diff_order) # 填充history中的初始值（差分前的原始起点） for i in range(diff_order): pred_original[i] = history[i] # 逐点累加：pred_original[t] = pred_original[t-1] + pred_diff[t-diff_order] for t in range(diff_order, len(pred_original)): pred_original[t] = pred_original[t-1] + pred_diff[t-diff_order] # 返回预测的forecast_steps个点（跳过history部分） return pred_original[diff_order:] # 示例：还原单个样本的36步预测 sample_pred_scaled = np.random.rand(1, 36) # 模拟模型输出 pred_original = inverse_transform_prediction( scaler, sample_pred_scaled, history, diff_order=1 ) print(f"还原后预测长度: {len(pred_original)}, 值域: [{pred_original.min():.2f}, {pred_original.max():.2f}]")

这里pred_original[diff_order:]是关键——diff_order=1时，pred_original[0]是history存的起点，真正预测的是pred_original[1:]这36个点。

5.2 多步预测评估指标：不只是RMSE，还要看方向一致性

工业场景中，预测值的方向准确性（上涨/下跌判断）往往比绝对误差更重要。因此除RMSE、MAE外，必须计算：

def evaluate_forecast(y_true, y_pred): """ y_true, y_pred: (N, forecast_steps) 二维数组 return: dict of metrics """ n_samples, n_steps = y_true.shape metrics = {} # RMSE & MAE mse = np.mean((y_true - y_pred) ** 2) mae = np.mean(np.abs(y_true - y_pred)) metrics['RMSE'] = np.sqrt(mse) metrics['MAE'] = mae # Direction Accuracy：比较相邻点变化符号 # y_true_dir[i,j] = sign(y_true[i,j] - y_true[i,j-1])，j>0 y_true_dir = np.sign(np.diff(y_true, axis=1)) # (N, forecast_steps-1) y_pred_dir = np.sign(np.diff(y_pred, axis=1)) da = np.mean(y_true_dir == y_pred_dir) * 100 metrics['Direction_Accuracy_%'] = da # MAPE（过滤y_true为0的点） non_zero_mask = y_true != 0 mape = np.mean(np.abs((y_true[non_zero_mask] - y_pred[non_zero_mask]) / y_true[non_zero_mask])) * 100 metrics['MAPE_%'] = mape return metrics # 示例评估 y_true_sample = np.random.randn(10, 36) + 95.0 # 模拟真实值 y_pred_sample = y_true_sample + np.random.randn(10, 36) * 0.5 # 加噪声预测 results = evaluate_forecast(y_true_sample, y_pred_sample) for k, v in results.items(): print(f"{k}: {v:.3f}")

np.diff(..., axis=1)沿列方向（时间步）求差，得到变化方向。np.sign()将正负变化转为+1/-1/0，直接比较相等性即可统计方向准确率。

5.3 预测可视化：带置信区间的滚动预测图

单点预测不够，业务需要知道不确定性。这里用分位数回归思想：训练三个模型分别预测10%、50%、90%分位数，构成预测区间。

import matplotlib.pyplot as plt def plot_forecast_with_uncertainty(y_true, y_pred_lower, y_pred_median, y_pred_upper, lookback=50, forecast_steps=36): """ 绘制带置信区间的预测图 y_true: (T,) 真实序列 y_pred_*: (N, forecast_steps) 预测结果，N为预测次数（如滚动预测的起始点数） """ # 取最后一次预测（最远的未来） last_true = y_true[-forecast_steps:] last_lower = y_pred_lower[-1] last_median = y_pred_median[-1] last_upper = y_pred_upper[-1] plt.figure(figsize=(12, 5)) plt.plot(range(len(y_true)-forecast_steps, len(y_true)), last_true, label='True', color='black', linewidth=2) plt.plot(range(len(y_true)-forecast_steps, len(y_true)), last_median, label='Median', color='blue', linestyle='--') plt.fill_between(range(len(y_true)-forecast_steps, len(y_true)), last_lower, last_upper, alpha=0.3, color='blue', label='10%-90% CI') plt.xlabel('Time Step') plt.ylabel('Value') plt.title(f'LSTM Forecast with Uncertainty (lookback={lookback}, forecast={forecast_steps})') plt.legend() plt.grid(True) plt.show() # 示例调用（需先训练三个模型） # plot_forecast_with_uncertainty(raw_data, lower_preds, median_preds, upper_preds)

图中灰色区域是10%-90%置信区间，若真实值频繁突破该区间，说明模型不确定性估计不足，需增加dropout或集成多个模型。

6. 避坑指南：LSTM时间序列预测中五个必踩的“经典陷阱”

6.1 陷阱1：测试集泄露（Test Set Leakage）——你以为的“未来”其实是“过去”

现象：模型在测试集上RMSE低至0.01，但上线后预测完全失效。
原因：在构造滑动窗口前，对整个数据集（含测试集）做了归一化或差分。这导致测试集信息“泄漏”到训练过程，模型实际是在拟合已知分布而非预测未知值。
解决：严格按时间顺序划分数据——先切分train/val/test，再对训练集单独拟合scaler和差分history，测试集只能用训练集参数做transform，绝不能fit。代码中必须有scaler.fit(train_data)和scaler.transform(test_data)的明确分离。

6.2 陷阱2：窗口边界错位（Window Boundary Misalignment）——预测起点对不上

现象：预测曲线整体右移1个时间步，或与真实值完全不重叠。
原因：滑动窗口构造时，X[i]对应y[i]的索引计算错误。常见错误是for i in range(lookback, len(data))但y.append(data[i])，这导致X[0]用data[0:lookback]预测data[lookback]，而X[1]用data[1:lookback+1]预测data[lookback+1]，看似正确，但若forecast_steps>1，y[i]应为data[i:i+forecast_steps]，此时i的上限必须是len(data)-forecast_steps，否则越界。
解决：在create_multistep_dataset中强制校验i+forecast_steps <= len(data)，并在循环内加assert：

for i in range(lookback, len(data) - forecast_steps + 1): assert i + forecast_steps <= len(data), f"Index overflow at i={i}" X.append(...) y.append(data[i:i+forecast_steps])

6.3 陷阱3：LSTM输入维度混乱（Input Dimension Confusion）——三维变二维的静默失败

现象：训练不报错，但loss不下降，预测值全为常数。
原因：X本应是(N, lookback, features)，但误写成(N, lookback*features)。例如对单变量数据，X.reshape(-1, lookback)生成二维数组，LSTM接收后自动视为(batch, features)，lookback被吞掉。
解决：在model.fit()前打印X.shape并校验：

print(f"X shape before training: {X.shape}") assert len(X.shape) == 3, f"X must be 3D, got {len(X.shape)}D" assert X.shape[1] == lookback, f"X second dim must be lookback={lookback}, got {X.shape[1]}"

6.4 陷阱4：状态重置时机错误（Hidden State Reset Timing）——跨batch的记忆污染

现象：训练loss震荡剧烈，验证集性能忽高忽低。
原因：在每个batch结束时未重置LSTM hidden state，导致前一个batch的状态污染下一个batch。尤其在batch_first=False（PyTorch默认）时更易发生。
解决：在每个epoch开始时重置，或在dataloader迭代中显式管理：

for epoch in range(num_epochs): hidden = None # 每个epoch重置 for batch_idx, (x_batch, y_batch) in enumerate(train_loader): pred, hidden = model(x_batch, hidden) # ... loss计算 ... # 关键：每个batch后detach hidden，防止梯度爆炸 if hidden is not None: hidden = tuple(h.detach() for h in hidden)

hidden.detach()切断计算图，避免反向传播时追溯到上个batch。

6.5 陷阱5：多步预测的“自回归幻觉”（Autoregressive Illusion）——用假数据喂模型

现象：多步预测曲线平滑但严重偏离真实值，误差随步长指数增长。
原因：误以为训练好的单步模型可通过“预测→喂回→再预测”实现多步，但模型从未见过自己预测的值，分布偏移（Distribution Shift）导致失败。
解决：坚持直接多步训练（Direct Multi-step Forecasting）。若必须用单步模型，应采用多输出头（Multi-head Output）：LSTM最后一层接36个独立全连接层，每个输出一步，共享LSTM特征提取器。这比循环预测稳定得多。

7. 从“能跑通”到“可交付”：一个生产就绪的预测服务封装技巧

7.1 模型序列化：保存scaler、history、模型权重的原子操作

训练完的模型不能只存.pt文件，必须打包所有依赖组件。我一般用joblib保存scaler和history，torch.save存模型，再用zip打包：

import joblib import torch def save_model_artifacts(model, scaler, history, diff_order, path_prefix="lstm_model"): """ 保存完整模型资产 """ # 保存PyTorch模型 torch.save({ 'model_state_dict': model.state_dict(), 'input_size': 1, 'hidden_size': 64, 'num_layers': 2, 'output_size': 36, }, f"{path_prefix}_weights.pt") # 保存预处理组件 joblib.dump(scaler, f"{path_prefix}_scaler.pkl") joblib.dump({'history': history, 'diff_order': diff_order}, f"{path_prefix}_preproc.pkl") # 打包成zip（便于交付） import zipfile with zipfile.ZipFile(f"{path_prefix}.zip", 'w') as zf: zf.write(f"{path_prefix}_weights.pt") zf.write(f"{path_prefix}_scaler.pkl") zf.write(f"{path_prefix}_preproc.pkl") print(f"模型资产已打包至 {path_prefix}.zip") # 调用保存 save_model_artifacts(model, scaler, history, diff_order=1)

交付时只给.zip，对方解压后即可加载，避免版本冲突。

7.2 预测服务API：Flask轻量封装，支持单点/批量预测

from flask import Flask, request, jsonify import numpy as np import torch import joblib app = Flask(__name__) # 全局加载模型（启动时执行） model = LSTMModel(input_size=1, hidden_size=64, num_layers=2, output_size=36) checkpoint = torch.load("lstm_model_weights.pt") model.load_state_dict(checkpoint['model_state_dict']) model.eval() # 设为评估模式 scaler = joblib.load("lstm_model_scaler.pkl") preproc = joblib.load("lstm_model_preproc.pkl") history = preproc['history'] diff_order = preproc['diff_order'] @app.route('/predict', methods=['POST']) def predict(): try: data = request.json # data: {"history_sequence": [92.1, 92.3, ...], "forecast_steps": 36} hist_seq = np.array(data['history_sequence']) forecast_steps = data.get('forecast_steps', 36) # 预处理：差分 → 归一化 → 构造窗口 diff_hist, _ = apply_differencing(hist_seq, order=diff_order) scaled_diff = scaler.transform(diff_hist.reshape(-1, 1)).flatten() # 构造单样本输入：(1, lookback, 1) lookback = 50 if len(scaled_diff) < lookback: raise ValueError(f"History length {len(scaled_diff)} < lookback {lookback}") X_input = scaled_diff[-lookback:].reshape(1, lookback, 1) # 预测 with torch.no_grad(): X_tensor = torch.tensor(X_input, dtype=torch.float32) pred_scaled, _ = model(X_tensor) # <p> <a href="https://download.csdn.net/download/weixin_43471818/11166952" style="color:#ec7500;font-size:14px;"> 本文还有配套的精品资源，点击获取 </a> <img alt="menu-r.4af5f7ec.gif" src="https://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif" style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;"> </p>