OrderFilledEvent 接入订单追踪系统

现状诊断

当前 fill 处理路径

user 频道 data.fills  →  _publish_user_events()  →  OrderFilledEvent  →  🕳️ 无消费者（数据丢失）
userFills 频道        →  handle_message()         →  _on_user_fill()   →  直接累计（旧直连路径）

两条路径状态不一致：

• user 频道的 fills：_publish_user_events 已发布 OrderFilledEvent（优先级 CRITICAL，metadata 含正确 fill_id），但 EventBus 上零订阅者，成交数据完全丢失，不进入订单追踪。
• userFills 频道：_on_user_fill 已正确解析 data["fills"] 嵌套结构（含向后兼容），通过 _make_fill_id 去重后直接调用 _accumulate_fill，走的是旧直连路径而非 EventBus。

根本问题

唯一遗留问题：OrderFilledEvent 是"已搭好发射台但从未点火"的组件——有定义、有发布、有 CRITICAL 优先级，但没有任何消费者。这不是事件多余，而是订阅方缺位。

两路来源中，userFills 频道目前能正常工作（通过旧直连路径），user 频道的成交数据完全丢失。

已完成的修复

EventBus 特性确认

经代码核查，EventBus.publish 为同步阻塞调用，直接遍历 handler 列表执行，不涉及队列或线程池。因此：

• 消息顺序与 WebSocket 到达顺序完全一致，无时序风险
• publish(event) 在行为上等价于直接方法调用，无额外开销
• 使用 EventBus 的目的是解耦（基础设施层不直接持有业务层引用），而非异步
• 同步调用意味着 WebSocket 消息处理线程会阻塞至 handler 执行完毕；正常情况下延迟极低（微秒级），但需注意避免在 handler 内执行耗时操作

设计目标

1. 修复 user 频道 fills 丢失：让 WebSocketOrderManager 成为 OrderFilledEvent 的消费者。
2. 统一成交入口：两路 fill 来源（user 频道 + userFills 频道）均通过 OrderFilledEvent 进入订单追踪，_fill_ids 去重保证不重复累计。
3. 保持行为不变：订单状态机、加权均价计算、grace 超时、HTTP 兜底逻辑全部不动。
4. 架构方向正确：基础设施层（enhanced_ws_manager）只发事件，业务层（WebSocketOrderManager）自行订阅，避免跨层耦合；后续 TradeLogger、RiskManager 等新消费者只需 subscribe，无需改动现有代码。

架构变化

变更前

realtime_kline_service_base.on_message
    ├── channel == "orderUpdates"  →  mgr.handle_message()  →  _on_order_update()
    └── channel == "userFills"     →  mgr.handle_message()  →  _on_user_fill()

enhanced_ws_manager._publish_user_events (user 频道 fills)
    └── publish(OrderFilledEvent)  →  【无消费者，数据丢失】

变更后

realtime_kline_service_base.on_message
    ├── channel == "orderUpdates"  →  mgr.handle_message()  →  _on_order_update()  ← 不变
    └── channel == "userFills"     →  【不再路由到 handle_message】

enhanced_ws_manager（两路来源，均通过 _make_fill_id 发布带前缀 fill_id 的事件）
    ├── _publish_user_events (user 频道 fills)   →  publish(OrderFilledEvent + fill_id)  ─┐
    └── _cache_latest_data  (userFills 频道)     →  publish(OrderFilledEvent + fill_id)  ─┤
                                                                                           ↓
                                                                               EventBus（同步）
                                                                                           ↓
                                                               WebSocketOrderManager._on_order_filled_event()
                                                                   └── _fill_ids 去重 → _accumulate_fill()

实现步骤

执行顺序：前提（提取共享函数）→ 步骤一（修复发布端）→ 步骤二（接入消费端）→ 步骤三（移除旧路由）。
步骤一和步骤二完成、验证通过（见步骤三前提标准）后，再执行步骤三；避免在新路径未验证前就切断旧路径。

✅ 前提：提取共享函数 _make_fill_id（已完成）

这是本次变更的基础。发布端两处与消费端必须调用同一函数，否则同一笔成交在两路产生不同 key，去重失效。

当前状态：已完成。_make_fill_id 现为 src/events/trading_events.py 的模块级函数，原 websocket_order_manager._fill_key 已删除，两处调用均已迁移至 _make_fill_id。

函数定义（位于 src/events/trading_events.py 顶部）：

import hashlib

