时序预测的深度模型架构演进:从 Transformer 到 PatchTST 的工业级工程实践

时序预测(Time Series Forecasting)是工业场景中最为普遍的需求之一——从电力负载预测、网络流量规划,到供应链库存管理、金融波动率建模,几乎所有运营决策都建立在"未来值预测"之上。传统方法如 ARIMA、指数平滑(Exponential Smoothing)和 Prophet 在处理单变量、低噪声、稳定周期性的场景时表现优异,但面对多变量非线性关联、长序列预测、分布漂移等问题时往往捉襟见肘。

2022 年以来,深度学习模型在时序预测领域取得了质的飞跃:从最初的 LSTM/GRU 到 Transformer 架构的时序适配,再到 PatchTST 的补丁化革新,以及 iTransformer 的变量翻转范式,每一次架构创新都在重塑我们对"时序建模"的理解。本文将深入剖析这一演进路径背后的核心设计决策,并结合生产级代码示例和性能对比数据,为你构建完整的工程实践框架。


一、为什么传统时序方法面临天花板

1.1 ARIMA 家族的本质局限

ARIMA(Autoregressive Integrated Moving Average)的核心假设是线性可加性——当前值是过去残差和过去观测值的线性组合。这个假设在以下场景下迅速崩塌:

  • 多变量耦合:当多个时序变量存在非线性相互影响时(如气象预测中温度、湿度、气压的耦合),ARIMA 无法有效建模变量间的条件依赖。
  • 长记忆依赖:ARIMA 依赖于差分阶数 d 来消除非平稳性,但这会同时削弱长程信号,导致 100+ 步预测时信息衰减严重。
  • 突变与断点:金融市场中的黑天鹅事件、电力网络中的短路故障等突变模式,ARIMA 无法自适应。

1.2 Prophet 的工程妥协

Facebook 开源的 Prophet 本质上是一个可加回归模型,它将时序分解为趋势(trend)、季节性(seasonality)和节假日效应(holidays)三部分。Prophet 的优势在于零配置即可产生合理的预测,这使其成为业务快速验证的首选方案。

但 Prophet 的局限性也很明显:趋势建模使用分段逻辑斯蒂曲线或线性曲线,无法捕捉加速/减速变化;季节性使用傅里叶级数展开,本质上还是全局周期性假设;最重要的是,Prophet 无法利用外部协变量(exogenous variables)进行联合预测。

# Prophet 典型使用:单变量预测 + 节假日效应
from prophet import Prophet
import pandas as pd

df = pd.read_csv('energy_load.csv')  # 列: ds (日期), y (负载)
df['ds'] = pd.to_datetime(df['ds'])

model = Prophet(
    changepoint_prior_scale=0.05,  # 趋势变化灵敏度
    seasonality_mode='multiplicative',
    yearly_seasonality=True,
    weekly_seasonality=True
)
model.add_country_holidays(country_name='CN')
model.fit(df)

future = model.make_future_dataframe(periods=48, freq='30min')
forecast = model.predict(future)
# 输出: yhat (预测值), yhat_lower, yhat_upper (置信区间)

上述代码的简洁性掩盖了一个事实:Prophet 的内部是一种特殊的贝叶斯 GAM(广义可加模型),当数据中存在非线性交互或多变量依赖时,它无法像端到端神经网络那样从数据中自动学习表征。


二、深度学习时序模型的三次范式转移

2.1 第一次转移:MLP 也能做时序预测(2021)

2021 年初,N-BEATS(Neural Basis Expansion Analysis)的发表打破了"时序预测必须用 RNN/Transformer"的思维定式。N-BEATS 的架构极其简单:完全基于多层全连接网络(MLP),通过分块输入(lookback window)和双残差堆叠(double residual stacking)机制来实现时序分解。

N-BEARS 的核心洞察是:时序数据中的趋势和季节性可以通过 MLP 的基展开(basis expansion)自动学习,不需要显式的分解步骤。

import torch
import torch.nn as nn

class NBEATSBlock(nn.Module):
    """N-BEATS 基础块:全连接网络 + 趋势/季节性基展开"""
    def __init__(self, input_size, hidden_size, forecast_size, backcast_size, num_layers=4):
        super().__init__()
        self.fc_stack = nn.ModuleList()
        in_features = input_size
        for _ in range(num_layers):
            self.fc_stack.append(nn.Sequential(
                nn.Linear(in_features, hidden_size),
                nn.ReLU()
            ))
            in_features = hidden_size

        # 基展开层:将隐藏表示映射回 backcast 和 forecast
        self.backcast_layer = nn.Linear(hidden_size, backcast_size)
        self.forecast_layer = nn.Linear(hidden_size, forecast_size)

    def forward(self, x):
        # x: [batch, input_size]
        for layer in self.fc_stack:
            x = layer(x)
        backcast = self.backcast_layer(x)   # 重建输入(残差连接用)
        forecast = self.forecast_layer(x)   # 预测输出
        return backcast, forecast

