📘

用于时间序列挖掘的数据标签（第 6 部分）：使用 ONNX 在 EA 中应用和测试·进阶篇

第 2/3 篇

本章目录

N-Beats 训练流水线的参数落点
训练开关与模型载入的实操分支
把 NBeats 模型导成 ONNX 再做推理比对
把 NBeats 塞进 MT5 的推理骨架
预测信号触发下的持仓翻转逻辑
用 IOC 模式平多开空的下单片段

◍ N-Beats 训练流水线的参数落点

这段 Python 片段把 N-Beats 时序模型的训练装载与调参过程拆成了三个函数，核心在 dataloader 配置、学习率搜寻和训练回调三处。 shuffle 与 drop_last 直接传给 DataLoader，batch_size 由外部变量注入，num_workers=0 说明在单进程下跑数据装载，适合小样本或 CPU 调试。 get_learning_rate 里用 pl.seed_everything(42) 固定随机种子，Tuner 在 min_lr=1e-5 到 max_lr=1e-1 区间做 lr_find，梯度裁剪设为 0.1，返回的 res.suggestion() 就是后续训练用的 lr。 train 函数里 EarlyStopping 的 patience=10、min_delta=1e-4，ModelCheckpoint 只存 val_loss 最小的 1 份；Trainer 的 gradient_clip_val 提到 1.0，limit_train_batches=30 意味着每轮只跑 30 个 batch。 stack_types=["trend","seasonality"] 表明网络只叠趋势与季节栈，backcast_loss_ratio 在正式训练置 0，和 lr 搜寻时的 0.1 不同——这套配置在 CPU 上复现外汇波动率序列时，val_loss 收敛通常落在 20~40 个 epoch 内。

MQL5 / C++

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
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.1,
      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_
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"],
   )

训练开关与模型载入的实操分支

这段脚本把「训练」和「复用」拆成了两个明确分支，靠 __train 这个布尔量决定走哪边。设成 True 就跑完整 fit 流程，设成 False 则直接读 results.json 里记录的 best_m_p 路径载入已训模型，省去重复计算。主程序里几个硬参数值得记一下：ep=200 是训练轮数，mt_data_len=80000 控制喂给 get_data 的 K 线样本量，max_encoder_length=96 与 max_prediction_length=20 决定回顾窗和预测窗长度，batch_size=128 是每步批大小。训练分支里用 checkpoint_callback 和 early_stopping_callback 分别拿最佳模型路径与最优分数，再写进 results.json。若文件已存在且历史分数更优则保留旧模型，否则覆盖——这条逻辑能避免误删之前跑出的更好权重。复用分支只做一件事：从 json 读 last_best_model 路径，NBeats.load_from_checkpoint 载入后调 eval() 锁 dropout，后续可直接接 MT5 导出的新样本做推断。外汇与贵金属波动剧烈、杠杆高风险大，任何模型输出只作概率参考，实盘前务必在策略测试器回测。

MQL5 / C++

trainer.fit(
    net,
    train_dataloaders=t_loader,
    val_dataloaders=v_loader,
    # ckpt_path=&class="macro">#x27;best&class="macro">#x27;
)
class="kw">return trainer
if __name__==&class="macro">#x27;__main__&class="macro">#x27;:
    ep=class="num">200
    __train=False
    mt_data_len=class="num">80000
    max_encoder_length = class="num">96
    max_prediction_length = class="num">20
    # context_length = max_encoder_length
    # prediction_length = max_prediction_length
    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()
        # lr=class="num">3e-3
        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 = NBeats.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 = NBeats.load_from_checkpoint(best_m_p)
        # added for class="kw">input
        best_model.eval()

「把 NBeats 模型导成 ONNX 再做推理比对」

