MetaTrader 4 + Python 经由 ZeroMQ：安全、可测试的桥接协议（2026）

2026 维护说明。 2019 年原文记录了两个真实的安装问题，但没有定义桥接协议，也没有可运行代码。本版依据原标题补写一份新教程；文末逐字保留完整原始导出，并把其中 Visual C++ 2015 的说法视为特定构建要求，而不是 2026 年的通用前置条件。

ZeroMQ 可以在 MetaTrader 4 Expert Advisor（EA）与 Python 之间传递消息，但它只是一种传输层。它不会替你定义协议、保护不可信进程、消除订单结果不明的状态，也不会让策略盈利。下面的示例故意只实现本机 health 请求：适合学习通信故障，不是实盘交易系统。

1. 范围与安全约定

先在离线终端或模拟账户中使用本指南。测试传输层时关闭自动交易。示例：

• 只绑定 IPv4 回环地址 tcp://127.0.0.1:5557；
• 只接受白名单操作 health；
• 不发送凭据、账户标识、价格或订单；
• 对消息、等待、队列、重试与关闭设置上限；
• 重试时保持请求 ID 和序列化字节不变，供服务端去重；
• 不承诺收益或成交质量。

启用 Allow DLL imports 等于让原生代码取得终端进程的权限。只加载你已核验源码、构建、架构、依赖和校验和的 DLL。复制来的 DLL 是代码执行边界，不只是“连接器”。

2. 架构与信任边界

保守的职责划分把交易权留在 MT4：

market / broker
      |
      v
MT4 terminal -> EA policy gate -> reviewed MQL4-to-libzmq adapter
                                    || loopback, fixed protocol
                                    /
                              Python worker
                         analytics / health only

MT4 掌管终端状态、tick 时间、账户模式、风险上限及未来可能存在的下单动作。Python 可以计算或回答健康检查，但其回复只是 EA 本地策略的一个输入。切勿把“收到了 ZeroMQ 回复”理解为“该命令可以安全执行”。

这条边界包含三个独立版本组件：MQL4 声明、原生适配器/libzmq 二进制，以及 Python/pyzmq。即使 Python 正确，版本或 ABI 不匹配也可能使终端崩溃。MQL4 编译器无法完整验证外部函数参数；MetaQuotes 还说明，导入 DLL 的调用在调用模块线程中运行。

3. 配置 MT4 DLL，不猜测运行库

历史说明要求安装 Microsoft Visual C++ 2015 Redistributable；这可能适用于当时的 libzmq.dll。真正的规则是安装你所审核的确切二进制所需的运行库。现代构建可能依赖不同版本，也可能静态链接。

加载 EA 前：

1. 核对终端与 DLL 架构；不要靠改文件名掩盖位数不符。
2. 把审核过的库放到终端文档规定的 MQL4/Libraries 搜索路径，并核查所有依赖 DLL。
3. 让每个 MQL4 #import 签名及调用约定与适配器 ABI 一致。优先采用窄小、审核过的适配器，不要直接暴露庞大的 C API。
4. 只为此 EA 启用 Allow DLL imports；运行时如果 IsDllsAllowed() 为假，立即安全失败。
5. 检查 Experts 与 Journal。MetaQuotes 说明，DLL 缺失或被禁止时，EA 会停止，直到重新初始化。

不要仅因文件名与本教程相符，就下载任意“MT4 ZeroMQ bridge”。

4. 用 OnTimer 驱动，而不是 OnTick

OnTick() 面向新报价，不是可靠的消息泵。MetaQuotes 说明：前一次 OnTick() 尚在执行时，新 tick 会被忽略。在其中阻塞接收会造成行情陈旧，也会卡住 EA 的其他工作。

在 OnInit() 创建一个定时器，在 OnTimer() 只执行一次短小、有界或非阻塞的桥接步骤，并在 OnDeinit() 清除。每个 MQL4 程序只有一个定时器；已有计时事件排队或执行时不会再追加同类事件。这能形成背压，也意味着处理函数必须很快结束。不要跨事件线程和工作线程共享 ZeroMQ socket；libzmq 明确说明普通 REQ/REP socket 不是线程安全的。

5. 定义精确的 multipart 协议

ZeroMQ 保证 multipart 消息的原子性，却不会替应用定义 schema。本指南规定恰好两个应用可见帧：

