使用PatchTST机器学习算法预测未来24小时的价格走势（基础篇）
「用PatchTST在MT5里推演未来24小时行情」
PatchTST 是一类基于 Transformer 的时间序列预测模型，核心思路是把历史价格切成不重叠的 patch（片段），再喂给注意力机制去学局部与全局依赖，输出未来若干步的走势估计。在 MT5 环境里接这套模型，意义不在于「算命」，而是把 24 小时窗口内的方向概率从肉眼复盘里剥离出来。 实际跑起来，作者于 2025 年 1 月 14 日在 MetaTrader 5 社区发布该思路，原文显示有 1093 次查看、15 条讨论，说明价格行为交易者对新范式接受度不低。但外汇与贵金属杠杆高、滑点乱，模型输出只能当「倾向」而非指令。 要验证它，你可以自己拉 EURUSD 的 H1 收盘序列，用 Python 训一个轻量 PatchTST，再把预测值通过 CSV 丢进 MT5 做视觉叠加。重点看预测轨道与实际 K 线的偏离带宽，而不是单点命中率。
◍ 从Transformer到MT5：PatchTST为何值得交易者上手
做价格行为的人大多盯K线形态，但时间序列预测里的Transformer革命已经渗透进外汇和贵金属量化圈。我在Huggingface上第一次碰到PatchTST时，它最戳人的两点很实在：训练快、训完接MQL5方便。这对想用AIGC工具辅助盯盘的交易者来说，意味着不用从零搭框架就能在MT5里验证模型。 必须坦白，我接触神经网络时间不长，但把PatchTST跑通并改写成MQL5这一套走下来，对理解复杂网络的训练、排错和调用帮助极大。外汇和贵金属杠杆高、跳空频繁，直接信模型信号风险极大，但用历史数据回测它的分块预测逻辑，能帮你建立对AI时序工具的直觉。 这过程像把刚学步的人扔进职业球队去踢世界杯决赛——夸张，但逼着你快速长经验。PatchTST把序列切成patch再编码，恰好适配MT5里按根数取样本的习惯，接下来我们会拆它的具体实现。
PatchTST 凭什么省掉繁琐预处理
PatchTST 能直接吃进 OHLC 甚至成交量，不像传统做法要先做一堆特征工程。它靠的是把序列切成「补丁（patches）」——类比 ViT 切图像，但落在时间序列上：比如补丁长度设 16，每个补丁就裹住 16 个连续价格值，先盯局部再管全局。 作者塞进来的 RevIn（可逆实例归一化）专门对付分布偏移。做 EA 的都懂那种痛：训好的模型跑几个月就失灵，得重优化。RevIn 在送入模型前用均值方差归一，预测时再反算回去，相当于内置了一层对抗市场态迁移的机制，还带可学习的 affine_bias 去修偏度峰度。 以 EURUSD H1 为例，拉 80000 根大约是 13 年数据量。整条链路是：MT5 取数 → RevIn → 序列分解出趋势与残差 → 两路都走补丁化、位置编码、Transformer 编码器 → 各自出预测后相加。多头注意力让模型跨补丁抓依赖，前馈层啃非线性，位置编码保住时间先后。 下面三行是归一与反归一的核心公式，外加带偏置的仿射变换；流水线那行可直接对照上面说的分解—相加结构在 MT5 里搭验证。
x = (x - mean) / std x = x * std + mean x = x * affine_weight + affine_bias 输入数据 -> RevIn -> 序列分解 -> 趋势分量 -> PatchTST 主干 -> TSTiEncoder -> Flatten_Head -> 趋势预测器 -> 残差分量 -> 趋势和残差相加 -> 最终预测
「Unfold 卡死 ONNX 转换的坑」
PatchTST 官方实现提供监督与无监督两套代码，接 MT5 必须走 ONNX，但原版主干里第一层就调用了 Unfold 做分块展开，ONNX 并不支持该算子，导出时会直接报转换错误，模型进不了 MQL5 环境。 为了绕过这个限制，我把 Unfold 那段改写成显式 for 循环手动切片拼 patch。代价是训练效率下降：在 n_layers=3、d_model=128 的默认配置下，多 epoch 跑大样本时循环开销会累积，但这是让模型能被 ONNX 吃进来的必要妥协。 改写后的全部逻辑我收进了 patchTST.py 单文件，后续训练直接用它。下面这段主干初始化展示了分块数怎么算：context_window 减 patch_len 除以 stride 加 1，若尾部补 pad 则再 +1，这一步搞清楚才能自己调 patch_len 和 stride 验证推理形状。
class PatchTST_backbone(nn.Module): def __init__(self, c_in:class="type">int, context_window:class="type">int, target_window:class="type">int, patch_len:class="type">int, stride:class="type">int, max_seq_len:Optional[class="type">int]=class="num">1024, n_layers:class="type">int=class="num">3, d_model=class="num">128, n_heads=class="num">16, d_k:Optional[class="type">int]=None, d_v:Optional[class="type">int]=None, d_ff:class="type">int=class="num">256, norm:str=&class="macro">#x27;BatchNorm&class="macro">#x27;, attn_dropout:class="type">class="kw">float=class="num">0., dropout:class="type">class="kw">float=class="num">0., act:str="gelu", key_padding_mask:class="type">bool=&class="macro">#x27;auto&class="macro">#x27;, padding_var:Optional[class="type">int]=None, attn_mask:Optional[Tensor]=None, res_attention:class="type">bool=True, pre_norm:class="type">bool=False, store_attn:class="type">bool=False, pe:str=&class="macro">#x27;zeros&class="macro">#x27;, learn_pe:class="type">bool=True, fc_dropout:class="type">class="kw">float=class="num">0., head_dropout = class="num">0, padding_patch = None, pretrain_head:class="type">bool=False, head_type = &class="macro">#x27;flatten&class="macro">#x27;, individual = False, revin = True, affine = True, subtract_last = False, verbose:class="type">bool=False, **kwargs): super().__init__() # RevIn self.revin = revin if self.revin: self.revin_layer = RevIN(c_in, affine=affine, subtract_last=subtract_last) # Patching self.patch_len = patch_len self.stride = stride self.padding_patch = padding_patch patch_num = class="type">int((context_window - patch_len)/stride + class="num">1) if padding_patch == &class="macro">#x27;end&class="macro">#x27;: # can be modified to general case self.padding_patch_layer = nn.ReplicationPad1d((class="num">0, stride)) patch_num += class="num">1 # Backbone self.backbone = TSTiEncoder(c_in, patch_num=patch_num, patch_len=patch_len, max_seq_len=max_seq_len, n_layers=n_layers, d_model=d_model, n_heads=n_heads, d_k=d_k, d_v=d_v, d_ff=d_ff, attn_dropout=attn_dropout, dropout=dropout, act=act, key_padding_mask=key_padding_mask, padding_var=padding_var, attn_mask=attn_mask, res_attention=res_attention, pre_norm=pre_norm, store_attn=store_attn, pe=pe, learn_pe=learn_pe, verbose=verbose, **kwargs) # Head
◍ PatchTST 头部与前向的数据流拆解
这段类定义负责把多变量时序张量送进 Transformer 主干并吐出预测窗口。初始化时先算 head_nf = d_model * patch_num，也就是每个变量展平后的特征总长；n_vars 直接取通道数 c_in，pretrain_head 与 head_type 决定走预训练头还是 flatten 头。 forward 的第一步可选 RevIN 归一：若开启 revin，就先把 [bs, nvars, seq_len] 转成 [bs, seq_len, nvars] 做 norm 再转回来，避免不同品种量纲互相污染。 打补丁环节用 z.unfold(dimension=-1, size=patch_len, step=stride) 在序列维滑窗，得到 [bs, nvars, patch_num, patch_len]，再 permute 成 [bs, nvars, patch_len, patch_num] 喂给 backbone。主干输出 [bs, nvars, d_model, patch_num]，经 head 压成 [bs, nvars, target_window]。 一个坑点：unfold 算子不支持 ONNX 导出，报错 'input size not accessible'。想在 MT5 外做量化推理的服务端若用 ONNX，得手动展开该算子，否则模型导不出。外汇与贵金属行情波动剧烈，此类模型仅作概率参考，实盘高风险。
self.head_nf = d_model * patch_num
self.n_vars = c_in
self.pretrain_head = pretrain_head
self.head_type = head_type
self.individual = individual
if self.pretrain_head:
self.head = self.create_pretrain_head(self.head_nf, c_in, fc_dropout) # custom head passed as a partial func with all its kwargs
elif head_type == &class="macro">#x27;flatten&class="macro">#x27;:
self.head = Flatten_Head(self.individual, self.n_vars, self.head_nf, target_window, head_dropout=head_dropout)
def forward(self, z): # z: [bs x nvars x seq_len]
# norm
if self.revin:
z = z.permute(class="num">0,class="num">2,class="num">1)
z = self.revin_layer(z, &class="macro">#x27;norm&class="macro">#x27;)
z = z.permute(class="num">0,class="num">2,class="num">1)
# do patching
if self.padding_patch == &class="macro">#x27;end&class="macro">#x27;:
z = self.padding_patch_layer(z)
z = z.unfold(dimension=-class="num">1, size=self.patch_len, step=self.stride) # z: [bs x nvars x patch_num x patch_len]
z = z.permute(class="num">0,class="num">1,class="num">3,class="num">2) # z: [bs x nvars x patch_len x patch_num]
# model
z = self.backbone(z) # z: [bs x nvars x d_model x patch_num]
z = self.head(z) # z: [bs x nvars x target_window]
# denorm
if self.revin:
z = z.permute(class="num">0,class="num">2,class="num">1)
z = self.revin_layer(z, &class="macro">#x27;denorm&class="macro">#x27;)
z = z.permute(class="num">0,class="num">2,class="num">1)
class="kw">return z把序列切成可训练的滑动窗口
在把行情序列送进 Transformer 类模型前，得先按 patch 方式切块。上面这段逻辑干的事很直接：先拆出 batch_size、n_vars、seq_len 三个维度，再用滑动窗口把长度为 seq_len 的序列切成若干段。 循环里 i 从 0 起步，步长取 self.stride，每次截 self.patch_len 长度。当 seq_len 是 100、patch_len 是 16、stride 是 8 时，会切出 11 个 patch（(100-16)/8 + 1），最后一个窗口右端正好落在 99。这个颗粒度直接决定模型能看到多少局部结构，调 stride 就是在延迟与细节之间做权衡。 在 MT5 里接 AIGC 分析模块时，这种切片参数要和外围特征采样频率对齐，否则会出现重叠过多或漏采样。贵金属与外汇波动聚集明显，patch 太长容易抹掉突变，太短则噪声放大，建议先用历史 tick 复算一遍再固定。
batch_size, n_vars, seq_len = z.size() patches = [] for i in range(class="num">0, seq_len - self.patch_len + class="num">1, self.stride): patches.append(z[:, :, i:i+self.patch_len])