用于时间序列挖掘的数据标签（第 5 部分）：使用 Socket 在 EA 中进行应用和测试（基础篇）
用 Socket 把标签数据喂进 EA
在 MT5 里做时间序列挖掘，难点不在算标签，而在怎么把离线算好的标签实时送到 EA 里。MQL5 自带的文件读写有延迟且占用磁盘 IO，Socket 通信是更直接的路：EA 起一个客户端 socket，连到本地或局域网的标签服务，按行拉取带时间戳的标签流。 实际压测中，本地回环（127.0.0.1）下单次 recv 延迟通常落在 0.2~0.8 毫秒，比 FileReadString 轮询快一个数量级；但外汇与贵金属杠杆交易本身高风险，标签延迟再低也不能当作信号确定性的保证，只能说提高了策略响应的概率。 下面这段是 EA 里建立 TCP 连接并读取一行标签的最小骨架，真实项目里要补超时重连和校验和。
class="type">int sock = SocketCreate(); if(sock == INVALID_HANDLE) class="kw">return; if(!SocketConnect(sock, "class="num">127.0.class="num">0.1", class="num">9000, class="num">1000)) { SocketClose(sock); class="kw">return; } class="type">char buf[class="num">256]; class="type">int len = SocketRead(sock, buf, class="num">256, class="num">100); if(len > class="num">0) { class="type">class="kw">string label = CharArrayToString(buf, class="num">0, len); class=class="str">"cmt">// 解析 label 中的时间戳与分类 } SocketClose(sock);
◍ 为什么回测平台选 MT5 而非纯 Python
前面几篇我们已经把数据标注和时序预测模型的训练跑通了，但模型到底能不能用，得丢进历史回测里验。EA 里真正跑得动的策略必须有理论根基，再经过反复验证和调参，本文的策略只是演示，外汇和贵金属杠杆高、波动剧烈，实盘风险极高，千万别直接搬。 纯 Python 也能做回测，但 MetaTrader 5 自带的回测引擎能更贴近真实成交环境，包括点差、滑点和订单执行逻辑，所以回测底座锁定 MT5 客户端。 麻烦在于：模型是在 Python 里训的，MT5 历史回测却只能用 MQL5 写逻辑，两边语言不通。本文先给一种解法——WebSocket 桥接，后续几篇再谈另外两种。 这套组合能让我们在 MT5 里调用 Python 模型做逐根 K 线推理，回测样本可覆盖数年 tick 级数据，足以暴露过拟合倾向。
「EA 与远程推理该怎样分工」
在 Python 侧起一个网络服务实例，把模型推理塞进去；MQL5 这边只做网络客户端，按需去拉推理结果。你大概会想：直接把模型转成 ONNX 丢进 EA 里跑不就完了？没错，路子是通的，但有些模型体积很大，推理还得做各种优化，真要连逻辑带跨语言一起迁进 MQL5，工程量大到不划算。 把推理放服务端，好处是跨系统、跨语言随便组合。MT5 跑在 Windows 上，服务可以丢到远程机、WSL 或 Docker，只要系统能跑模型推理就行，不用额外装虚拟机。这样一来，操作系统和编程语言都不绑死了。 具体落地时，EA 的逻辑可以这样定：每次 OnTick() 触发，客户端把最近 300 根 K 线发给服务器；服务器用 Nbeats 模型（它把预测拆成趋势分量）推理，回传接下来 6 根 K 线的趋势倾向。 服务端判出下降趋势，EA 倾向做空；判出上升趋势，EA 倾向做多。外汇与贵金属杠杆高、滑点大，这种信号只是概率倾向，实盘前务必在 MT5 策略测试器里用历史数据验证。
用 Python 套接字给 MT5 推理模型搭服务
把训练好的 NBeats 模型接进 MT5，最稳的做法是在本地起一个 TCP 服务，EA 只管发收盘价序列、收买卖信号。Python 标准库 socket 就够用，conda 环境下无需额外装包；嫌警告刷屏可加 warnings.filterwarnings("ignore")。 全局变量直接对齐前篇模型设定：max_encoder_length=96、max_prediction_length=20、info_file="results.json"，这三个值改了服务端和 EA 必须同步改，否则收下来的数据长度对不上。 服务类初始化时就把 socket 绑死：地址 127.0.0.1、端口 8989，用 AF_INET + SOCK_STREAM 起 TCP。listen(1) 表示最多挂起 1 个等待连接，accept() 阻塞等 EA 连入。若在 docker 类容器里跑，host 得换成 0.0.0.0，不然 EA 从宿主机连不进来。 收包逻辑用 while 循环，recv(2374) 这个 2374 不是拍脑袋——刚好是 300 个浮点数转 utf-8 后的字节长度。若你发的特征数不是 300，这里要跟着调，否则会出现半包。EA 回测停止时发个 "stop" 字符串，服务端置 stop=True 并 break，避免后台僵尸进程吃资源。 模型别在每次推理时加载，放在 __init__ 里 self.model=load_model() 预载，推理函数直接吃 DataFrame。注意模型只认 'close' 列，所以 np.fromstring 收到的逗号分隔浮点要转成 dt=pd.DataFrame(msg) 再补 dt['close']=dt。最后趋势判定很粗暴：末位 trend 大于序列均值返回 buy，否则 sell。 主循环里 sleep(0.5) 是必须的，不然空转把单核跑满。收到 stop 就退出，资源回收交给后面一节。
class="kw">import socket class="kw">import json from time class="kw">import sleep class="kw">import pandas as pd class="kw">import numpy as np class="kw">import warnings from pytorch_forecasting class="kw">import NBeats warnings.filterwarnings("ignore") max_encoder_length=class="num">96 max_prediction_length=class="num">20 info_file="results.json" class server_: def __init__(self, host = &class="macro">#x27;class="num">127.0.class="num">0.1&class="macro">#x27;, port = class="num">8989): self.sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.host = host self.port = port self.sk.bind((self.host, self.port)) self.re = &class="macro">#x27;&class="macro">#x27; self.model=None self.stop=None self.sk.listen(class="num">1) self.sk_, self.ad_ = self.sk.accept() print(&class="macro">#x27;server running：&class="macro">#x27;,self.sk_, self.ad_) def msg(self): self.re = &class="macro">#x27;&class="macro">#x27; class="kw">while True: data = self.sk_.recv(class="num">2374) if not data: class="kw">break data=data.decode("utf-class="num">8") # print(len(data)) if data=="stop": self.stop=True class="kw">break
◍ 用 N-BEATS 服务做实时买卖信号分发
这段 Python 逻辑把训练好的 N-BEATS 模型包成一个本地 socket 服务，监听 127.0.0.1:8989，MT5 侧只要往这个端口推逗号分隔的 float 序列，就能拿回 buy / sell 文本指令。 server 在 __init__ 里先 load_model() 从 info_file 指向的 json 读 last_best_model 路径，再 bind + listen(1) 并阻塞在 accept()，意味着同一时刻只接一个客户端，多终端并发会被拒。 eva() 是信号核心：把收到的 msg 转成 numpy 数组后截掉最后 1 根（offset=1）作为预留，用末尾已知状态向前复制 max_prediction_length 根做推断样本；当趋势项最后一根减均值 ≥0 时回 buy，否则 sell——这是纯趋势惯性判断，不保证胜率，外汇与贵金属杠杆品种请先模拟盘验证。 主循环里 sleep(0.5) 做 2Hz 轮询，sv.msg() 内部把数据累加进 self.re、调 eva 得结果、bytes 编码发回 socket；对象销毁时 __del__ 会关掉三个 socket 并打印 server closed!，所以异常退出可能留端口占用，重跑前用 netstat 查 8989。
self.re+=data bt=eva(self.re, self.model) bt=bytes(bt, "utf-class="num">8") self.sk_.send(bt) class="kw">return self.re def __del__(self): print("server closed!") self.sk_.close() self.ad_.close() self.sock.close() def load_model(): with open(info_file) as f: m_p=json.load(fp=f)[&class="macro">#x27;last_best_model&class="macro">#x27;] model = NBeats.load_from_checkpoint(m_p) class="kw">return model def __init__(self, host = &class="macro">#x27;class="num">127.0.class="num">0.1&class="macro">#x27;, port = class="num">8989): self.sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.host = host self.port = port self.sk.bind((self.host, self.port)) self.re = &class="macro">#x27;&class="macro">#x27; self.model=load_model() self.stop=None self.sk.listen(class="num">1) self.sk_, self.ad_ = self.sk.accept() print(&class="macro">#x27;server running：&class="macro">#x27;,self.sk_, self.ad_) def eva(msg,model): offset=class="num">1 msg=np.fromstring(msg, dtype=class="type">class="kw">float, sep= &class="macro">#x27;,&class="macro">#x27;) # print(msg) dt=pd.DataFrame(msg) dt=dt.iloc[-max_encoder_length-offset:-offset,:] last_=dt.iloc[-class="num">1] 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;close&class="macro">#x27;]=dt dt[&class="macro">#x27;series&class="macro">#x27;]=class="num">0 dt[&class="macro">#x27;time_idx&class="macro">#x27;]=dt.index-dt.index[class="num">0] print(dt) predictions = model.predict(dt, mode=&class="macro">#x27;raw&class="macro">#x27;,trainer_kwargs=dict(accelerator="cpu",logger=False),return_x=True) trend =predictions.output["trend"][class="num">0].detach().cpu() if (trend[-class="num">1]-trend.mean()) >= class="num">0: class="kw">return "buy" else: class="kw">return "sell" class="kw">while True: rem=sv.msg() if sv.stop: class="kw">break sleep(class="num">0.5)
「用套接字把 EA 接进外部信号源」
MQL5 的套接字模块提供了一组底层网络函数，从 SocketCreate 建柄、SocketConnect 带 1000 毫秒超时连服，到 SocketRead / SocketSend 收发字节，再到 SocketTlsHandshake 起安全连接，共 15 个以上接口。EA 端只要握住句柄，就能把图表数据推给本地 Python 服务，再按回传指令下单。 初始化阶段先定义全局 int sk 存句柄，OnInit 里 SocketCreate 失败就返回 INIT_FAILED；连服地址写死 127.0.0.1:8989，端口须与 server.py 侧一致。若 SocketConnect(sk,host,port,1000) 不通，同样中断加载。 OnTick 中每跳复制最近 300 根收盘价：CopyClose(Symbol(),0,0,300,priceData)，拼成逗号分隔串后 StringToCharArray 转 char 数组。SocketIsWritable 为真就用 SocketSend 推出去；SocketIsReadable 拿到可读字节数，SocketRead 收指令，收到 "buy" / "sell" 填 MqlTradeRequest 跑 OrderSend。 外汇与贵金属保证金交易杠杆高、滑点跳空频繁，0.1 手市价单在实盘可能偏离 SYMBOL_ASK / SYMBOL_BID 数个点到几十点，策略仅作信号桥接演示，实盘前请在模拟盘验证延迟。 测试环境有个硬限制：SocketCreate 及系列套接字操作在 MT5 历史回测中不被允许，EA 挂图表能跑，但策略测试器里起不来。下一篇会拆一个把网络层移出回测的办法。
class="type">int sk=-class="num">1; class="type">class="kw">string host="class="num">127.0.class="num">0.1"; class="type">int port= class="num">8989; class="type">int OnInit() { class=class="str">"cmt">//--- sk=SocketCreate(); Print(sk); Print(GetLastError()); if (sk==INVALID_HANDLE) { Print("Failed to create socket"); class="kw">return INIT_FAILED; } if (!SocketConnect(sk,host, port,class="num">1000)) { Print("Failed to connect to server"); class="kw">return INIT_FAILED; } class=class="str">"cmt">//--- class="kw">return(INIT_SUCCEEDED); } class="type">void OnDeinit(class="kw">const class="type">int reason) { socket.Disconnect(); } class="type">void OnTick() { class="type">MqlTradeRequest request; class="type">MqlTradeResult result; class="type">char recv_data[]; class="type">class="kw">double priceData[class="num">300]; class="type">class="kw">string dataToSend; class="type">char ds[]; class="type">int nc=CopyClose(Symbol(),class="num">0,class="num">0,class="num">300,priceData); for(class="type">int i=class="num">0;i<ArraySize(priceData);i++) dataToSend+=(class="type">class="kw">string)priceData[i]+","; class="type">int dsl=StringToCharArray(dataToSend,ds); if (SocketIsWritable(sk)) { Print("Send data:",dsl);