帧	字节数	规则
0	7	ASCII 常量 LZMT4/1
1	最多 65,536	UTF-8 JSON；严格对象；禁止重复键、NaN 和无穷值

请求 schema（禁止额外键）：

字段	类型与约束
schema	固定值 lazying.mt4.bridge.request
version	整数 1
request_id	UUID 字符串；重试时不变
client_id	1–64 个 ASCII 字母、数字、.、_ 或 -
session_id	UUID 字符串；每个有意启动的客户端会话新建
sequence	正整数，从 1 开始逐一递增
sent_at_utc	以 Z 结尾的 RFC 3339 UTC 时间
max_age_ms	100 至 60,000 的整数
operation	本版本固定为 health
payload	空 JSON 对象

回复的键也必须精确为：schema、version、三个关联值（request_id、session_id、sequence）、status、processed_at_utc、result、error。成功时 status: "ok"、result 为对象、error: null；拒绝时 status: "error"、result: null，错误对象包含稳定的 code 和安全的 message。只有请求无法解码时，关联值才可为 null。

不要在这条边界使用 send_pyobj() 或 recv_pyobj()。PyZMQ 文档说明它们依赖 Python pickle；解码不可信 pickle 可能执行任意代码。

6. 分开定义价格新鲜度与时间语义

至少要区分三种时钟：

• MqlTick.time 是某一品种最新已知 tick 携带的服务器时间。
• 在 OnTimer() 中，TimeCurrent() 表示 Market Watch 中任一品种最后已知的服务器时间；它不能证明目标品种刚刚成交。
• 单调时钟适合计算本机超时，却不能给另一台机器上的事件定时。

未来的价格消息因此应包含 symbol、bid、ask、tick_server_time、terminal_received_at_utc 和有效期。EA 在采取动作前必须再次将目标品种的最新 tick 与终端当前状态比较。Python 的墙钟时间不能证明报价新鲜。超过声明时钟偏差的未来时间和晚于 max_age_ms 的消息均应拒绝；若桥接跨主机，两端还须同步时钟。

健康检查示例用 UTC 检查消息年龄，用单调时钟测重试耗时；它不携带价格。

7. 理解 REQ/REP 故障状态

REQ/REP 是锁步状态机：REQ 先发后收，REP 先收后发。严格 REQ 超时后，在同一个 socket 上再次发送会违反状态机。应重建 socket、重连，并用相同 request_id 重发同一份序列化请求。

一个棘手但正常的过程是：

1. Python 发出请求。
2. 服务端已经处理。
3. 回复延迟或丢失。
4. Python 超时，无法知道处理是否发生。

所以传输重试必须配合应用层幂等。下方服务端按 (client_id, session_id, request_id) 缓存成功回复；若同一 ID 对应不同字节就拒绝。它先查缓存、再验序号，真正的重试才能得到先前回复。内存缓存足够演示健康检查，却不适用于订单；生产执行器需要持久账本，并与 MT4 权威状态对账。

受支持的 libzmq 构建还提供 relaxed/correlated REQ 选项。官方文档警告 relaxed 模式会丢弃较早回复，应配合 correlation，避免把旧回复错配给新请求。本示例采用严格 REQ 并显式重建 socket，让故障行为更清楚。

8. 限制队列、等待与关闭时间

在 connect() 或 bind() 之前设置连接相关 socket 选项：

控制项	示例策略	原因
RCVTIMEO、SNDTIMEO	客户端 600 ms	避免无限阻塞；处理 EAGAIN/zmq.Again
LINGER	一次性请求 socket 为 0	有界替换/关闭；未发消息会被丢弃
SNDHWM、RCVHWM	10 条消息	限制传输队列；实际行为取决于传输与 socket 类型
MAXMSGSIZE	65,536 字节	由 libzmq 拒绝过大的入站帧，并辅以应用检查
IMMEDIATE	客户端为 1	连接完成前不排队发送数据
RECONNECT_IVL、RECONNECT_IVL_MAX	100/1,000 ms	限制重连频率并退避

默认 linger 和超时都可能无限等待，不要把这些默认值带进 EA 生命周期。高水位按消息数计算，不按字节；它也不能替代 65,536 字节的应用上限。

9. 身份认证与加密