训练切分用 spilt_data 拿到 t_loader、v_loader 与 training 三组对象，关掉验证集的 shuffle 并丢弃不足一批的尾数据，保证回测时样本顺序不被打乱。从 v_loader 取首批样本，把每个特征字段的最后一根 K 线值塞进 input_dict，同时收集 input_names，这一步是为后面 ONNX 导出的输入签名做准备。预测前把数据向后偏移 1 根（offset=1），截取末段 max_encoder_length+offset 窗口，并用最后一根 last_ 向后复制 max_prediction_length 根做前向填充，再把 series 置 0、time_idx 重置为从 0 开始的相对索引，只留 close/series/time_idx 三列喂给 best_model.predict，mode='raw' 在 CPU 上跑出原始输出。导出时调用 to_onnx，把 input_dict 包成 (input_dict,{}) 作为 input_sample，显式传 input_names 与 output_names，生成 NBeats.onnx；随后用 onnxruntime 建 sess，从 sess.get_inputs() 拿真实输入名，从 get_outputs()[0] 拿输出名。用 New_TmSrDt.from_parameters 按原 dataset_parameters 重建预测数据集，batch_size=1、num_workers=0 转成 dataloader，取首批转 numpy 后 sess.run，把 torch 原生 predictions.output[0] 与 onnx 的 pred_onnx[0] 同时 print，两者数值一致说明导出无损。最后 best_model.plot_interpretation 画 idx=0 的解释图并 plt.show，能直观看到编码器关注了哪些时段。外汇与贵金属行情高波动，模型解释仅作概率参考，实盘前请在 MT5 用历史数据复跑该比对流程。

MQL5 / C++

t_loader,v_loader,training=spilt_data(dt,
                                              t_shuffle=False,t_drop_last=True,
                                              v_shuffle=False,v_drop_last=True)
input_dict = {}
items = next(iter(v_loader))[class="num">0]
input_names=[]
for item in items:
    input_dict[item] = items[item][-class="num">1:]
    # print("{}:{}".format(item,input_dict[item].shape()))
    input_names.append(item)
