时间序列标签可解释分解：用 NBEATS 把预测拆给你看（基础篇）
（1/3）· 多数预测模型是黑箱，这篇用 NBEATS 把趋势与季节拆成可读成分
直接拿收盘价喂模型的人，常忽略协变量带来的偏置与误读。把预测当神谕而不看分解结构，实盘里容易在拐点吃暗亏。外汇贵金属波动受多重因子牵引，高杠杆下误判代价直接翻倍。
◍ 用标签把价格序列拆开看
做价格行为研究时，直接拿原始收盘价跑模型往往解释力弱。把时间序列按事件打标签（例如突破、回撤、持有），再对标签做可解释性分解，能看清哪一段行情由哪类行为驱动。 在 MT5 里可以用历史数据接口给每根 bar 贴标签。下面这段把「当前收盘价高于前 N 根最大收盘价」记为突破事件，否则为 0，便于后续统计分布。 [CODE] int handle; // 指标句柄，留空示意 for(int i=0;i<rates_total;i++){ double max_prev = 0; for(int j=1;j<=N;j++){ if(Close[i-j] > max_prev) max_prev = Close[i-j]; } if(Close[i] > max_prev) label[i] = 1; // 突破 else label[i] = 0; // 非突破 } [/CODE] 上面代码逐行拆解：先定义句柄占位；外层循环遍历每根 bar；内层循环找前 N 根收盘价最大值；若当前收盘价越过该值则标 1，否则标 0。 实盘验证时建议取 EURUSD 的 H1 数据，N=20 回看。统计显示，标 1 的 bar 后续 5 根内继续创新高概率倾向高于标 0 组，但外汇高杠杆品种回撤快，仅作样本观察，不构成方向判定。
class="type">int handle; class=class="str">"cmt">// 指标句柄，留空示意 for(class="type">int i=class="num">0;i<rates_total;i++){ class="type">class="kw">double max_prev = class="num">0; for(class="type">int j=class="num">1;j<=N;j++){ if(Close[i-j] > max_prev) max_prev = Close[i-j]; } if(Close[i] > max_prev) label[i] = class="num">1; class=class="str">"cmt">// 突破 else label[i] = class="num">0; class=class="str">"cmt">// 非突破 }
「从单变量预测转向协变量可解释性」
上一篇只用了收盘价单变量喂给 NHITS 做预测，这一篇换 NBEATS 演示，核心不在换模型，而是把「模型怎么想」拆开给你看。引入多协变量的目的，是回答一个问题：除了收盘价，还有哪些量能帮模型把次日收盘判断得更稳。 NBEATS 和 NHITS 都属于结构可解释的时间序列模型，你完全可以拿文中提到的库套到别的模型上验证自己的假设。前 three 篇已经铺好了数据标签地基：用 EA 操作图打趋势标签、Python 做带标签数据集、以及标签数据的具体用法。 外汇与贵金属属高风险品种，本文给的是方法层面的解法，不是直接能跑实盘的圣杯。真要接 MT5 实盘，得自己补参数调优和稳定性验证，原文也明确提醒过这点。
NBEATS 的双残差结构与可解释基函数
NBEATS 不挑输入输出的序列长度，也不需要针对外汇或贵金属价格做特征工程或缩放预处理。它用多项式与傅立叶级数当基函数，能把趋势和季节成分拆开，这一点对看裸 K 的人理解周期很有用。 模型核心是双残差叠加：每个 block 有两条分支，一条往未来做预测，一条往过去做回溯。这种对偶拓扑让梯度更好训，也方便反推每一层到底学了什么。 可解释性靠的是给展开系数加函数形式约束。比如想让某层只管季节性，就把输出强制成周期函数，层就会偏向抓往复波动而不是乱拟合噪声。 协变量分六类：static 类不随时间变，time_varying_known 类未来已知（如假期、日期），time_varying_unknown 类未来未知（如趋势方向）。接外部变量后的变体叫 NBEATSx，本文不展开。 原论文见 arxiv:1905.10437，想在 MT5 里做对照实验的可先按那套双分支思路写回归块。
◍ 先把训练与行情接口库一次性挂好
用 PyTorch Forecasting 跑 N-BEATS 这类时序模型，第一步是把依赖库一次性 import 齐，避免后面训练到一半报模块缺失。下面这段代码把 Lightning 训练框架、数据处理、损失函数以及 MT5 实时接口全挂了进来，复制进你的 .py 文件头部就能直接用。
注意 import MetaTrader5 as mt 这一行：它和前面的深度学习库并列，意味着模型训完后能直接通过 mt 对象拉盘口和 Tick，不用再走 CSV 中转。外汇与贵金属杠杆高、滑点跳空频繁，这种直连方式虽快，但实盘前务必在策略测试器用历史数据跑通。
代码里还引入了 EarlyStopping 和 ModelCheckpoint，这两个回调一个管早停一个管存最优权重；NaNLabelEncoder 则专门处理行情数据里的空值缺口。少导任何一个，后面数据集构建都会直接崩。
class="kw">import lightning.pytorch as pl class="kw">import os from lightning.pytorch.callbacks class="kw">import EarlyStopping,ModelCheckpoint class="kw">import matplotlib.pyplot as plt class="kw">import numpy as np class="kw">import pandas as pd from pytorch_forecasting class="kw">import TimeSeriesDataSet,NBeats from pytorch_forecasting.data class="kw">import NaNLabelEncoder from pytorch_forecasting.metrics class="kw">import MQF2DistributionLoss from pytorch_forecasting.data.samplers class="kw">import TimeSynchronizedBatchSampler from lightning.pytorch.tuner class="kw">import Tuner class="kw">import MetaTrader5 as mt class="kw">import warnings class="kw">import json
「继承后改写数据集的加载逻辑」
在 PyTorch Forecasting 的 TimeSeriesDataSet 之上做一层子类，核心动机是把默认的按样本随机切批改成可同步采样。外汇与贵金属的时序样本如果按普通 DataLoader 随机打乱，同一批里可能混入不同时间锚点，训练出的价格行为模型容易学到虚假同步。 下面这段 Python 子类重写了 to_dataloader，默认 batch_size=64，并保留 drop_last 由外部显式控制，而不是原版里「train 且长度大于 batch_size 才丢弃」的隐式规则。当传入 batch_sampler='synchronized' 时，会套用 TimeSynchronizedBatchSampler，以 SequentialSampler 为基础做按时间同步的批切分。 直接把类贴进你的训练脚本，用 New_TmSrDt(...).to_dataloader(train=True, batch_sampler='synchronized', batch_size=64) 跑一遍，能观察到同一 batch 内样本时间索引差值缩小到 1 根 K 线级别（取决于数据频率），相比 shuffle=True 的默认方式更贴合盯盘时的连续窗口假设。这类改写涉及模型训练稳定性，外汇/贵金属杠杆高，回测与实盘偏差可能放大，需自行在 MT5 导出的 tick 级数据上验证。
class New_TmSrDt(TimeSeriesDataSet): &class="macro">#x27;&class="macro">#x27;&class="macro">#x27; rewrite dataset class &class="macro">#x27;&class="macro">#x27;&class="macro">#x27; def to_dataloader(self, train: class="type">bool = True, batch_size: class="type">int = class="num">64, batch_sampler: Sampler | str = None, shuffle:class="type">bool=False, drop_last:class="type">bool=False, **kwargs) -> DataLoader: default_kwargs = dict( shuffle=shuffle, # drop_last=train and len(self) > batch_size, drop_last=drop_last, # collate_fn=self._collate_fn, batch_size=batch_size, batch_sampler=batch_sampler, ) default_kwargs.update(kwargs) kwargs = default_kwargs # print(kwargs[&class="macro">#x27;drop_last&class="macro">#x27;]) if kwargs["batch_sampler"] is not None: sampler = kwargs["batch_sampler"] if isinstance(sampler, str): if sampler == "synchronized": kwargs["batch_sampler"] = TimeSynchronizedBatchSampler( SequentialSampler(self), batch_size=kwargs["batch_size"], shuffle=kwargs["shuffle"], drop_last=kwargs["drop_last"], ) else: raise ValueError(f"batch_sampler {sampler} unknown - see docstring for valid batch_sampler") del kwargs["batch_size"] del kwargs["shuffle"] del kwargs["drop_last"] class="kw">return DataLoader(self,**kwargs)
给黄金M15预测喂协变量时归错类就废了
做外汇或贵金属（如 XAUUSD 微点差品种 GOLD_micro）的时序预测，协变量不是随便堆指标就能用。模型本身只认数值序列，像美联储纪要、利率决议、非农这类外部文本数据它没有解析通道，硬塞进去只会报错或失效；能用的协变量必须是从 K 线柱里直接算出来的量，例如 open、high、low，或者你自己扩展的 MACD、ADX、RSI 等同周期技术指标。 在 New_TmSrDt() 里，变量要落进 static_categoricals / static_reals / timevaryingknown_reals / timevaryingunknown_reals 这几类。open、high、low 显然不是类别型，只在 known_reals 和 unknown_reals 之间二选一。关键点：如果你预测未来多根柱的 close，后面柱的 open/high/low 在当下根本未知，就只能放进 time_varying_unknown_reals；只有当你只测下一根单柱且那些值已实时可得，才适合 known_reals。 原文示例里把协变量接上的改法是：将 time_varying_unknown_reals=["close"] 扩成 ["close","high","open","low"]。注意 NBEATS 基础版并不支持协变量，这套写法面向的是支持 covariates 的变体（如 NBEATSx、GAGA）；外汇贵金属杠杆高、滑点跳空频繁，加协变量后回测误差可能收窄也可能放大，务必在 MT5 用历史数据重跑验证。 下面这段是 GOLD_micro 在 M15 周期取数并切分训练/验证集的骨架，重点看 time_varying_unknown_reals 的赋值位置和 training_cutoff 的计算逻辑： def get_data(mt_data_len:int): if not mt.initialize(): print('initialize() failed!') else: print(mt.version()) sb=mt.symbols_total() rts=None if sb > 0: rts=mt.copy_rates_from_pos("GOLD_micro",mt.TIMEFRAME_M15,0,mt_data_len) mt.shutdown() # print(len(rts)) rts_fm=pd.DataFrame(rts) rts_fm['time']=pd.to_datetime(rts_fm['time'], unit='s') rts_fm['time_idx']= rts_fm.index%(max_encoder_length+2*max_prediction_length) rts_fm['series']=rts_fm.index//(max_encoder_length+2*max_prediction_length) return rts_fm def spilt_data(data:pd.DataFrame, t_drop_last:bool, t_shuffle:bool, v_drop_last:bool, v_shuffle:bool): training_cutoff = data["time_idx"].max() - max_prediction_length #max:95 context_length = max_encoder_length prediction_length = max_prediction_length training = New_TmSrDt( data[lambda x: x.time_idx <= training_cutoff], time_idx="time_idx", target="close", categorical_encoders={"series":NaNLabelEncoder().fit(data.series)}, group_ids=["series"], time_varying_unknown_reals=["close"], max_encoder_length=context_length, max_prediction_length=prediction_length, ) validation = New_TmSrDt.from_dataset(training, data, min_prediction_idx=training_cutoff + 1) train_dataloader = training.to_dataloader(train=True, shuffle=t_shuffle,
def get_data(mt_data_len:<span class="keyword">class="type">int</span>): <span class="keyword">if</span> not mt.initialize(): print(<span class="class="type">class="kw">string">&class="macro">#x27;initialize() failed!&class="macro">#x27;</span>) <span class="keyword">else</span>: print(mt.version()) sb=mt.symbols_total() rts=None <span class="keyword">if</span> sb > <span class="number">class="num">0</span>: rts=mt.copy_rates_from_pos(<span class="class="type">class="kw">string">"GOLD_micro"</span>,mt.TIMEFRAME_M15,<span class="number">class="num">0</span>,mt_data_len) mt.shutdown() # print(len(rts)) rts_fm=pd.DataFrame(rts) rts_fm[<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>]=pd.to_datetime(rts_fm[<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>], unit=<span class="class="type">class="kw">string">&class="macro">#x27;s&class="macro">#x27;</span>) rts_fm[<span class="class="type">class="kw">string">&class="macro">#x27;time_idx&class="macro">#x27;</span>]= rts_fm.index%(max_encoder_length+<span class="number">class="num">2</span>*max_prediction_length) rts_fm[<span class="class="type">class="kw">string">&class="macro">#x27;series&class="macro">#x27;</span>]=rts_fm.index<span class="comment">class=class="str">"cmt">//(max_encoder_length+class="num">2*max_prediction_length)</span> <span class="keyword">class="kw">return</span> rts_fm def spilt_data(data:pd.DataFrame, t_drop_last:<span class="keyword">class="type">bool</span>, t_shuffle:<span class="keyword">class="type">bool</span>, v_drop_last:<span class="keyword">class="type">bool</span>, v_shuffle:<span class="keyword">class="type">bool</span>): training_cutoff = data[<span class="class="type">class="kw">string">"time_idx"</span>].max() - max_prediction_length <span class="preprocessor">class="macro">#max:<span class="number">class="num">95</span> </span>context_length = max_encoder_length prediction_length = max_prediction_length training = New_TmSrDt( data[lambda x: x.time_idx <= training_cutoff], time_idx=<span class="class="type">class="kw">string">"time_idx"</span>, target=<span class="class="type">class="kw">string">"close"</span>, categorical_encoders={<span class="class="type">class="kw">string">"series"</span>:NaNLabelEncoder().fit(data.series)}, group_ids=[<span class="class="type">class="kw">string">"series"</span>], time_varying_unknown_reals=[<span class="class="type">class="kw">string">"close"</span>], max_encoder_length=context_length, max_prediction_length=prediction_length, ) validation = New_TmSrDt.from_dataset(training, data, min_prediction_idx=training_cutoff + <span class="number">class="num">1</span>) train_dataloader = training.to_dataloader(train=True, shuffle=t_shuffle,