回环地址加固定端口是最安全的起步范围，但其他本机进程仍可能连接。添加操作系统防火墙规则，并以最小权限账户运行。

跨网络时，ZeroMQ 安全机制并不等价：

• ZMTP NULL 不提供认证或保密性。
• PLAIN 以明文发送用户名与密码，本身不适合不可信网络。
• CURVE 是 ZeroMQ 用于认证与保密性的机制。

只有当确切 libzmq 构建以及 MQL4 适配器均暴露兼容的 CURVE 配置，而且你已核验密钥分发与服务器公钥固定时，才使用 CURVE。否则保持端点私有，并叠加独立认证的加密隧道。绝不要把裸 MT4 bridge 端口发布到互联网。安全还要覆盖元数据和密钥；日志中不得记录秘密。

10. 幂等与顺序

区分三个概念：

• request_id：一个逻辑请求的身份；其所有重试保持不变。
• session_id：一次有意客户端运行的身份；只在开始新会话时改变。
• sequence：会话内预期顺序；不可静默重置。

缓存回复及规范请求字节的哈希。同一 ID 若带不同字节必须拒绝；若序号跳号或倒退，且又不匹配已缓存请求，也必须拒绝。示例把回复缓存和每会话序号表都限制为 1,024 项、60 秒。状态过期或被淘汰后，仍在运行的客户端必须生成新的 session_id，并让这个新会话从序号 1 开始，不能猜测旧会话进度。这是客户端约定：有界的演示服务端无法永久记住所有已过期 ID，否则它仍会接受旧 ID 下的序号 1。

健康检查天然无害，交易命令则不是。在添加任何交易操作前，应定义持久命令账本、明确状态机（received、validated、submitted、confirmed、rejected、unknown），并依据平台订单/历史对账。绝不能把 unknown 提交当成确定失败后直接重试。

11. 重连、看门狗与断路器

传输重连不等于应用恢复。实用的看门狗应追踪最后一个有效且关联正确的回复、连续超时、socket 重建、协议错误及各品种最后新鲜 tick。少量连续失败后打开断路器：

• 停止接收新的外部命令；
• 不因 bridge 故障就平仓或反向开仓；
• 继续在 MT4 本地观察；
• 健康试运行一段时间后才能恢复，最好还要求手动重新启用。

Python、EA、终端或计算机重启时，创建新会话；接受命令前先对账持久状态。不能仅从 TCP 重连推断应用状态。

12. 最小本机健康检查协议

在一个空的临时目录创建以下三个文件。它们有意不包含订单操作。

protocol.py：

from __future__ import annotations

import json
import re
import uuid
from datetime import datetime, timezone
from typing import Any

PROTOCOL = b"LZMT4/1"
MAX_FRAME_BYTES = 65_536
REQUEST_SCHEMA = "lazying.mt4.bridge.request"
REPLY_SCHEMA = "lazying.mt4.bridge.reply"
REQUEST_KEYS = {
    "schema", "version", "request_id", "client_id", "session_id",
    "sequence", "sent_at_utc", "max_age_ms", "operation", "payload",
}
REPLY_KEYS = {
    "schema", "version", "request_id", "session_id", "sequence",
    "status", "processed_at_utc", "result", "error",
}
CLIENT_ID = re.compile(r"[A-Za-z0-9_.-]{1,64}Z")
ERROR_CODE = re.compile(r"[A-Z0-9_]{1,64}Z")


class ProtocolError(ValueError):
    pass


def utc_now() -> str:
    return (
        datetime.now(timezone.utc)
        .isoformat(timespec="milliseconds")
        .replace("+00:00", "Z")
    )


def parse_utc(value: str) -> datetime:
    if not isinstance(value, str) or not value.endswith("Z"):
        raise ProtocolError("timestamp must be an RFC 3339 UTC string ending in Z")
    try:
        parsed = datetime.fromisoformat(value[:-1] + "+00:00")
    except ValueError as exc:
        raise ProtocolError("timestamp is invalid") from exc
    if parsed.tzinfo is None or parsed.utcoffset() != timezone.utc.utcoffset(parsed):
        raise ProtocolError("timestamp must use UTC")
    return parsed


def _no_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ProtocolError(f"duplicate JSON key: {key}")
        result[key] = value
    return result


