Dial - 建立网络连接

统一的连接入口，根据地址协议选择连接实现。本次扩展 WebSocket（ws / wss）连接，保留已有 TCP / TLS 支持；数据库连接尚未实现。

WebSocket 连接由框架托管：后台线程负责收消息放入队列，断线后自动重连并恢复订阅，空闲时自动发心跳。策略只需在主循环里取消息。

语法

# WebSocket 连接
ws = Dial("wss://fstream.binance.com/ws")
ws = Dial("wss://fstream.binance.com/ws", timeout=10000, auto_reconnect=True)

# 设置连接超时、自动重连和消息队列上限
ws = Dial(
    "wss://fstream.binance.com/ws",
    timeout=10000,
    auto_reconnect=True,
    max_queue=10000,
    max_queue_bytes=16 * 1024 * 1024,
)

# TCP / TLS 连接
conn = Dial("example.com:80")
conn = Dial("tcp://example.com:80")
conn = Dial("tls://example.com:443")

参数

参数传递方式

支持 Dial(address)、Dial(address, timeout)，其余连接选项均使用关键字参数，例如 Dial(address, timeout=10000, auto_reconnect=True)。

地址仅描述连接目标，路径和 ?query 原样传给服务端，不用于设置本地连接参数。连接选项不接受整体配置字典，也不支持在地址后拼接配置后缀。

header 可直接传请求头字典或列表；payload 可直接传消息字符串、字节串、字典或列表，不需要将这些内容编码到连接地址中。代理地址中的用户名、密码仍需按 URL 规则编码特殊字符。

不支持的调用格式、未知 WebSocket 选项、无效数值、错误布尔值及尚未支持的协议都会记录错误并返回 None，不会静默忽略，也不会把数据库地址当作 TCP 连接。

时间参数统一使用毫秒，例如 timeout=50 表示 50 毫秒。read() / write() 分别是 recv() / send() 的别名；read(0) / recv_nowait() 不等待，send() / write() 返回是否发送成功的布尔值。

WebSocket 选项（仅 ws / wss 生效）

通过关键字参数传入，例如 Dial(url, timeout=10000, idle_timeout=30000)。

布尔参数接受 Python True / False 或字符串 "true" / "false"；时间须为有限非负数，interval 和 reconnect_max_delay 须大于 0；队列条数须为正整数，字节上限须为非负整数。

地址格式

数据库协议（例如 sqlite3://、mysql://、postgres://、clickhouse://）当前明确返回 None 并记录“不支持协议”。未安装数据库驱动，也未提供数据库连接的 exec() 等方法。

返回值

返回连接对象。建立连接失败时返回 None，策略应检查返回值。

WebSocket 连接对象

stats() 返回示例：

{
    "url": "wss://fstream.binance.com/ws",
    "connected": True,
    "closed": False,
    "queued": 0,              # 队列中未取的消息数
    "queued_bytes": 0,        # 排队消息的内容字节数（不含 Python 对象等额外内存）
    "dropped": 0,             # 因队列满被丢弃的消息数
    "reconnect_count": 1,
    "uptime": 3600.5,         # 本次连接已持续的秒数
    "last_recv_ago": 0.1,     # 距上次收到数据的秒数
    "last_error": "ConnectionError: 服务端关闭连接 (code=1000)"
}

TCP 连接对象

工作机制

超时与断线的区别

recv() 优先返回队列中已有的消息，即使连接正在重连或已经关闭，尚未取出的消息仍可读取。队列为空时，等待超时或连接永久关闭会返回 None。需要区分状态时看 ws.connected 和 ws.closed。正常使用时，网络断线不会使 recv() 抛出异常，也不会在每次读取时反复写错误日志。

自动重连

开启 auto_reconnect（默认）时，后台线程会在以下情况重连：

• 服务端关闭连接，或网络断开
• 超过 idle_timeout 收不到任何数据（连接假死）
• Binance 等交易所的 24 小时连接上限到期

重连后会自动恢复两类订阅：通过 subscribe() 保存的 stream，以及通过 send(..., persist=True) 保存的消息。先重放 stream 订阅（每批最多 100 个），再按保存顺序重发持久消息。直接用 send() 发送的订阅消息不会重发。

握手和订阅重放均完成后，connected 才变为 True，reconnect_count 才增加。重放发送失败会记录错误、关闭此次连接并继续重试。未设置 interval 时采用指数退避，加入随机抖动后的等待时间也不会超过 reconnect_max_delay；设置 interval 时使用固定间隔。发送成功不等于交易所接受订阅，策略仍需处理服务端的订阅确认或错误消息。