# N-BEATS 的双残差堆叠允许逐步剥离信号成分
# Block1: 输入原始序列 → 输出 trend 预测 + 去除趋势后的残差
# Block2: 输入残差 → 输出 seasonality 预测 + 进一步残差
# ...最终预测 = 所有 block 的 forecast 之和

N-BEATS 在 M4 竞赛数据集上的表现首次证明了纯 MLP 可以匹敌甚至超越传统统计方法。但它的致命缺陷是:无法利用多变量协变量,因为 MLP 的输入是 flat vector,不具备变量间的结构建模能力。

2.2 第二次转移:Transformer 的时序适配(2022)

2022 年,Informer(AAAI 2022 Best Paper)首次将 Transformer 架构系统性地适配到时序预测场景,解决了原生 Transformer 在长序列上的 O(n²) 计算瓶颈。

Informer 的三个核心创新:

ProbSparse Self-Attention:传统注意力对所有时间步计算两两关联,复杂度 O(L²)。Informer 通过 KL 散度衡量每个 Query 的稀疏性,只选择 Top-u 个主导 Query 计算全注意力,其余 Query 使用近似,将复杂度降至 O(L log L)。

Self-attention Distilling:通过一维卷积+池化层逐步缩短序列长度,每一层的输入长度减半,形成自注意力蒸馏金字塔,进一步降低计算开销。

Generative Style Decoder:一次性生成整个预测序列,而非逐步自回归解码。Decoder 输入包含从历史序列中采样的 "start token" 和目标位置的占位符。

import torch
import torch.nn as nn
import math

class ProbSparseAttention(nn.Module):
    """Informer 的 ProbSparse 自注意力:只关注少数活跃 Query"""
    def __init__(self, d_model, n_heads, factor=5):
        super().__init__()
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads
        self.factor = factor  # 采样因子,控制选择的 Query 数量

        self.W_Q = nn.Linear(d_model, d_model)
        self.W_K = nn.Linear(d_model, d_model)
        self.W_V = nn.Linear(d_model, d_model)
        self.fc = nn.Linear(d_model, d_model)

    def _prob_QK(self, Q, K, sample_k, n_top):
        """通过 KL 散度选择 Top-n 个活跃 Query"""
        B, H, L_K, D = K.shape
        _, _, L_Q, _ = Q.shape

        # 采样 Key:每个 head 随机采样 L_K * sample_k 个 Key
        K_expand = K.unsqueeze(-3).expand(B, H, L_Q, L_K, D)
        index = torch.randint(0, L_K, (L_Q, sample_k))
        K_sample = K_expand[:, :, torch.arange(L_Q).unsqueeze(1), index, :]

        # 计算采样后的 QK 注意力分数
        Q_K_sample = torch.matmul(Q.unsqueeze(-2), K_sample.transpose(-2, -1)).squeeze(-2)

        # 计算稀疏性度量:M(Q_i) = max_k(a_ik) - mean_k(a_ik)
        M = Q_K_sample.max(-1)[0] - Q_K_sample.mean(-1)

        # 选择 Top-n 个 Query(稀疏性分数最高的)
        M_top = M.topk(n_top, sorted=False)[1]
        Q_reduce = Q[torch.arange(B)[:, None, None], torch.arange(H)[None, :, None], M_top]

        # 用筛选后的 Query 计算完整注意力
        Q_K = torch.matmul(Q_reduce, K.transpose(-2, -1))
        return Q_K, M_top

    def forward(self, Q, K, V, attn_mask=None):
        B, L_Q, _ = Q.shape
        _, L_K, _ = K.shape

        Q = self.W_Q(Q).view(B, L_Q, self.n_heads, self.d_k).transpose(1, 2)
        K = self.W_K(K).view(B, L_K, self.n_heads, self.d_k).transpose(1, 2)
        V = self.W_V(V).view(B, L_K, self.n_heads, self.d_k).transpose(1, 2)

        # 采样因子计算要保留的 Query 数量
        sample_k = int(self.factor * math.log(L_K))
        n_top = int(self.factor * math.log(L_Q))

        Q_K, M_top = self._prob_QK(Q, K, sample_k, n_top)
        attn = torch.softmax(Q_K / math.sqrt(self.d_k), dim=-1)

        # 使用完整 V 计算注意力输出
        V_reduce = V[torch.arange(B)[:, None, None], torch.arange(self.n_heads)[None, :, None], M_top]
        attn_out = torch.matmul(attn, V_reduce)

        # 恢复 shape
        attn_out = attn_out.transpose(1, 2).contiguous().view(B, n_top, self.d_model)
        return attn_out, attn

