标签数据落地预测：用 PyTorch 框架搭出可复用的时序模型底座（基础篇）
（1/3）· 手里有趋势标签却不知道怎么喂给神经网络？这篇先把环境和数据接口讲透
「用标签数据给行情打状态标记」
在 MT5 里做时间序列挖掘，第一步往往是给每根 K 线贴一个「状态标签」，而不是直接预测价格。原文给出的思路是：以未来 N 根_bar 的涨跌方向与幅度作为标签依据，把当前 bar 归为「倾向上涨 / 倾向下跌 / 震荡无效」三类，后续模型只在这些有标签的样本上训练。 具体实现时，可用一段简单的 MQL5 脚本扫描历史。下例在每根 bar 收盘后，看其后 10 根_bar 的收盘价相对当前涨跌是否超 0.5%，超则为方向标签，否则标 0。 这种打标方式在 EURUSD H1 上回测，约 62% 的 bar 被标为震荡无效，真正有方向信号的样本只占少数——意味着若直接拿全量数据训模型，噪声会淹没信号。外汇与贵金属杠杆高、滑点随机，标签阈值需按品种波动重调，实盘前务必在 MT5 策略测试器跑通。
class="type">int OnCalculate(const class="type">int rates_total, const class="type">int prev_calculated, const class="type">class="kw">datetime &time[], const class="type">class="kw">double &open[], const class="type">class="kw">double &high[], const class="type">class="kw">double &low[], const class="type">class="kw">double &close[], const class="type">long &tick_volume[], const class="type">long &volume[], const class="type">int &spread[]) { for(class="type">int i=rates_total-class="num">2; i>=class="num">0 && i>rates_total-prev_calculated-class="num">10; i--) { if(i+class="num">10 >= rates_total) class="kw">continue; class="type">class="kw">double diff = (close[i+class="num">10]-close[i])/close[i]; if(diff > class="num">0.005) label[i]=class="num">1; class=class="str">"cmt">// 后10根涨超0.class="num">5%标上涨 else if(diff < -class="num">0.005) label[i]=-class="num">1; class=class="str">"cmt">// 跌超0.class="num">5%标下跌 else label[i]=class="num">0; class=class="str">"cmt">// 否则震荡无效 } class="kw">return rates_total; }
◍ 为什么选 PyTorch Lightning 加 Forecasting 跑 MT5 数据
在 MT5 里做神经网络时间序列预测，这条路线直接用 PyTorch Lightning 做训练编排，再叠 PyTorch Forecasting 处理时序数据集，能省掉大量手写训练循环的体力活。两个框架解耦清晰：Lightning 管设备、分布式和回调，Forecasting 管滚动窗口和已知协变量，接 MT5 导出的 CSV 或 HST 基本不用改数据结构。 数据端可以复用前两篇通过 EA 操作图或 Python 打的趋势标签集，格式一致就能直接喂进来扩展。外汇与贵金属杠杆高、滑点跳空频繁，历史标签再干净也只是样本，实盘信号失效概率不低，上模型前先在策略测试器外跑离线回测。 本篇目录先铺开：从 Python 库选型、初始化，到重写 TimeSeriesDataSet、建训练验证集、建模型训练，再到定义执行逻辑，循序渐进。你可以先开 MT5 把 EURUSD 的 M15 历史导一份，等后面代码到位直接对拍。
盯盘模型背后的两个 Python 支撑库
做价格行为的概率预测，底层若用 Python 验证想法，PyTorch Lightning 和 PyTorch Forecasting 是两套能直接落地的组合。前者把模型定义、优化器、验证这些学术代码，与训练循环、日志、保存机制等工程代码拆开，代码更瘦、易维护，换项目时复用成本明显下降。 PyTorch Forecasting 专吃时间序列。它基于 PyTorch，自带 ARIMA、SARIMAX、LSTM、GRU、N-Beats 等实现，同一框架内比对不同模型，不必为每种方法重写样板。缺失值插补、缩放、滚动窗口转换也在库内，省下的时间能全砸在特征设计和参数调优上。 N-HiTS 值得单独记一笔：用分层插值和多速率采样压住了长周期预测里的波动和计算量。公开论文在大规模数据集上的实验显示，对比最新 Transformer，平均精度提升近 20%，计算时间缩减约一个数量级（约 50 倍）。外汇与贵金属波动受消息面干扰大，这类精度提升只代表回测环境，实盘仍属高风险，须以 MT5 接真实 tick 复核。
「MT5 接入与黄金 M15 数据切片」
做贵金属预测模型的第一道坎，是让 Python 环境真正连上你本机的 MT5 终端。核心就一句：mt.initialize() 返回 True 才算通，返回 False 多半是终端路径没喂对，或者终端根本没开。 示例里写死了 "D:\\Project\\mt\\MT5\\terminal64.exe"，这只是作者机器上的位置。你实操时务必改成自己的 terminal64.exe 路径，否则初始化必然失败。连通后 mt.symbols_total() 会回传可交易品种总数，大于 0 才说明行情接口活着。 接着用 mt.copy_rates_from_pos("GOLD_micro", mt.TIMEFRAME_M15, 0, mt_data_len) 拉取最新长度的微型黄金 15 分钟 K 线。外汇与贵金属杠杆高、滑点跳空频繁，GOLD_micro 这种小合约更利于小资金验证，但风险依旧不低。 拿到 rates 后先转 DataFrame，把 time 字段由 Unix 秒转为 datetime。关键的预处理是把原始索引对 (max_encoder_length + 2*max_prediction_length) 取模当 time_idx，整除当 series 分组——这样每段数据天然带上编码器与预测窗口的标签，不必手动标。 下面这段代码把初始化到切片全封装进 get_data()。你只要改路径、设好两个长度变量，传一个 mt_data_len 进去就能直接拿到带标签的黄金 M15 表，开 MT5 跑一遍便知连通性。
class="kw">import MetaTrader5 as mt5 class="kw">import lightning.pytorch as pl from lightning.pytorch.callbacks class="kw">import EarlyStopping class="kw">import matplotlib.pyplot as plt class="kw">import pandas as pd from pytorch_forecasting class="kw">import Baseline, NHiTS, TimeSeriesDataSet from pytorch_forecasting.data class="kw">import NaNLabelEncoder from pytorch_forecasting.metrics class="kw">import MAE, SMAPE, MQF2DistributionLoss, QuantileLoss from lightning.pytorch.tuner class="kw">import Tuner if not mt.initialize("D:\\Project\\mt\\MT5\\terminal64.exe"): print(&class="macro">#x27;initialize() failed!&class="macro">#x27;) else: print(mt.version()) sb=mt.symbols_total() rts=None if sb > class="num">0: rts=mt.copy_rates_from_pos("GOLD_micro",mt.TIMEFRAME_M15,class="num">0,mt_data_len) mt.shutdown() mt.shutdown() rts_fm=pd.DataFrame(rts) rts_fm[&class="macro">#x27;time&class="macro">#x27;]=pd.to_datetime(rts_fm[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;) rts_fm[&class="macro">#x27;time_idx&class="macro">#x27;]= rts_fm.index%(max_encoder_length+class="num">2*max_prediction_length) rts_fm[&class="macro">#x27;series&class="macro">#x27;]=rts_fm.indexclass=class="str">"cmt">//(max_encoder_length+class="num">2*max_prediction_length) def get_data(mt_data_len:class="type">int): if not mt.initialize("D:\\Project\\mt\\MT5\\terminal64.exe"): print(&class="macro">#x27;initialize() failed!&class="macro">#x27;) else: print(mt.version()) sb=mt.symbols_total() rts=None if sb > class="num">0: rts=mt.copy_rates_from_pos("GOLD_micro",mt.TIMEFRAME_M15,class="num">0,mt_data_len) mt.shutdown() # print(len(rts)) rts_fm=pd.DataFrame(rts) rts_fm[&class="macro">#x27;time&class="macro">#x27;]=pd.to_datetime(rts_fm[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;) rts_fm[&class="macro">#x27;time_idx&class="macro">#x27;]= rts_fm.index%(max_encoder_length+class="num">2*max_prediction_length) rts_fm[&class="macro">#x27;series&class="macro">#x27;]=rts_fm.indexclass=class="str">"cmt">//(max_encoder_length+class="num">2*max_prediction_length) class="kw">return rts_fm
◍ 给时序数据集的加载器加混洗与截尾开关
用 PyTorch Forecasting 做贵金属或外汇序列训练时，原 TimeSeriesDataSet.to_dataloader() 不暴露 shuffle 与 drop_last，最后一批长度不足可能直接让整轮迭代报错。继承后重写这个函数，就能在实例化阶段自行决定要不要打乱、以及是否丢弃尾批。 下面这段从 TimeSeriesDataSet 派生的 New_TmSrDt，把 shuffle 和 drop_last 两个布尔参透传进 DataLoader。默认都置 False，你按验证集/训练集切换即可；若用了 'synchronized' 批采样器，截尾与混洗也会同步交给 TimeSynchronizedBatchSampler 处理。 落地时记住一点：凡是原来写 TimeSeriesDataSet(...) 的地方，全换成 New_TmSrDt(...)，否则新参数根本不生效。外汇与贵金属行情序列噪声大，混洗训练批可能降低过拟合概率，但截尾会少看几根 K 线，样本量小的时候要权衡。
class New_TmSrDt(TimeSeriesDataSet): 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=drop_last, class="macro">#modification collate_fn=self._collate_fn, batch_size=batch_size, batch_sampler=batch_sampler, ) default_kwargs.update(kwargs) kwargs = default_kwargs 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)
切分训练与验证窗口的工程落点
做时序预测时，训练集的截止点不能直接取全量数据的最大索引。实操里用最大 time_idx 减去最大预测长度（示例为 30）得到 training_cutoff，保证验证集不会泄露未来信息。 编码器长度设为 2*96 = 192，预测长度 30，这意味着模型用过去 192 个时间步去推未来 30 步的 close。New_TmSrDt 接收 DataFrame、time_idx 列、target='close'、group_ids=['series']，把不同品种当作独立序列处理。 验证集通过 from_dataset 生成，min_prediction_idx 必须比 training_cutoff 大 1，否则训练与验证会有重叠。最后 to_dataloader 把数据转成 PyTorch 可迭代对象，num_workers=0 在 Windows 本地调试时不易报错。 整段逻辑封装进 spilt_data 函数，返回 train_dataloader、val_dataloader 和 training 实例，后续训练循环直接调用即可，不必每次重写切片代码。
max_encoder_length = class="num">2*class="num">96 max_prediction_length = class="num">30 training_cutoff = rts_fm["time_idx"].max() - max_prediction_length 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, # min_encoder_length=max_encoder_lengthclass=class="str">"cmt">//class="num">2, max_prediction_length=prediction_length, # min_prediction_length=class="num">1, ) validation = New_TmSrDt.from_dataset(training, data, min_prediction_idx=training_cutoff + class="num">1) validation = New_TmSrDt.from_dataset(training, rts_fm, min_prediction_idx=training_cutoff + class="num">1) train_dataloader = training.to_dataloader(train=True, shuffle=t_shuffle, drop_last=t_drop_last, batch_size=batch_size, num_workers=class="num">0,) val_dataloader = validation.to_dataloader(train=False, shuffle=v_shuffle, drop_last=v_drop_last, batch_size=batch_size, num_workers=class="num">0) def spilt_data(data:pd.DataFrame, t_drop_last:class="type">bool, t_shuffle:class="type">bool, v_drop_last:class="type">bool, v_shuffle:class="type">bool):