def _reject_constant(value: str) -> None:
    raise ProtocolError(f"non-finite JSON number: {value}")


def encode_object(value: dict[str, Any]) -> bytes:
    try:
        raw = json.dumps(
            value,
            ensure_ascii=True,
            allow_nan=False,
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")
    except (TypeError, ValueError) as exc:
        raise ProtocolError("object is not strict JSON") from exc
    if len(raw) > MAX_FRAME_BYTES:
        raise ProtocolError("JSON frame is too large")
    return raw


def decode_object(raw: bytes) -> dict[str, Any]:
    if len(raw) > MAX_FRAME_BYTES:
        raise ProtocolError("JSON frame is too large")
    try:
        value = json.loads(
            raw.decode("utf-8"),
            object_pairs_hook=_no_duplicate_keys,
            parse_constant=_reject_constant,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ProtocolError("frame is not strict UTF-8 JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("JSON root must be an object")
    return value


def _uuid(value: Any, field: str) -> None:
    if not isinstance(value, str):
        raise ProtocolError(f"{field} must be a UUID string")
    try:
        uuid.UUID(value)
    except ValueError as exc:
        raise ProtocolError(f"{field} must be a UUID string") from exc


def decode_request(parts: list[bytes]) -> tuple[dict[str, Any], bytes]:
    if len(parts) != 2 or parts[0] != PROTOCOL:
        raise ProtocolError("expected exactly two frames with protocol LZMT4/1")
    request = decode_object(parts[1])
    if set(request) != REQUEST_KEYS:
        raise ProtocolError("request keys do not match version 1")
    if request["schema"] != REQUEST_SCHEMA or request["version"] != 1:
        raise ProtocolError("unsupported request schema or version")
    _uuid(request["request_id"], "request_id")
    _uuid(request["session_id"], "session_id")
    if not isinstance(request["client_id"], str) or not CLIENT_ID.fullmatch(request["client_id"]):
        raise ProtocolError("client_id has an invalid format")
    if type(request["sequence"]) is not int or request["sequence"] < 1:
        raise ProtocolError("sequence must be a positive integer")
    if type(request["max_age_ms"]) is not int or not 100 <= request["max_age_ms"] <= 60_000:
        raise ProtocolError("max_age_ms is out of range")
    parse_utc(request["sent_at_utc"])
    if request["operation"] != "health" or request["payload"] != {}:
        raise ProtocolError("only health with an empty payload is allowed")
    return request, parts[1]


def make_reply(
    request: dict[str, Any] | None,
    *,
    status: str,
    result: dict[str, Any] | None,
    error: dict[str, str] | None,
) -> dict[str, Any]:
    return {
        "schema": REPLY_SCHEMA,
        "version": 1,
        "request_id": request.get("request_id") if request else None,
        "session_id": request.get("session_id") if request else None,
        "sequence": request.get("sequence") if request else None,
        "status": status,
        "processed_at_utc": utc_now(),
        "result": result,
        "error": error,
    }


def decode_reply(parts: list[bytes]) -> dict[str, Any]:
    if len(parts) != 2 or parts[0] != PROTOCOL:
        raise ProtocolError("reply framing is invalid")
    reply = decode_object(parts[1])
    if set(reply) != REPLY_KEYS:
        raise ProtocolError("reply keys do not match version 1")
    if reply["schema"] != REPLY_SCHEMA or reply["version"] != 1:
        raise ProtocolError("unsupported reply schema or version")
    if reply["status"] not in {"ok", "error"}:
        raise ProtocolError("reply status is invalid")
    for field in ("request_id", "session_id"):
        if reply[field] is not None:
            _uuid(reply[field], field)
    if reply["sequence"] is not None and (
        type(reply["sequence"]) is not int or reply["sequence"] < 1
    ):
        raise ProtocolError("reply sequence is invalid")
    parse_utc(reply["processed_at_utc"])
    if reply["status"] == "ok":
        if not isinstance(reply["result"], dict) or reply["error"] is not None:
            raise ProtocolError("success reply shape is invalid")
    else:
        error = reply["error"]
        if (
            reply["result"] is not None
            or not isinstance(error, dict)
            or set(error) != {"code", "message"}
            or not isinstance(error.get("code"), str)
            or not ERROR_CODE.fullmatch(error["code"])
            or not isinstance(error.get("message"), str)
            or not 1 <= len(error["message"]) <= 256
        ):
            raise ProtocolError("error reply shape is invalid")
    return reply

health_server.py：

from __future__ import annotations

import argparse
import hashlib
import json
import time
from collections import OrderedDict
from datetime import datetime, timezone

import zmq

from protocol import (
    MAX_FRAME_BYTES,
    PROTOCOL,
    ProtocolError,
    decode_request,
    encode_object,
    make_reply,
    parse_utc,
)

ENDPOINT = "tcp://127.0.0.1:5557"
CACHE_LIMIT = 1_024
CACHE_TTL_SECONDS = 60
STREAM_LIMIT = 1_024
STREAM_TTL_SECONDS = 60
FUTURE_SKEW_MS = 5_000


def log(event: str, **fields: object) -> None:
    print(json.dumps({"event": event, **fields}, sort_keys=True), flush=True)


def configure(socket: zmq.Socket) -> None:
    socket.setsockopt(zmq.LINGER, 0)
    socket.setsockopt(zmq.RCVTIMEO, 1_000)
    socket.setsockopt(zmq.SNDTIMEO, 1_000)
    socket.setsockopt(zmq.RCVHWM, 10)
    socket.setsockopt(zmq.SNDHWM, 10)
    socket.setsockopt(zmq.MAXMSGSIZE, MAX_FRAME_BYTES)


def prune_expired(mapping: OrderedDict, now_monotonic: float, ttl_seconds: float) -> None:
    for expired_key in [
        item_key
        for item_key, item in mapping.items()
        if now_monotonic - item[-1] > ttl_seconds
    ]:
        del mapping[expired_key]


def trim_oldest(mapping: OrderedDict, limit: int) -> None:
    while len(mapping) > limit:
        mapping.popitem(last=False)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--delay-first-ms", type=int, default=0)
    args = parser.parse_args()
    if not 0 <= args.delay_first_ms <= 5_000:
        parser.error("--delay-first-ms must be between 0 and 5000")

    cache: OrderedDict[
        tuple[str, str, str], tuple[str, dict[str, object], float]
    ] = OrderedDict()
    stream_state: OrderedDict[tuple[str, str], tuple[int, float]] = OrderedDict()
    delayed = False
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    configure(socket)
    socket.bind(ENDPOINT)
    log("ready", endpoint=ENDPOINT, pyzmq=zmq.pyzmq_version(), libzmq=zmq.zmq_version())

    try:
        while True:
            try:
                parts = socket.recv_multipart()
            except zmq.Again:
                continue

            started = time.monotonic()
            request = None
            duplicate = False
            try:
                request, raw = decode_request(parts)
                now = datetime.now(timezone.utc)
                age_ms = (now - parse_utc(request["sent_at_utc"])).total_seconds() * 1_000
                if age_ms < -FUTURE_SKEW_MS:
                    raise ProtocolError("request timestamp is too far in the future")
                if age_ms > request["max_age_ms"]:
                    raise ProtocolError("request is stale")

                key = (request["client_id"], request["session_id"], request["request_id"])
                stream = (request["client_id"], request["session_id"])
                fingerprint = hashlib.sha256(raw).hexdigest()
                now_monotonic = time.monotonic()
                prune_expired(cache, now_monotonic, CACHE_TTL_SECONDS)
                prune_expired(stream_state, now_monotonic, STREAM_TTL_SECONDS)
                cached = cache.get(key)
                if cached is not None:
                    if cached[0] != fingerprint:
                        raise ProtocolError("request_id was reused with different bytes")
                    reply = cached[1]
                    duplicate = True
                else:
                    expected = stream_state.get(stream, (0, now_monotonic))[0] + 1
                    if request["sequence"] != expected:
                        raise ProtocolError(f"expected sequence {expected}")
                    reply = make_reply(
                        request,
                        status="ok",
                        result={"service": "python-local-health", "protocol": "LZMT4/1"},
                        error=None,
                    )
                    cache[key] = (fingerprint, reply, now_monotonic)
                    stream_state[stream] = (request["sequence"], now_monotonic)
                    stream_state.move_to_end(stream)
                    trim_oldest(cache, CACHE_LIMIT)
                    trim_oldest(stream_state, STREAM_LIMIT)
            except ProtocolError as exc:
                reply = make_reply(
                    request,
                    status="error",
                    result=None,
                    error={"code": "PROTOCOL_REJECTED", "message": str(exc)},
                )

            if args.delay_first_ms and not delayed and reply["status"] == "ok":
                delayed = True
                time.sleep(args.delay_first_ms / 1_000)

            try:
                socket.send_multipart([PROTOCOL, encode_object(reply)])
            except zmq.Again:
                log("send_timeout", action="exit_for_supervisor_restart")
                return 2
            log(
                "reply",
                request_id=reply["request_id"],
                sequence=reply["sequence"],
                status=reply["status"],
                duplicate=duplicate,
                elapsed_ms=round((time.monotonic() - started) * 1_000, 1),
            )
    except KeyboardInterrupt:
        log("stopping")
        return 0
    finally:
        socket.close(linger=0)
        context.term()


if __name__ == "__main__":
    raise SystemExit(main())

health_client.py：

from __future__ import annotations

import json
import time
import uuid

import zmq

from protocol import MAX_FRAME_BYTES, PROTOCOL, ProtocolError, decode_reply, encode_object, utc_now

ENDPOINT = "tcp://127.0.0.1:5557"
TIMEOUT_MS = 600
MAX_ATTEMPTS = 3


def open_request_socket(context: zmq.Context) -> zmq.Socket:
    socket = context.socket(zmq.REQ)
    socket.setsockopt(zmq.LINGER, 0)
    socket.setsockopt(zmq.RCVTIMEO, TIMEOUT_MS)
    socket.setsockopt(zmq.SNDTIMEO, TIMEOUT_MS)
    socket.setsockopt(zmq.RCVHWM, 10)
    socket.setsockopt(zmq.SNDHWM, 10)
    socket.setsockopt(zmq.MAXMSGSIZE, MAX_FRAME_BYTES)
    socket.setsockopt(zmq.IMMEDIATE, 1)
    socket.setsockopt(zmq.RECONNECT_IVL, 100)
    socket.setsockopt(zmq.RECONNECT_IVL_MAX, 1_000)
    socket.connect(ENDPOINT)
    return socket


def main() -> int:
    request = {
        "schema": "lazying.mt4.bridge.request",
        "version": 1,
        "request_id": str(uuid.uuid4()),
        "client_id": "local-smoke-test",
        "session_id": str(uuid.uuid4()),
        "sequence": 1,
        "sent_at_utc": utc_now(),
        "max_age_ms": 10_000,
        "operation": "health",
        "payload": {},
    }
    frames = [PROTOCOL, encode_object(request)]
    context = zmq.Context()
    started = time.monotonic()
    try:
        for attempt in range(1, MAX_ATTEMPTS + 1):
            socket = open_request_socket(context)
            try:
                socket.send_multipart(frames)
                reply = decode_reply(socket.recv_multipart())
            except zmq.Again:
                print(json.dumps({"event": "timeout", "attempt": attempt}))
            except ProtocolError as exc:
                print(json.dumps({"event": "protocol_error", "message": str(exc)}))
                return 2
            else:
                if (
                    reply["request_id"] != request["request_id"]
                    or reply["session_id"] != request["session_id"]
                    or reply["sequence"] != request["sequence"]
                ):
                    print(json.dumps({"event": "correlation_error"}))
                    return 2
                print(json.dumps({
                    "event": "reply",
                    "attempt": attempt,
                    "elapsed_ms": round((time.monotonic() - started) * 1_000, 1),
                    "reply": reply,
                }, sort_keys=True))
                return 0 if reply["status"] == "ok" else 2
            finally:
                socket.close(linger=0)
            time.sleep(0.15)
        print(json.dumps({"event": "unavailable", "attempts": MAX_ATTEMPTS}))
        return 1
    finally:
        context.term()


if __name__ == "__main__":
    raise SystemExit(main())

每个 socket 都在同一线程内创建、使用和关闭。序列化后的 frames 在重试循环外只构造一次。

13. 接入 MT4 前先在本机测试

使用当前 Python 环境以及适合该环境的 PyZMQ 官方发行包：

python -m venv .venv
. .venv/bin/activate                  # Windows PowerShell: .venvScriptsActivate.ps1
python -m pip install --upgrade pip pyzmq
python health_server.py

在第二个终端运行：

python health_client.py

随后停止服务端，用故意延迟首个回复的方式重启一次：

python health_server.py --delay-first-ms 900

客户端应先报告一次超时，重建严格 REQ socket，重发同一字节，并在后续尝试收到缓存回复。服务端应把后一次请求记录为 duplicate: true。还应测试：服务端不存在、畸形 JSON、多一帧、同一 ID 改变内容、跳序号、过期时间戳，以及 Ctrl-C 正常关闭。

这只证明本机 framing、校验、超时和重试，不代表 MQL4 适配器已经通过。对该边界应在两端添加逐字节 fixture 测试，在一次性终端中核验位数/ABI，并先把 MQL4 生成的帧与 Python fixture 比较，再在正常配置中允许 DLL 导入。

14. 扩展前的 MT4 端安全门

在下列事项都已明确并独立测试之前，只保留 health：

• 模拟账户门（IsDemo()）与醒目的手动启用/停用开关；
• DLL 权限门，以及终端自己的交易权限/交易上下文检查；
• 操作和品种白名单；绝不接受任意 MQL 函数名；
• 动作前用 SymbolInfoTick() 检查目标品种当前 tick 的新鲜度；
• 最大规模、风险敞口、未完成请求数、点差/滑点策略和单日亏损；
• 以 MT4 权威权益为依据、需人工重启的锁定式单日止损；
• 每个已提交或结果不明命令的持久幂等与对账；
• 对畸形、迟到、重复、乱序或不可用消息安全失败。

即使 Python 另有一套风控，也要把强制限制保留在 EA。陈旧、被入侵或断线的 Python 进程不能放宽终端上限。只在模拟账户测试；模拟行为也不能保证实盘流动性、滑点或成交。

15. 可观测性与操作检查表

用 UTC 时间和单调耗时记录结构化事件。实用字段包括协议版本、EA/适配器构建、pyzmq/libzmq 版本、请求 ID、会话 ID、序号、操作、年龄、延迟、结果、重复标志、拒绝代码、连续失败数与 socket 重建数。对账户信息脱敏，永不记录密钥；日后 payload 可能包含策略数据，也不应原样记录。

每次测试前：

1. 核验二进制校验和、位数、ABI 声明、依赖来源和 DLL 权限。
2. 确认回环绑定与防火墙范围，检查端口没有被意外进程占用。
3. 确认两端协议/版本、严格 framing、schema fixture 和大小上限。
4. 确认超时、HWM、linger、重试数、缓存容量及过期策略都有上限。
5. 确认看门狗会打开断路器，且不能静默重新启用。
6. 确认 Python 停止、变慢、发送畸形消息或重启时，MT4 仍响应。
7. 保存已脱敏日志、版本与测试结果。

16. 主要官方文档

• MetaQuotes：导入函数与 DLL 加载行为、`#import` 声明及 `IsDllsAllowed()`
• MetaQuotes：`EventSetTimer()`、事件生成规则、`SymbolInfoTick()`、`TimeCurrent()`、`IsDemo()`及 `AccountEquity()`
• libzmq：`zmq_socket`及 `zmq_setsockopt`
• ZeroMQ RFC：ZMTP 安全机制、ZAP 认证及 CurveZMQ
• PyZMQ：序列化指南及核心 API
• Python：`json` 严格性选项及 `uuid`

17. 2019 年原始导出（逐字保留）

下面是完整源导出，作为历史证据逐字节保留。相对图片引用属于存档，因此有意不改写。

---
id: 1968
title: 'MetaTrader 4 + Python: ZeroMQ'
slug: 'metatrader-4-python-zeromq'
date: '2019-07-02T14:05:13'
modified: '2019-07-02T14:19:47'
status: 'publish'
link: 'https://blog.lazying.art/en/html/securities-forex/metatrader/1968/metatrader-4-python-zeromq.html'
author: 'Lachlan Chen'
categories:
  - 'MetaTrader'
---

Issues:

- *Microsoft Visual C++ 2015 Redistributable* need to be installed
- *Allow DLL imports *should be checked when an EA is loaded: Cannot call ‘libzmq.dll::zmq_ctx_new’, DLL is not allowed

![](images/image-1024x498.png)