payload 与 persist=True 共用持久消息记录，在第一次握手完成后、启动读线程前发送；首次发送失败则 Dial() 返回 None。重连时仍按“stream 订阅、持久消息”的顺序重放，不提供需要等待服务端确认的多步认证流程。

重连期间的行为：

• 普通 send() 返回 False，消息不会缓存。
• subscribe()、send(..., persist=True) 即使因尚未连上而返回 False，仍会保存恢复记录；unsubscribe() 即使返回 False，也会移除对应恢复记录。
• 订阅变更与重放串行执行。如果重放正在发送，subscribe()、unsubscribe()、send(..., persist=True) 和 clear_persist() 可能等待本轮重放结束后再执行，以避免旧订阅覆盖退订操作。
• recv() 仍优先读取已有消息；队列为空时，可能等待至指定超时。需要立即返回时使用 recv(0) 或 recv_all()。

同一个 stream 只保存一份；持久消息按序列化后的内容去重，但每次显式调用 send() 仍会尝试发送。重放使用保存的原始内容，不会更新其中的时间戳、签名或令牌；需要重新生成这些字段的订阅应由策略管理，不能直接持久化过期请求。

unsubscribe() 只管理 subscribe() 保存的 stream；通过 persist=True 保存的自定义订阅应先清理持久记录，再按交易所协议发送退订消息。clear_persist() 不会撤销已发出的订阅或取消正在发送的重放。

调用 close() 后不会再次发布连接。若关闭时已有握手正在进行，该握手返回的 socket 会被关闭，不会使 connected 重新变为 True；已永久关闭的对象也不会接受新的恢复记录。

心跳

服务端发来的 ping 帧由框架自动回复 pong。连接空闲超过 ping_interval 时，框架会主动发送 ping 帧来保活，同时用它检测连接是否还活着。策略不需要自己维护协议层心跳。

如果交易所要求应用层心跳（例如 Gate 的 futures.ping JSON 消息），仍需策略按交易所文档自己发送。

不阻塞多连接

消息由后台线程接收，recv(timeout_ms) 只是从队列里取。主循环里同时处理多个连接时，建议使用 recv_all() 或 recv(0)：某个连接没有消息时立即返回，不会拖慢其他连接。

接收解压与队列限制

未设置 compress 时，文本和二进制消息仍分别返回 str 和 bytes。启用后，只对二进制数据消息进行指定格式的解压，再按 UTF-8 解码为 str；普通文本消息（如应用层心跳）保持原样，协议层 ping / pong 不受影响。这是应用消息解压，不是 permessage-deflate。

解压格式错误、数据截断或解压结果不是有效 UTF-8 时，进入断线处理流程并记录 last_error。请先确认交易所实际使用的压缩格式，避免持续重连。

max_queue 和 max_queue_bytes 同时生效：超过任一上限时丢弃最旧消息；如果单条消息自身超过 max_queue_bytes，则只丢弃该条，不清空已有队列。设置字节上限后，解压也在超过该上限时停止，避免先完整展开超大消息。所有丢弃都计入 stats()["dropped"] 并使用限频日志提示；queued_bytes 随读取同步减少。该上限不是整个连接的内存上限，底层 WebSocket 帧缓冲不计入。

示例

完整的双交易所、多币种盘口示例见 Binance / Gate 50 档监控策略。该策略用两个工作线程维护盘口，主线程刷新状态栏。汇总表每个币种一行，深度表每个档位一行，均并排显示 Binance 与 Gate 的数据；遇到丢消息或序号断档时重新同步，不会跳过增量消息继续展示错误盘口。

汇总中的中间价为 (买一价 + 卖一价) / 2，价差为 Binance 中间价 - Gate 中间价（USDT），价差率为 价差 / Gate 中间价 × 100%，保留六位小数。正值表示 Binance 更贵；任一侧盘口未就绪、过期或缺少买卖报价时，价差显示 --。中间价差仅供行情对比，不等于扣除手续费、滑点后的可成交收益。

示例1：订阅实时行情

import json

def main():
    ws = Dial("wss://fstream.binance.com/ws")
    if not ws:
        raise RuntimeError("WebSocket 连接失败")

    # subscribe() 的订阅在重连后会自动恢复
    ws.subscribe(["btcusdt@bookTicker", "ethusdt@bookTicker"])

    while True:
        msg = ws.recv(1000)
        if not msg:
            continue
        data = json.loads(msg)
        if data.get("e") == "bookTicker":
            Log(f"{data['s']} 买一 {data['b']} 卖一 {data['a']}")

示例2：同时处理多个交易所

import json