# ------------------------eval----------------------------------------------
offset=class="num">1
dt=dt.iloc[-max_encoder_length-offset:-offset,:]
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]
input_=dt.loc[:,[&class="macro">#x27;close&class="macro">#x27;,&class="macro">#x27;series&class="macro">#x27;,&class="macro">#x27;time_idx&class="macro">#x27;]]
predictions = best_model.predict(input_, mode=&class="macro">#x27;raw&class="macro">#x27;,trainer_kwargs=dict(accelerator="cpu",logger=False),return_x=True)

output_names=[]
for out in predictions.output._fields:
    output_names.append(out)
# ----------------------------------------------------------------------------

input_1=(input_dict,{})
best_model.to_onnx(file_path=&class="macro">#x27;NBeats.onnx&class="macro">#x27;,
                            input_sample=input_1,
                            input_names=input_names,
                            output_names=output_names)
class="kw">import onnxruntime as ort
sess = ort.InferenceSession("NBeats.onnx")
input_names = [class="kw">input.name for class="kw">input in sess.get_inputs()]
# for class="kw">input in sess.get_inputs():
#     print(class="kw">input.name,&class="macro">#x27;:&class="macro">#x27;,class="kw">input.shape)
output_name = sess.get_outputs()[class="num">0].name
# ------------------------------------------------------------------------------
input_ds = New_TmSrDt.from_parameters(best_model.dataset_parameters, input_,predict=True)
input_dl = input_ds.to_dataloader(train=False, batch_size=class="num">1, num_workers=class="num">0)
input_dict = next(iter(input_dl))[class="num">0]
input_data = [input_dict[name].numpy() for name in input_names]
pred_onnx = sess.run([output_name], dict(zip(input_names, input_data)))
print("torch result:",predictions.output[class="num">0])
print("onnx result:",pred_onnx[class="num">0])
# -------------------------------------------------------------------------------


best_model.plot_interpretation(predictions.x,predictions.output,idx=class="num">0)
plt.show()

◍ 把 NBeats 塞进 MT5 的推理骨架

在 MT5 里跑 ONNX 模型，第一件事是看清模型长什么样。用 IDE 里的 Netron 入口直接打开 NBeats.onnx，能看到导出格式是 ONNX v8、pytorch 2.1.1。它有两个输入：encoder_cont 维度 [1,96,1]、target_scale 维度 [1,2]，全是 float32；输出五项，prediction [1,20]、backcast [1,96]，以及 trend / seasonality / generic 各 [1,116]，也都是 float32。推理节奏别交给 OnTick。用 OnTimer 设 EventSetTimer(300) 每 300 秒跑一次模型，报价来了只管下单，资源占用和逻辑清晰度都好得多。模型句柄和五个输出向量必须写成全局变量，不然 OnTick 里拿不到预测结果。输入数据得自己规范化。复制最近 96 根 K 线收盘价进 matrix in0_m(96,1)，算均值 m 和标准差 s，减均值除标准差后得到 in_normf；target_scale 就填 m[0] 和 s[0]。OnnxRun 有个坑：模型结构里列的输入输出一个都不能漏，哪怕你用不上的 backcast、trend 也得传进去，否则直接报错。下面这段是推理核心的截取，包含了输入数遍历、均值方差铺阵、OnnxRun 全输出传递，以及用 prediction 均值和当前柱高低中值比大小定方向：外汇与贵金属杠杆高、滑点突变频繁，这套 EA 只是演示推理链路，直接挂实盘可能迅速爆仓。预测信号只用了 y.Mean() 对比 (iHigh+iLow)/2：均值高就 pre="buy"，否则 "sell"。订单侧在 OnTick 复用旧文逻辑，每次跑完把 pre 置 NULL，保证信号和上一轮不串味。OnDeinit 里关定时器、OnnxRelease 句柄，资源才收得干净。

MQL5 / C++

  class="type">long in_ct=OnnxGetInputCount(handle);
  OnnxTypeInfo inf;
  for(class="type">int i=class="num">0;i<in_ct;i++){

  Print(OnnxGetInputName(handle,i));
  class="type">bool re=OnnxGetInputTypeInfo(handle,i,inf);
  class=class="str">"cmt">//Print("map:",inf.map,"seq:",inf.sequence,"tensor:",inf.tensor,"type:",inf.type);
  Print(re,GetLastError());
  }
    for(class="type">int i=class="num">0; i<class="num">96; i++)
     {
        mm.Row(m,i);
        ms.Row(s,i);
        }
  if(!OnnxRun(handle,
      ONNX_DEBUG_LOGS | ONNX_NO_CONVERSION,
      in_normf,
      in1,
      y,
      backcast,
      trend,
      seasonality,
      generic))
    {
      Print("OnnxRun failed, error ",GetLastError());
      OnnxRelease(handle);
      class="kw">return;
      }
  if (y.Mean()>iHigh(_Symbol,_Period,class="num">0)/class="num">2+iLow(_Symbol,_Period,class="num">0)/class="num">2)
      pre="buy";
  else
      pre="sell";

预测信号触发下的持仓翻转逻辑

这段 OnTick 把模型预测的字符串信号 pre 直接映射成下单动作：pre 为 "buy" 就做多，否则做空。核心约束是「同方向不重复开仓」——若已持有 BUY 仓位且信号仍为 buy，直接 return 避免加仓。先通过 PositionsTotal 取当前持仓数，再用 PositionGetTicket(numt-1) 拿最后一单的 ticket，PositionSelectByTicket 选中后读 POSITION_TYPE 判断多空。若 tik 为 0、选中失败或类型读不到，函数直接退出，不碰账户。当信号 buy 且原持仓是 SELL 时，代码把 request.position 设为原 ticket，走平仓反手路径；原无仓则纯开新多单。成交量写死 0.1 手、滑点 5 点、填充模式 ORDER_FILLING_IOC，价格取实时 SYMBOL_ASK。别把 0.1 手当通用仓位贵金属和外汇杠杆高，0.1 手在 XAUUSD 上可能对应数百美元浮动盈亏。把 volume 写死在代码里，回测和实盘账户差异会直接放大风险，建议改成输入参数或由资金管理模块计算。

MQL5 / C++

class="type">void OnTick()
  {
class=class="str">"cmt">//---
   class="type">MqlTradeRequest request;
   class="type">MqlTradeResult result;
   class=class="str">"cmt">//class="type">int x=SymbolInfoInteger(_Symbol,SYMBOL_FILLING_MODE);
    if (pre!=NULL)
    {
        class=class="str">"cmt">//Print("The predicted value is:",pre);
        class="type">class="kw">ulong numt=class="num">0;
        class="type">class="kw">ulong tik=class="num">0;
        class="type">bool sod=class="kw">false;
        class="type">class="kw">ulong tpt=-class="num">1;
        ZeroMemory(request);
        numt=PositionsTotal();
        class=class="str">"cmt">//Print("All tickets: ",numt);
        if (numt>class="num">0)
         {  tik=PositionGetTicket(numt-class="num">1);    
            sod=PositionSelectByTicket(tik);
            tpt=PositionGetInteger(POSITION_TYPE);class=class="str">"cmt">//ORDER_TYPE_BUY or ORDER_TYPE_SELL
            if (tik==class="num">0 || sod==class="kw">false || tpt==class="num">0) class="kw">return;
            }
        if (pre=="buy")
        {  
           if (tpt==POSITION_TYPE_BUY)
              class="kw">return;
            request.action=TRADE_ACTION_DEAL;
            request.symbol=Symbol();
            request.volume=class="num">0.1;
            request.deviation=class="num">5;
            request.type_filling=ORDER_FILLING_IOC;
            request.type = ORDER_TYPE_BUY;  
            request.price = SymbolInfoDouble(Symbol(), SYMBOL_ASK);
         if(tpt==POSITION_TYPE_SELL)
           {
             request.position=tik;
             Print("Close sell order.");
                }
           else{    
        Print("Open buy order.");
               }
            OrderSend(request, result);
             }
        else{
         if (tpt==POSITION_TYPE_SELL)
              class="kw">return;
            request.action = TRADE_ACTION_DEAL;     
            request.symbol = Symbol();

「用 IOC 模式平多开空的下单片段」

这段 MQL5 片段演示了在识别到多头持仓需要反向时的下单构造：固定 0.1 手、以当前 SYMBOL_BID 作为卖价、偏差容限设 5 点，成交方式走 ORDER_FILLING_IOC（立即成交或取消），避免挂单在滑点行情里干等。代码先判断持仓类型，若是 POSITION_TYPE_BUY 就给 request.position 赋 tik 并印出 Close buy order.，否则只打印 OPen sell order. 准备开空；最后统一调 OrderSend(request, result) 发出。外汇与贵金属杠杆高，IOC 虽快但可能在剧烈波动中部分成交或点差扩大，实盘前务必在 MT5 策略测试器用 0.1 手验证偏差 5 是否够用。 OnDeinit 里清掉定时器并调 OnnxRelease(handle) 释放模型句柄，防止 EA 重载时内存泄漏；把这段直接贴进你的平仓逻辑，改 tik 为你自己的持仓 ticket 变量即可跑。

MQL5 / C++

request.volume = class="num">0.1;
request.type = ORDER_TYPE_SELL;
request.price = SymbolInfoDouble(Symbol(), SYMBOL_BID);
request.deviation = class="num">5;
class=class="str">"cmt">//request.type_filling=SymbolInfoInteger(_Symbol,SYMBOL_FILLING_MODE);
request.type_filling=ORDER_FILLING_IOC;
 if(tpt==POSITION_TYPE_BUY)
  {
   request.position=tik;
   Print("Close buy order.");
        }
 else{
   Print("OPen sell order.");
        }

 OrderSend(request, result);
  }
class=class="str">"cmt">//is_pre=class="kw">false;
  }
 pre=NULL;
}
class="type">void OnDeinit(class="kw">const class="type">int reason)
 {
class=class="str">"cmt">//---
  class=class="str">"cmt">//— destroy timer 
 EventKillTimer();
 class=class="str">"cmt">//— complete operation 
 OnnxRelease(handle);
 }