时间序列挖掘的数据标签（第4部分）：使用标签数据的可解释性分解·进阶篇
训练与验证数据加载器的参数落点
这段 Python 风格的加载器构造，决定了时序数据怎么喂进模型。训练集开启 shuffle 会打乱样本顺序，但外汇与贵金属分钟级序列一旦乱序，窗口内的前后依赖就断了，回测里可能虚高准确率，实盘大概率衰减。 num_workers=0 是刻意的单进程设定。MT5 外接 Python 环境跑 PyTorch 时，多 worker 在 Windows 下常触发 pickle 死锁，0 值虽慢却稳，适合先跑通再谈并行。 drop_last 控制尾数不足一个 batch 的片段是否丢弃。若 batch_size=64 而样本剩 70 条，设 True 只留 64 条，能避免最后小批的梯度抖动；设 False 则多喂 6 条，验证集指标可能因此偏移 0.3%~0.8%。
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)
class="kw">return train_dataloader,val_dataloader,training「调学习率前先看清接口差异」
在 PyTorch Lightning 里给 N-BEATS 找合适学习率，直接调 Tuner 的 lr_find 最省事。但接 NBeats.from_dataset 时有两个坑得先记住：它没有 hidden_size 参数可传，loss 也不能用 MQF2DistributionLoss 那套。 下面这段是实测可用的取数函数。pl.seed_everything(42) 锁随机种子，保证你本地跑出来的 lr 建议值可复现；Trainer 只跑 CPU、梯度裁剪 0.1，纯粹为轻量探测。 net 用 from_dataset 建，学习率先给 3e-2、权重衰减 1e-2、backcast_loss_ratio 设 0 让回看段不参与损失、优化器选 AdamW。lr_find 扫的范围从 1e-5 到 1e-1，返回的 res.suggestion() 就是建议起点。 外汇与贵金属序列用这类模型做辅助判断，本身属高风险实验，建议值只是概率意义上的较优区，不是保本参数。
def get_learning_rate(): pl.seed_everything(class="num">42) trainer = pl.Trainer(accelerator="cpu", gradient_clip_val=class="num">0.1,logger=False) net = NBeats.from_dataset( training, learning_rate=class="num">3e-2, weight_decay=class="num">1e-2, backcast_loss_ratio=class="num">0.0, optimizer="AdamW", ) res = Tuner(trainer).lr_find( net, train_dataloaders=t_loader, val_dataloaders=v_loader, min_lr=class="num">1e-5, max_lr=class="num">1e-1 ) # print(f"suggested learning rate: {res.suggestion()}") lr_=res.suggestion() class="kw">return lr_
◍ 训练函数里的早停与堆栈设定
把 N-BEATS 接到行情序列上，核心就是写一个 train() 把早停、检查点和模型初始化串起来。前面系列讲过分解思路，这里直接落代码，不绕弯。 EarlyStopping 盯的是 val_loss，min_delta 设 1e-4、patience=10，意味着验证损失连续 10 轮下降不到万分之一就停，能省掉不少无效 epoch。ModelCheckpoint 只留 val_loss 最小的 1 份，文件名带轮次和损失值，方便回看哪版更稳。 NBeats.from_dataset() 必须传 stack_types，这里用默认思路填了 ["trend", "seasonality"]，即可解释的趋势+季节栈；另有一个 "generic" 纯黑盒选项，想跑基准对比时可以换。weight_decay=1e-2 配合 AdamW，对过拟合偏高的小样本外汇序列更友好。 Trainer 里 limit_train_batches=30 限制每轮只跑 30 个 batch，梯度裁剪 1.0，CPU 上跑也能控住显存和抖动。trainer.fit 喂入 t_loader / v_loader 后返回，后续直接拿 trainer 做预测或继续 load 最佳 ckpt。
def train(): early_stop_callback = EarlyStopping(monitor="val_loss", min_delta=class="num">1e-4, patience=class="num">10, verbose=True, mode="min") ck_callback=ModelCheckpoint(monitor=&class="macro">#x27;val_loss&class="macro">#x27;, mode="min", save_top_k=class="num">1, filename=&class="macro">#x27;{epoch}-{val_loss:.2f}&class="macro">#x27;) trainer = pl.Trainer( max_epochs=ep, accelerator="cpu", enable_model_summary=True, gradient_clip_val=class="num">1.0, callbacks=[early_stop_callback,ck_callback], limit_train_batches=class="num">30, enable_checkpointing=True, ) net = NBeats.from_dataset( training, learning_rate=lr, log_interval=class="num">10, log_val_interval=class="num">1, weight_decay=class="num">1e-2, backcast_loss_ratio=class="num">0.0, optimizer="AdamW", stack_types = ["trend", "seasonality"], ) trainer.fit( net, train_dataloaders=t_loader, val_dataloaders=v_loader, # ckpt_path=&class="macro">#x27;best&class="macro">#x27; ) class="kw">return trainer
跑训练脚本前先把环境踩平
这套 NHiTS 价格预测的训练与推断逻辑沿用前文实现，核心参数已经固化在入口脚本里：epoch 设 200，编码器回望长度是 2×96=192 根 K 线，预测窗口 30 根，batch_size 128，单次喂入 200000 条行情样本。 动手前有个坑必须排掉——本机没装 TensorBoard 的话，脚本运行会抛一堆莫名其妙的回调错误，不是模型本身的问题。先 pip 装好再跑，能省掉大半调试时间。 代码里 __train 开关控制训练还是直接加载已有模型：设为 True 时走数据切分、学习率调度、早停与 checkpoint 回调，并把最优模型路径和分数写进 results.json；设为 False 则直接读 json 里的 last_best_model 做加载。 训练跑完会循环画出 10 张预测对照图（best_model.plot_prediction 里 range(10) 写死），随机抽一张就能肉眼看拟合偏差。外汇与贵金属波动剧烈，这类深度学习预测仅作概率参考，实盘试错务必用小仓位。
if __name__==&class="macro">#x27;__main__&class="macro">#x27;: ep=class="num">200 __train=False mt_data_len=class="num">200000 max_encoder_length = class="num">2*class="num">96 max_prediction_length = class="num">30 batch_size = class="num">128 info_file=&class="macro">#x27;results.json&class="macro">#x27; warnings.filterwarnings("ignore") dt=get_data(mt_data_len=mt_data_len) if __train: # print(dt) # dt=get_data(mt_data_len=mt_data_len) t_loader,v_loader,training=spilt_data(dt, t_shuffle=False,t_drop_last=True, v_shuffle=False,v_drop_last=True) lr=get_learning_rate() trainer__=train() m_c_back=trainer__.checkpoint_callback m_l_back=trainer__.early_stopping_callback best_m_p=m_c_back.best_model_path best_m_l=m_l_back.best_score.item() # print(best_m_p) if os.path.exists(info_file): with open(info_file,&class="macro">#x27;r+&class="macro">#x27;) as f1: last=json.load(fp=f1) last_best_model=last[&class="macro">#x27;last_best_model&class="macro">#x27;] last_best_score=last[&class="macro">#x27;last_best_score&class="macro">#x27;] if last_best_score > best_m_l: last[&class="macro">#x27;last_best_model&class="macro">#x27;]=best_m_p last[&class="macro">#x27;last_best_score&class="macro">#x27;]=best_m_l json.dump(last,fp=f1) else: with open(info_file,&class="macro">#x27;w&class="macro">#x27;) as f2: json.dump(dict(last_best_model=best_m_p,last_best_score=best_m_l),fp=f2) best_model = NHiTS.load_from_checkpoint(best_m_p) predictions = best_model.predict(v_loader, trainer_kwargs=dict(accelerator="cpu",logger=False), return_y=True) raw_predictions = best_model.predict(v_loader, mode="raw", return_x=True, trainer_kwargs=dict(accelerator="cpu",logger=False)) for idx in range(class="num">10): # plot class="num">10 examples best_model.plot_prediction(raw_predictions.x, raw_predictions.output, idx=idx, add_loss_to_title=True) plt.show() else: with open(info_file) as f: best_m_p=json.load(fp=f)[&class="macro">#x27;last_best_model&class="macro">#x27;] print(&class="macro">#x27;model path is:&class="macro">#x27;,best_m_p) best_model = NHiTS.load_from_checkpoint(best_m_p) offset=class="num">1 dt=dt.iloc[-max_encoder_length-offset:-offset,:]
「用末值填充做多步前向预测」
在拿到最新一根 K 线后，先把末尾行复制 max_prediction_length 次，制造一段‘静止’的未来窗口。这一步不引入任何外部假设，只是给模型一个统一长度的待预测容器。 df 里新增的 series 列全置 0，time_idx 则重算为距首行的偏移量，保证时间轴连续且从 0 计数。模型靠 time_idx 定位时序，而非原始时间戳。 调用 best_model.predict 时指定 mode='raw' 并在 cpu 上跑，可避开 gpu 依赖直接出原始张量；plot_prediction 配合 show_future_observed=False 只画预测轨迹。开 MT5 导出的收盘序列跑这段，能直观看到未来 N 根的水平外推，外汇与贵金属波动剧烈，此类外推仅作概率参考，实盘须自担高风险。
last_=dt.iloc[-class="num">1] # print(len(dt)) for i in range(class="num">1,max_prediction_length+class="num">1): dt.loc[dt.index[-class="num">1]+class="num">1]=last_ dt[&class="macro">#x27;series&class="macro">#x27;]=class="num">0 # dt[&class="macro">#x27;time_idx&class="macro">#x27;]=dt.apply(lambda x:x.index,args=class="num">1) dt[&class="macro">#x27;time_idx&class="macro">#x27;]=dt.index-dt.index[class="num">0] # dt=get_data(mt_data_len=max_encoder_length) predictions = best_model.predict(dt, mode=&class="macro">#x27;raw&class="macro">#x27;,trainer_kwargs=dict(accelerator="cpu",logger=False),return_x=True) best_model.plot_prediction(predictions.x,predictions.output,show_future_observed=False) plt.show()
◍ 把预测拆成季节与趋势
NBEATS 的特别之处，在于能把一条预测曲线硬性拆成季节性分量与趋势分量，而不是只给一个汇总值。本例只选了这两类基函数，所以分解结果天然只有两路；若换组合，分解维度也会跟着变。 训练跑完想看内部结构，或在线推理时想实时拆，都只要调一次 plot_interpretation。下面这段是原文给的调用方式，直接贴进你的 Python 训练脚本就能出图。 [CODE] for idx in range(10): # plot 10 examples best_model.plot_interpretation(x, raw_predictions, idx=idx) best_model.plot_interpretation(predictions.x,predictions.output,idx=0) [/CODE] 循环里对前 10 个样本逐个画解释图，最后再单独对第 0 号预测画一次。代码无参可调，复制即跑。 图出来后别被吓到：示例里分解效果看着一般，是因为没细调超参、关键数据指标也没科学设定，多数参数停在默认态。外汇与贵金属行情高波动、易跳空，这类未优化模型直接上实盘风险极高，仅适合在 MT5 外接 Python 环境里做离线验证。 留 10 个样本的解释图对比看，你能直观发现趋势项在哪几根被季节项反向拉扯——这一步本该在调参前做完，而不是之后。
for idx in range(class="num">10): # plot class="num">10 examples best_model.plot_interpretation(x, raw_predictions, idx=idx) best_model.plot_interpretation(predictions.x,predictions.output,idx=class="num">0)