def main():
    binance = Dial("wss://fstream.binance.com/ws")
    gate = Dial("wss://fx-ws.gateio.ws/v4/ws/usdt")
    if not binance or not gate:
        raise RuntimeError("WebSocket 连接失败")

    binance.subscribe("btcusdt@bookTicker")
    # 非 Binance 格式的订阅用 persist=True，重连后自动重发
    gate.send({
        "time": Unix(),
        "channel": "futures.book_ticker",
        "event": "subscribe",
        "payload": ["BTC_USDT"],
    }, persist=True)

    while True:
        # recv_all() 不等待，一次取出所有已到达的消息
        for msg in binance.recv_all():
            handle_binance(json.loads(msg))
        for msg in gate.recv_all():
            handle_gate(json.loads(msg))
        Sleep(10)

示例3：查看连接状态

ws = Dial("wss://fstream.binance.com/ws")
if not ws:
    raise RuntimeError("WebSocket 连接失败")
ws.subscribe("btcusdt@bookTicker")
last_reconnects = 0

while True:
    msg = ws.recv(1000)

    if ws.closed:
        Log("连接已关闭")
        break

    if ws.reconnect_count != last_reconnects:
        last_reconnects = ws.reconnect_count
        Log(f"连接已恢复，累计重连 {last_reconnects} 次", ws.stats())

    if msg:
        ...

示例4：自己管理断线

ws = None

while True:
    if ws is None:
        ws = Dial("wss://fstream.binance.com/ws", auto_reconnect=False)
        if ws is None:
            Sleep(3000)
            continue
        ws.subscribe("btcusdt@bookTicker")

    msg = ws.recv(1000)
    if ws.closed:
        Log("连接断开:", ws.stats()["last_error"])
        ws.close()
        ws = None
        continue
    if msg:
        ...

示例5：TCP 连接

conn = Dial("tcp://example.com:80", 5000)

if conn:
    conn.write("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
    response = conn.read(4096)
    if response:
        Log(f"响应: {response}")
    conn.close()

示例6：代理、解压与初始订阅

以下为配置格式示例，地址、压缩方式和订阅内容应按目标服务实际要求替换：

ws = Dial(
    "wss://example.com/feed",
    timeout=10000,
    proxy="socks5h://127.0.0.1:1080",
    header={"X-Client": "strategy"},
    compress="gzip",
    mode="recv",
    auto_reconnect=True,
    interval=2000,
    max_queue=10000,
    max_queue_bytes=16 * 1024 * 1024,
    payload={"method": "SUBSCRIBE", "params": ["ticker"], "id": 1},
)
if not ws:
    raise RuntimeError("连接或初始订阅发送失败")

# 已有方法不变；不要重复发送 payload 中的订阅
for message in ws.recv_all():
    Log(message)
ws.close()

注意事项

1. 检查返回值：建立连接失败时返回 None。建立之后的断线由框架自动处理。

2. 证书校验：默认校验 SSL 证书。如果报“证书校验失败”，先检查系统时间和 CA 证书；确认地址可信后，才考虑传 verify_ssl=False。

3. 消息积压：主循环处理太慢时，队列会达到条数或字节上限，stats()["dropped"] 会增加，日志里也会有提示。高频行情建议用 recv_all() 批量处理。

4. 订阅恢复：只有 subscribe()、send(..., persist=True) 和 payload 保存的订阅会在重连后恢复；是否被交易所接受仍需检查服务端响应。

5. 应用层心跳：协议层的 ping/pong 由框架处理；交易所自定义的 JSON 心跳仍需策略发送。

6. 关闭连接：不再使用时调用 close()，关闭后不会再重连。

7. 协议选择：

   • 实时行情：WebSocket
   • 简单请求：HttpQuery
   • 自定义协议：TCP / TLS

相关 API

• HttpQuery - HTTP 请求
• Sleep - 休眠

开发维护

Dial 协议入口 负责地址解析、参数校验和协议分发，WebSocket 实现 只负责连接生命周期与消息收发。未来接入数据库时，为对应协议新增独立处理器及连接类型，再加入入口的协议映射；不要把数据库执行方法添加到 WebSocket 对象，也不要给未实现的协议提供成功形状的占位对象。

源码 SDK 与 内嵌 SDK 必须保持一致，依赖清单也需同步。开发运行优先使用本地源码，打包脚本会用源码覆盖内嵌副本。SOCKS5 支持使用 python-socks，已列入 SDK 依赖。

在仓库根目录运行以下回归测试，覆盖两份实现的配置解析、订阅恢复、解压、队列限制、并发竞态及副本一致性，并使用本地 SOCKS5 / WebSocket 服务验证实际收发，无需连接真实交易所：

python3 -B -m unittest discover -s test -p test_dial.py