Informer 的出现标志着时序预测正式进入"大模型时代"。但 Transformer 架构在处理时序数据时仍存在两个本质问题:位置编码不能很好地表达时序相对距离关系,以及全局注意力在超长序列(>1000 步)上的计算负担仍重。

2.3 第三次转移:PatchTST 的补丁化革新(2023)

2023 年,PatchTST(ICLR 2024)提出了一个反直觉但极其有效的思想:将时序数据切成小补丁(patch)作为 Transformer 的输入 token,而非单步时间点。

PatchTST 的核心洞见是:时序数据在局部时间段内具有高度冗余性(adjacent time steps are highly correlated),因此单步 token 既浪费计算资源,也增加了 Transformer 学习局部模式的难度。通过将每 N 个连续步合并为一个 patch,可以:

  1. 缩短序列长度:长度为 L 的输入被压缩为 L/p 个 patch(p 为 patch 长度),自注意力复杂度从 O(L²) 降为 O((L/p)²)
  2. 增强局部语义:每个 patch 携带了局部趋势、波动模式,比单步值更具"语义"
  3. 实现通道独立(Channel Independence):每个变量独立通过 Transformer,避免多变量混合建模时的维度灾难
class PatchEmbedding(nn.Module):
    """PatchTST 的补丁嵌入层"""
    def __init__(self, patch_len, stride, d_model, dropout=0.1):
        super().__init__()
        self.patch_len = patch_len
        self.stride = stride
        # 将 patch_len 步映射为 d_model 维向量
        self.projection = nn.Linear(patch_len, d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):
        """
        x: [batch, n_vars, seq_len]
        输出: [batch, n_vars, n_patches, d_model]
        """
        B, C, L = x.shape

        # 将序列切分为重叠 patch
        # unfold 输出: [B, C, n_patches, patch_len]
        x = x.unfold(dimension=-1, size=self.patch_len, step=self.stride)

        # 投影到 d_model 维度
        x = self.projection(x)  # [B, C, n_patches, d_model]
        x = self.dropout(x)
        return x


class PatchTST_backbone(nn.Module):
    """PatchTST 主干网络:通道独立的 Transformer + 补丁化输入"""
    def __init__(self, patch_len=16, stride=8, d_model=128, n_heads=8, 
                 n_layers=3, d_ff=256, dropout=0.1):
        super().__init__()
        self.patch_len = patch_len
        self.n_layers = n_layers
        self.channel_mix = False  # 通道独立模式

        self.patch_embed = PatchEmbedding(patch_len, stride, d_model, dropout)

        # Transformer Encoder 层
        self.encoder_layers = nn.ModuleList([
            nn.TransformerEncoderLayer(
                d_model=d_model,
                nhead=n_heads,
                dim_feedforward=d_ff,
                dropout=dropout,
                batch_first=True,
                activation='gelu'
            ) for _ in range(n_layers)
        ])

        # 输出投影层:从 patch 维度回推步数
        self.flatten = nn.Flatten(start_dim=-2)

    def forward(self, x):
        """
        x: [batch, seq_len, n_vars]
        """
        B, L, C = x.shape
        n_patches = (L - self.patch_len) // self.stride + 1

        # 补丁嵌入
        # 将 [B, L, C] 重塑为 [B*C, L, 1] 处理每个通道
        x = x.permute(0, 2, 1)  # [B, C, L]
        x_patch = self.patch_embed(x)  # [B, C, n_patches, d_model]

        # 对每个通道独立应用 Transformer
        preds = []
        for c in range(C):
            x_c = x_patch[:, c]  # [B, n_patches, d_model]
            for layer in self.encoder_layers:
                x_c = layer(x_c)
            preds.append(x_c)

        output = torch.stack(preds, dim=1)  # [B, C, n_patches, d_model]
        return output

PatchTST 在多个标准数据集上(ETTh1/ETTh2/ETTm1/Weather/Electricity/Traffic)击败了同期所有 Transformer-based 方法,同时计算效率提升 3-5 倍。更重要的是,它证明了"少即是多"——通过合理的输入表征设计(补丁化),简单的 Transformer Encoder 就能产生卓越的时序预测性能。


三、架构对比与生产选型指南

在实际生产环境中,选择哪种架构需要综合考虑以下维度:

架构 适用场景 序列长度限制 多变量支持 训练成本 推理延迟
Prophet 快速基线、季节性强、节假日效应明显 无限制 ❌ 极低 毫秒级
ARIMA 小规模单变量、平稳/可差分序列 无限制 ❌ 低 毫秒级
N-BEATS 单变量通用预测、中等序列长度 ~500步 ❌ 中等 10毫秒级
Informer 长序列预测、多变量依赖建模 ~2000步 ✅ 高 100毫秒级
PatchTST 多变量预测、局部模式丰富的场景 ~5000步(补丁压缩后) ✅(通道独立) 中等 50毫秒级
iTransformer 变量间关联强烈的多变量预测 ~5000步 ✅(变量独立+依赖建模) 中等 50毫秒级

3.1 关键工程教训

教训一:长序列不等于好预测

很多工程师有一个直觉性的误区:输入的历史窗口越长,模型获得的信息越多,预测越准确。实际上,NIPS 2022 的一项研究(Zeng et al.)表明,一个简单的线性映射(Linear Projection)在足够长的历史窗口上就能匹配甚至超越复杂 Transformer。

这意味着,对于很多工业时序预测场景,问题可能不在于模型能力不足,而在于数据质量的缺失——采集频率不稳定、传感器漂移、异常值污染等因素才是限制预测精度的真正瓶颈。

教训二:通道独立 vs 通道混合

PatchTST 默认采用 Channel Independence(每个变量独立建模),而大多数 Transformer 模型采用 Channel Mixing(所有变量共同通过注意力混合建模)。实验证据表明:

  • 当变量数量较少(<20)且变量间关联强烈时(如电网中多节点负载),Channel Mixing 更有效
  • 当变量数量较大(>50)或变量间关联较弱时(如多商品销量预测),Channel Independence + 残差连接效果更好
  • 工程折中:可以先使用 Channel Independence 训练,再在推理阶段通过经济/物理约束(如产能上限、流量守恒)进行后校正

教训三:预测的对称性陷阱

深度学习时序模型在训练时通常使用均方误差(MSE)损失,这使得模型学到的是条件均值估计。但在很多业务场景中:

  • 电力调度预测高估会导致备用容量浪费,低估会导致断电风险——需要不对称损失函数
  • 零售销量预测高估会导致过期浪费,低估会损失销售额——需要分位数预测而非点预测

解决方案是使用分位数损失(Quantile Loss)或 CRPS(Continuous Ranked Probability Score)替代 MSE,输出预测分布而非单点值。

def quantile_loss(y_pred, y_true, quantiles=[0.1, 0.25, 0.5, 0.75, 0.9]):
    """
    分位数损失:用于生成概率预测而非单点预测
    y_pred: [B, n_quantiles, forecast_len]
    y_true: [B, forecast_len]
    """
    losses = []
    for i, q in enumerate(quantiles):
        errors = y_true - y_pred[:, i, :]
        loss = torch.max(q * errors, (q - 1) * errors)
        losses.append(loss.mean())
    return torch.stack(losses).mean()

四、生产部署的工程实践

4.1 数据管道设计

时序预测的工程质量 80% 取决于数据预处理。一个健壮的生产级数据管道应包含以下组件:

from abc import ABC, abstractmethod
import numpy as np
import pandas as pd
from typing import List, Optional

class TimeSeriesPipeline:
    """生产级时序数据管道"""

    def __init__(self, 
                 target_cols: List[str],
                 feature_cols: Optional[List[str]] = null,
                 scaler_type: str = 'robust',  # 'standard', 'minmax', 'robust'
                 freq: str = '1H',
                 fill_missing: str = 'interpolate',
                 outlier_clip: float = 3.0):    # 标准差倍数裁剪
        self.target_cols = target_cols
        self.feature_cols = feature_cols or []
        self.scaler_type = scaler_type
        self.freq = freq
        self.fill_method = fill_missing
        self.outlier_clip = outlier_clip

    def fit_transform(self, df: pd.DataFrame) -> np.ndarray:
        """拟合并转换训练数据"""
        # 1. 频率统一:重采样到目标频率
        df = df.resample(self.freq).mean()

        # 2. 缺失值处理
        if self.fill_method == 'interpolate':
            df = df.interpolate(method='time')
        elif self.fill_method == 'forward':
            df = df.fillna(method='ffill')

        # 3. 异常值裁剪(基于历史统计)
        if self.outlier_clip:
            rolling_mean = df[self.target_cols].rolling(window=24*7, min_periods=1).mean()
            rolling_std = df[self.target_cols].rolling(window=24*7, min_periods=1).std()
            df[self.target_cols] = df[self.target_cols].clip(
                lower=rolling_mean - self.outlier_clip * rolling_std,
                upper=rolling_mean + self.outlier_clip * rolling_std
            )

        # 4. 特征工程:时间编码(正弦/余弦编码处理周期性)
        df['hour_sin'] = np.sin(2 * np.pi * df.index.hour / 24)
        df['hour_cos'] = np.cos(2 * np.pi * df.index.hour / 24)
        df['day_sin'] = np.sin(2 * np.pi * df.index.dayofweek / 7)
        df['day_cos'] = np.cos(2 * np.pi * df.index.dayofweek / 7)
        df['month_sin'] = np.sin(2 * np.pi * df.index.month / 12)
        df['month_cos'] = np.cos(2 * np.pi * df.index.month / 12)

        # 5. 归一化
        from sklearn.preprocessing import RobustScaler
        self.scaler = RobustScaler()
        scaled_data = self.scaler.fit_transform(df[self.target_cols + self.feature_cols])

        return scaled_data