def _make_fill_id(fill: dict) -> str:
    """
    生成 fill 的唯一去重 key。
    - 有 tid（含 tid=0）：返回 "tid:{tid}"
    - 无 tid（即 None）：返回 "hash:{md5(oid:px:sz:time)}"

    注意：
    - 使用 `tid is not None` 判断，确保 tid=0 视为有效值。
    - hash 路径各字段直接使用 fill.get(key)（无缺省值），缺失时值为 None，
      f-string 产生 "None:..."，两路来源必须完全一致才能正确去重。
    - px/sz 使用原始字符串，不做 float 转换。
    """
    tid = fill.get("tid")
    if tid is not None:
        return f"tid:{tid}"
    raw = f"{fill.get('oid')}:{fill.get('px')}:{fill.get('sz')}:{fill.get('time')}"
    return f"hash:{hashlib.md5(raw.encode()).hexdigest()}"

两侧 import 方式：

from src.events.trading_events import _make_fill_id

步骤一：修复两路发布端，统一写入 fill_id

事件定义说明：OrderFilledEvent 继承自 Event，已有 metadata 字段，无需改 OrderFilledEvent 类本身。

文件：src/utils/websocket/enhanced_ws_manager.py

✅ 修改点 A：_publish_user_events（user 频道）—— 已完成

当前代码已正确调用 _make_fill_id 并写入 metadata：

fill_id = _make_fill_id(fill)
event = OrderFilledEvent(
    timestamp=datetime.now(),
    source="websocket",
    order_id=str(fill.get("oid", "")),
    client_order_id=fill.get("cloid"),
    symbol=fill.get("coin", ""),
    side=fill.get("side", ""),
    filled_qty=float(fill.get("sz", 0)),
    filled_price=float(fill.get("px", 0)),
    fee=float(fill.get("fee", 0)),
    closed_pnl=float(fill.get("closedPnl", 0)) if "closedPnl" in fill else None,
    metadata={"fill_id": fill_id},
)
self._event_bus.publish(event)

⬜ 修改点 B：_cache_latest_data（userFills 频道）—— 待完成

在现有 channel == "user" 分支旁新增 elif channel == "userFills" 分支。

注意数据结构（测试网实测验证）：Hyperliquid userFills 推送的顶层 data 为 WsUserFills { isSnapshot?, user, fills: WsFill[] }，fills 数组在 data["fills"] 下，不是 data 本身。快照消息含 isSnapshot: true，实时成交消息同样使用此结构（isSnapshot 缺省或为 false）。两种情况均需取 data["fills"]。

elif channel == "userFills":
    raw = msg.get("data", {})
    # data 结构为 WsUserFills { user, fills: WsFill[] }，取 fills 字段
    fills = raw.get("fills", []) if isinstance(raw, dict) else raw
    for fill in fills:
        oid = fill.get("oid")
        if oid is None:
            continue
        fill_id = _make_fill_id(fill)
        event = OrderFilledEvent(
            timestamp=datetime.now(),
            source="websocket",
            order_id=str(oid),
            client_order_id=fill.get("cloid"),
            symbol=fill.get("coin", ""),
            side=fill.get("side", ""),
            filled_qty=float(fill.get("sz", 0)),
            filled_price=float(fill.get("px", 0)),
            fee=float(fill.get("fee", 0)),
            closed_pnl=float(fill.get("closedPnl", 0)) if "closedPnl" in fill else None,
            metadata={"fill_id": fill_id},
        )
        self._event_bus.publish(event)

快照消息处理说明：userFills 首次连接时会推送一条快照（isSnapshot: true），包含该账户的历史成交数据。

• 已结束订单（不在 _tracking 中，或 status ≠ PENDING）：消费端 _on_order_filled_event 中 tracking is None / tracking.status != PENDING 分支静默跳过，无副作用。
• 重连时仍在 PENDING 的订单：快照中的历史 fill 会被正常累计并触发 _resolve，这是预期行为——补偿重连期间遗漏的成交数据。

两种情况均无需区分处理，现有 _tracking.get(oid) 防护已覆盖全部场景。

⬜ 步骤二：WebSocketOrderManager 订阅 OrderFilledEvent（待完成）

文件 A：src/trading/executor.py（约第 143 行）

# 变更前
self._ws_order_manager = WebSocketOrderManager(executor=self)

# 变更后
self._ws_order_manager = WebSocketOrderManager(executor=self, event_bus=self._event_bus)

文件 B：src/trading/websocket_order_manager.py

构造函数：增加 event_bus 参数，在末尾完成订阅。event_bus=None 仅供单元测试使用；生产路径由 Executor 强制传入，缺省时打印 WARNING，便于排查问题：

