用于多变量时间序列回归的 PatchTST 设计
代码 《交易机器学习》
总结
该模型适配器将 PatchTST（一种用于长周期时间序列预测的 Transformer 架构）应用于标量回归。它接收按时间和特征排列的序列，调整输入以适配骨干网络，并将各特征的输出汇总为一个预测值。其设计采用通道独立编码：每个特征通道通过共享的 Transformer 权重处理，编码器内部不进行跨通道混合。
该实现还使用可逆实例归一化，在编码前对每个样本和通道进行归一化，并在之后恢复其尺度。重叠分块保留局部序列结构，展平式预测头将编码后的序列映射为输出。最后由线性层将各通道输出合成为一个标量。这些架构选择说明了该适配器如何将通用预测模型映射到多变量回归接口；文档没有提供交易实验、数据集、基准测试或表现结果。其论述针对模型结构，因此此处尚未确立其对金融数据的预测价值。
核心观点
- PatchTST 将每个输入通道划分为多个分块，并使用共享权重编码各通道。
- 可逆实例归一化会移除并恢复每个样本、每个通道的统计特征。
- 该设计使用重叠分块和展平头，而非全局平均池化。
- 最后的线性投影将通道级输出合成为标量回归预测值。
- 该文档介绍了模型架构，但没有提供交易表现证据。
标签
全文
# patchtst.py
```py
"""PatchTST: patching + channel-independent Transformer for time series.
From Nie, Nguyen, Sinthong, Kalagnanam (2023), *A Time Series is Worth 64
Words: Long-term Forecasting with Transformers*, ICLR 2023.
Two structural properties distinguish the paper's PatchTST from naive
"tokenize-the-input-with-a-Transformer" baselines:
1. **Channel-independent patching.** Each feature channel is treated as its
own univariate sequence and passed through the same shared Transformer
weights. There is no cross-channel mixing inside the encoder. This file
delegates to ``PatchTST_backbone`` from the authors' repo to preserve
this exactly.
2. **RevIN (Reversible Instance Normalization).** Per-sample per-channel
statistics are removed before the backbone and re-added after, making
the model robust to distribution shift. The vendored backbone wires
this up when ``revin=True``.
Additionally, the paper uses **overlapping patches** (stride < patch_len)
and a **flatten + linear** prediction head rather than global mean pooling.
Adapter layer on top of the backbone:
- The backbone returns ``(batch, n_vars, target_window)``. For cross-sectional
scalar regression we set ``target_window=1`` and then project the per-channel
outputs to a single scalar via ``Linear(n_vars -> 1)``.
Reference implementation vendored from https://github.com/yuqinie98/PatchTST
(MIT License) into ``_reference/`` with import paths adjusted. RevIN is
vendored from https://github.com/ts-kim/RevIN (MIT License).
Interface preserved for the factory:
PatchTST(n_features: int, lookback: int, patch_size: int = 16, ...)
forward(x: (batch, seq_len, n_features)) -> (batch,)
"""
from __future__ import annotations
import torch
import torch.nn as nn
from case_studies.config.patchtst._reference import PatchTST_backbone
class PatchTST(nn.Module):
"""PatchTST channel-independent regressor.
Wraps the paper authors' ``PatchTST_backbone`` with a scalar regression
head. RevIN on by default; overlapping patches with stride=patch_size/2.
"""
def __init__(
self,
n_features: int,
lookback: int,
patch_size: int = 16,
stride: int | None = None,
d_model: int = 64,
n_heads: int = 4,
n_layers: int = 2,
d_ff: int | None = None,
dropout: float = 0.1,
attn_dropout: float = 0.0,
revin: bool = True,
affine: bool = True,
subtract_last: bool = False,
padding_patch: str = "end",
):
super().__init__()
if stride is None:
stride = max(1, patch_size // 2)
if d_ff is None:
d_ff = d_model * 4
self.backbone = PatchTST_backbone(
c_in=n_features,
context_window=lookback,
target_window=1,
patch_len=patch_size,
stride=stride,
n_layers=n_layers,
d_model=d_model,
n_heads=n_heads,
d_ff=d_ff,
attn_dropout=attn_dropout,
dropout=dropout,
revin=revin,
affine=affine,
subtract_last=subtract_last,
padding_patch=padding_patch,
head_type="flatten",
individual=False,
)
self.head = nn.Linear(n_features, 1)
def forward(self, x: torch.Tensor) -> torch.Tensor:
# x: (batch, seq_len, n_features) → backbone wants (batch, n_vars, seq_len)
z = x.permute(0, 2, 1)
# backbone out: (batch, n_vars, target_window=1)
z = self.backbone(z)
# collapse target_window and project across channels to scalar
z = z.squeeze(-1) # (batch, n_vars)
return self.head(z).squeeze(-1)
```在遵守原作品许可的前提下，附作者信息全文展示。 许可协议: MIT
此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。