4.2 在线学习与概念漂移检测

生产环境中的时序数据几乎从不满足独立同分布假设,概念漂移(Concept Drift)是常态而非例外。两个关键工程策略:

滑动窗口再训练:维护固定大小的滑动训练窗口(如最近 90 天),定期微调模型参数。这种策略平衡了模型对最新模式的适应性和计算开销。

ADWIN 漂移检测:ADaptive WINdowing(ADWIN)算法维护两个相邻子窗口的统计量差异,当差异超过阈值时触发模型重训练。

class ADWIN:
    """自适应窗口变化检测器:用于触发模型重训练"""
    def __init__(self, delta=0.002):
        self.delta = delta
        self.window = []
        self.width = 0
        self.total = 0.0
        self.variance = 0.0

    def add_element(self, value):
        self.window.append(value)
        self.width += 1
        self.total += value
        if self.width > 1:
            self.variance += (value - self.total / self.width) ** 2

    def detected_change(self):
        """检测是否发生概念漂移"""
        for i in range(1, self.width // 2):
            w0, w1 = i, self.width - i
            mu0 = sum(self.window[:i]) / w0
            mu1 = sum(self.window[i:]) / w1
            m = 1 / (1 / w0 + 1 / w1)
            epsilon = math.sqrt(2 / m * math.log(2 / self.delta))
            if abs(mu0 - mu1) > epsilon > 0:
                # 检测到漂移:截断窗口保留最近部分
                self.window = self.window[i:]
                self.width -= i
                self.total -= mu0 * w0
                return True
        return False

五、2024-2026 年的最新进展与展望

时序预测领域在 2024 年后进入了"Foundation Model 时代",几个值得关注的趋势:

TimesFM(Google, 2024):首个时序预测大语言模型,通过预训练大规模时序数据实现零样本预测。其核心思路是将时序 patch 视为 token 序列,使用 Decoder-only Transformer 进行 next-patch 预测。这标志着时序预测与 LLM 技术的深度融合。

Lag-Llama(2024):基于 Lag 特征增强的通用时序预测模型,通过将历史值编码为 Lag 特征来对齐语言模型的 token 空间,避免了复杂的数值 tokenizer 设计。

Mamba-based 时序模型(2025-2026):随着 Mamba 状态空间模型在效率上对 Transformer 取得优势,基于 Mamba 的时序预测探索逐渐增多。Mamba 的选择性扫描机制天然适合时序数据的顺序建模,同时保持线性复杂度。

从工程实践角度看,当前最佳实践是"金字塔策略": 1. 底层快速决策:使用 LightGBM/Prophet 处理高频实时监控(秒级/分钟级响应) 2. 中层业务优化:使用 PatchTST/iTransformer 处理小时/天级业务预测(库存调度、负载规划) 3. 上层战略规划:使用时序基础模型处理长期预测(月度/季度业务规划)


六、总结

时序预测的深度学习架构演进揭示了一个核心规律:架构创新的关键不在于参数量的增加,而在于输入表征的重新设计。从单步 token 到补丁 token,从全局注意力到局部通道独立,每一次突破都来自于对时序数据本身特性的更深入理解。

对于工程师而言,在选型时不应盲目追新,而应根据自身数据特征(序列长度、变量数量、采样频率、季节性强度)和业务需求(预测频率、精度要求、推理延迟预算)做出理性决策。最强大的模型往往不是参数最多的那一个,而是与数据特性最匹配的那一个。

点赞(0) 打赏

评论列表 共有 0 条评论

暂无评论
立即
投稿

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部