def __init__(self, executor, event_bus=None):
    self._executor = executor
    self._tracking: dict[int, OrderTracking] = {}
    self._lock = threading.Lock()
    self._http_busy: set[int] = set()

    self._event_bus = event_bus
    if event_bus is not None:
        event_bus.subscribe(OrderFilledEvent, self._on_order_filled_event)
    else:
        logger.warning(
            "WebSocketOrderManager: event_bus 未注入，user 频道 fills 将丢失，"
            "仅测试环境允许此配置"
        )

新增 handler：_on_order_filled_event

发布端已保证 metadata["fill_id"] 必然存在且格式正确。消费端直接读取，缺失时记录 WARNING 并跳过（不使用可能产生不一致 key 的备用路径）：

def _on_order_filled_event(self, event: OrderFilledEvent) -> None:
    oid = _safe_int_oid(event.order_id)
    if oid is None:
        return

    fill_px = event.filled_price
    fill_sz = event.filled_qty
    if fill_px <= 0 or fill_sz <= 0:
        return

    # fill_id 由发布端通过 _make_fill_id 写入，此处强依赖
    fill_id: str = event.metadata.get("fill_id") if event.metadata else None
    if not fill_id:
        logger.warning(
            "OrderFilledEvent 缺少 fill_id，发布端未正确调用 _make_fill_id，"
            "oid=%s，跳过此事件以避免去重 key 不一致", oid
        )
        return

    should_resolve = False
    resolve_tracking = None

    with self._lock:
        tracking = self._tracking.get(oid)
        if tracking is None or tracking.status != OrderStatus.PENDING:
            return
        if fill_id in tracking._fill_ids:
            logger.debug("跳过重复 fill: oid=%s fill_id=%s", oid, fill_id)
            return

        tracking._fill_ids.add(fill_id)
        self._accumulate_fill(tracking, fill_px, fill_sz)
        tracking.has_fill_price = True
        tracking.fill_count += 1
        logger.info(
            "EventBus fill 累计: %s oid=%s 本笔 px=%s sz=%s | "
            "累计 avg_px=%.6g total_sz=%.6g fills=%s fill_id=%s",
            event.symbol, oid, fill_px, fill_sz,
            tracking.avg_price, tracking.filled_size,
            tracking.fill_count, fill_id,
        )

        if tracking._ws_status is not None:
            should_resolve = True
            resolve_tracking = tracking

    # _resolve 在锁外调用。两路 fill 事件（user 频道与 userFills 频道）可能并发进入此处：
    # - 两次 _resolve 调用持有相同的 resolve_tracking 引用
    # - _resolve 内部通过 identity check（current is not tracking）保证幂等：
    #   第一次调用成功 pop _tracking[oid]，第二次因 current is None 直接返回 False
    # - 这是安全的幂等行为，两次并发最多只有一次成功解析
    if should_resolve:
        self._resolve(oid, resolve_tracking)

为何不保留备用 key：消费端自行构造备用 key 时，event.timestamp（datetime.now() 生成）与原始 fill["time"]（交易所毫秒时间戳）在值和精度上均不同，无法与发布端的 _make_fill_id 对齐。正确做法是在发布端保证 fill_id 必填，而非在消费端兜底。

取消订阅：在 manager 的 close() 或 reset() 方法中调用：

if self._event_bus is not None:
    self._event_bus.unsubscribe(OrderFilledEvent, self._on_order_filled_event)

防止 manager 重建时 handler 残留或多次订阅。

⬜ 步骤三：移除 userFills 旧路由（待完成，依赖步骤二验证通过）

前提（量化验证标准）：在测试网环境完成以下验证后，再执行此步骤：

1. 下 10 笔以上限价单，追踪成功率 100%（PENDING → FILLED 状态正确流转）
2. 手动触发 user 频道与 userFills 频道同时到达同一笔 fill，加权均价与 REST 查询结果误差 < 0.01%
3. 查看日志，无 "跳过重复 fill" 以外的 WARNING/ERROR
4. 以上验证在主网灰度（单账户小仓位）复测通过

文件：src/services/realtime_kline_service_base.py（约第 639 行）

# 变更前
if channel in ("orderUpdates", "userFills"):
    mgr.handle_message(msg)

# 变更后
if channel == "orderUpdates":          # userFills 已由 EventBus 接管，不再直接路由
    mgr.handle_message(msg)

缓冲区回放（重连恢复逻辑）：同步修改，缓冲消息回放时同样只对 orderUpdates 调用 handle_message，不对 userFills 回放。

取舍说明：移除旧路由后，重连期间的历史 userFills 消息不会被补偿回放。这是有意的取舍：

• 断线期间的成交补偿应通过 REST API 查询订单状态，而非回放 WebSocket 原始消息（现有 HTTP 兜底逻辑不变）。
• 若未来需要精确补偿，应在 WebSocketReconnectedEvent 的 handler 中主动拉取 fills，而不是回放原始 userFills。

步骤三验证通过后，必须删除以下死代码（步骤三完成后这两处将永远不被调用）：

1. websocket_order_manager.py 中的 _on_user_fill 方法（整个方法体）
2. handle_message 中对 channel == "userFills" 的分支

保留死代码会误导后续维护者以为该路径仍然生效，必须在步骤三验收时一并清理。

fill_id 生成规则（发布端与消费端必须一致）

关键约束：

• 发布端与消费端统一调用 _make_fill_id(fill)（定义于 src/events/trading_events.py），禁止使用裸字段读取方式——后者在 tid 存在时可能返回裸值（如 12345），缺少 tid: 前缀，导致去重失效。
• tid=0 是有效值：使用 if tid is not None 判断（而非 if tid），确保 tid=0 的 fill 返回 "tid:0"。若用 if tid: 则 tid=0 被误判为无 tid，走 hash 路径，产生不同 key，去重失效。
• hash 路径缺省值：各字段使用 fill.get(key)（无缺省值），字段缺失时值为 None，f-string 产生 "None:..."。若使用 fill.get(key, "") 则缺失字段产生 ":..."，两路 hash 不同，去重失效。
• px、sz 使用原始字符串（不做 float 转换）。
• 消费端强依赖 metadata["fill_id"]，不自行构造备用 key（原因：event.timestamp 与 fill["time"] 在值和精度上不同，无法对齐）。

联网文档与实测：响应体结构与 key 统一

官方数据结构（Hyperliquid）

• 文档：WebSocket Subscriptions 中定义：
  • userFills → WsUserFills { user, fills: WsFill[] }（fills 在嵌套字段中，不是 data 本身）
  • userEvents（或部分实现中的 user）→ WsUserEvent 可包含 { "fills": WsFill[] }
• WsFill 字段（与去重相关）：oid, tid（唯一成交 ID）, px, sz, time, coin, side, fee, closedPnl 等。两路 fill 来源均使用同一 WsFill 结构，故可用同一套 fill_id 规则。

实测与 key 协调

1. 在测试网观察真实响应：运行脚本捕获 WebSocket 消息与 fill 结构：

   TRADING_NETWORK=testnet python scripts/capture_fill_responses_testnet.py
   

   脚本会订阅 orderUpdates / userFills / user，在测试网下一笔小限价单（或手动下单），将收到的消息写入 scripts/out/fill_responses_<ts>.json，并在控制台打印各 channel 下 fill 的字段集合与样本，便于确认 tid、time 等在两路中的存在与格式。

2. key/id 统一约定：

   • 唯一标识：以 fill_id 作为去重键，与现有 OrderTracking._fill_ids 一致。
   • 生成方式：统一调用 _make_fill_id(fill)：有 tid 时用 tid:{tid}，无 tid 时用 hash:{md5(oid:px:sz:time)}；发布端与消费端共用同一函数，保证 user 频道与 userFills 频道对同一笔成交得到相同 fill_id，从而只累计一次。

3. 实测结果（两次测试网捕获，数据完全一致）：

   项目	结论
   实测时间	2026-02-22 12:27 / 2026-02-22 12:45（两次）
   userFills 顶层结构	data = {"isSnapshot": bool, "user": "0x...", "fills": [WsFill, ...]}
   fills 解析方式	必须取 data["fills"]，不可直接用 data
   WsFill 字段	coin, px(str), sz(str), side(str), time(int ms), startPosition, dir, closedPnl, hash, oid(int), crossed, fee(str), tid(int), feeToken, twapId
   tid 存在性	30+ 条 fill 均有 tid（整数），hash 备用路径极少触发
   px / sz 类型	字符串（如 "0.01804", "14288.0"），_make_fill_id 直接使用原始字符串，不做 float 转换
   oid 类型	整数（如 48913699327），OrderFilledEvent 存为 str(oid)
   side 值	"A"（Ask/卖出）或 "B"（Bid/买入），直接透传，无需转换
   orderUpdates 结构	data = [{"order": {...}, "status": "...", "statusTimestamp": int}]，直接数组

   关键验证：raw.get("fills", []) if isinstance(raw, dict) else raw 解析方式与实测数据结构完全匹配，可正确提取 fills 数组。

测试矩阵

变更文件汇总