第 4 章 · 02 OmsEngine 订单管理与状态缓存

第 4 章 · 02 OmsEngine 订单管理与状态缓存

本节摘要:本节精读 OmsEngine——VeighNa 的订单管理系统(OMS)。它本身不主动调任何交易接口,只做一件事:订阅 7 类事件(tick/order/trade/position/account/contract/quote),把每条事件的数据缓存进字典,并维护两个 active_xxx 子集跟踪"还活着的"订单和报价。它还管理一个 offset_converters 字典——按 gateway 名字懒初始化 OffsetConverter,为第 7 章的开平转换(上期所平今/平昨)提前铺路。本节最值得揣摩的是 process_order_event 的"三段式"(缓存 + active 子集 + OffsetConverter 更新),这套模式贯穿整个引擎。

内容来源:原项目源码 vnpy/trader/engine.py(OmsEngine 类 L360-587),精读并套用体系化模板。

学习目标

阅读完本节,你应当能够:

1. 列举 OmsEngine 的 9 个缓存字典 + 2 个 active 子集 + 1 个 converter 字典各自存什么。
2. 说清 register_event 订阅的 7 类事件分别对应哪个 process_xxx_event。
3. 读懂 process_order_event 的三段式(缓存 + active 子集 + OffsetConverter)。
4. 解释为什么 process_contract_event 要在这里懒初始化 OffsetConverter。
5. 区分 get_xxx(单查)与 get_all_xxx(全查)的实现差异。
6. 理解 update_order_request / convert_order_request / get_converter 三个方法如何为策略层服务。

一、OmsEngine 的定位

OmsEngine 是个纯被动的事件订阅者 + 字典缓存器。它不发起任何交易动作,只观察事件总线上发生了什么,把"最新状态"记下来供别人查:

为什么要把状态集中缓存?因为 VeighNa 是多组件架构——gateway 只管发事件,UI 要画表格、策略要读持仓、CTA 引擎要查合约,不能每个组件都各自维护一份状态(容易不一致)。OmsEngine 是单一数据源(single source of truth),谁要状态都来问它。

二、init:9 字典 + 2 子集 + 1 converter

engine.py:365-382:

365     def __init__(self, main_engine: MainEngine, event_engine: EventEngine) -> None:
367         super().__init__(main_engine, event_engine, "oms")
368
369         self.ticks: dict[str, TickData] = {}
370         self.orders: dict[str, OrderData] = {}
371         self.trades: dict[str, TradeData] = {}
372         self.positions: dict[str, PositionData] = {}
373         self.accounts: dict[str, AccountData] = {}
374         self.contracts: dict[str, ContractData] = {}
375         self.quotes: dict[str, QuoteData] = {}
376
377         self.active_orders: dict[str, OrderData] = {}
378         self.active_quotes: dict[str, QuoteData] = {}
379
380         self.offset_converters: dict[str, OffsetConverter] = {}
381
382         self.register_event()
​

按职责分三组:

1. 七大业务缓存(全量)

key 都是数据对象的"全局限定 ID"(vt_symbol/vt_orderid/vt_tradeid/vt_positionid/vt_accountid/vt_quoteid):

全量缓存(从不删除),只覆盖——每个新事件来了,用同样的 key 写入,自动覆盖旧值。所以字典里存的是"每个对象的最新快照"。

2. 两个 active 子集

为什么要单独维护"活跃子集"?因为策略/UI 最常用的查询是"现在我有几笔挂单未成交"——如果每次都从 orders 全量扫一遍调 is_active(),O(n) 太慢。维护一个 active 字典,查询变成 O(1) 的 list(active_orders.values())。

3. offset_converters

dict[str, OffsetConverter]:key 是 gateway_name,value 是该 gateway 专属的 OffsetConverter。注意这里不预先 new——只在 process_contract_event 时按需懒初始化(详第五节)。这为第 7 章的开平转换打地基。

三、register_event:订阅 7 类事件

engine.py:384-392:

384     def register_event(self) -> None:
386         self.event_engine.register(EVENT_TICK, self.process_tick_event)
387         self.event_engine.register(EVENT_ORDER, self.process_order_event)
388         self.event_engine.register(EVENT_TRADE, self.process_trade_event)
389         self.event_engine.register(EVENT_POSITION, self.process_position_event)
390         self.event_engine.register(EVENT_ACCOUNT, self.process_account_event)
391         self.event_engine.register(EVENT_CONTRACT, self.process_contract_event)
392         self.event_engine.register(EVENT_QUOTE, self.process_quote_event)
​

7 行,一一对应。注意只订阅"业务事件",不订阅 EVENT_LOG(日志归 LogEngine 管)、不订阅 EVENT_TIMER(OMS 不需要定时任务)。

7 个事件常量都是字符串前缀(如 EVENT_TICK = "eTick.",第 3 章讲过),EventEngine 的 register 机制是"前缀匹配"——所有以 "eTick." 开头的事件类型(如 "eTick.CTP"、"eTick.STOCK")都会路由到同一个 process_tick_event。

四、process_xxx 三种模式

7 个处理函数按结构分三种模式。

模式一:简单缓存型(process_tick/account/contract)

最简单,只写一个字典。process_tick_event(engine.py:394-397):

394     def process_tick_event(self, event: Event) -> None:
396         tick: TickData = event.data
397         self.ticks[tick.vt_symbol] = tick
​

两行——从事件取出 data,用 vt_symbol 作 key 写入。tick 更新极频繁(每秒成百上千次),所以处理函数必须极简——O(1) 字典赋值,绝不做任何额外计算。

process_account_event 和 process_contract_event 前半部分结构相同,只是 key 不同(vt_accountid/vt_symbol)。注意 process_contract_event 还多做了一件事——懒初始化 OffsetConverter,详第五节。

模式二:三段式(process_order/trade/position)

以最典型的 process_order_event(engine.py:399-414)为例:

399     def process_order_event(self, event: Event) -> None:
401         order: OrderData = event.data
402         self.orders[order.vt_orderid] = order           # 第①段:全量缓存
403
404         if order.is_active():                            # 第②段:维护 active 子集
405             self.active_orders[order.vt_orderid] = order
407         elif order.vt_orderid in self.active_orders:
408             self.active_orders.pop(order.vt_orderid)
409
411         converter = self.offset_converters.get(order.gateway_name, None)  # 第③段:喂 OffsetConverter
412         if converter:
413             converter.update_order(order)
​

三段式:

1. 全量缓存:self.orders[vt_orderid] = order。无脑覆盖,字典里永远是最新的订单状态。
2. active 子集维护:订单有 is_active() 方法(状态为 NOTTRADED/PARTTRADED 时为真)。活跃→加入子集;不活跃且原本在子集里→pop 出去。注意是 elif,不是 else——避免重复 pop。
3. OffsetConverter 更新:按 gateway_name 取对应的 converter,把这条订单喂给它。converter 用订单的 Offset(开/平/平今/平昨)信息推断持仓方向,第 7 章详讲。

process_trade_event 和 process_position_event 结构相同,只是没有"active 子集"那段——trade 是已成交事实(无所谓"活跃"),position 是持仓快照(也无所谓"活跃")。但它们都做第③段:喂 OffsetConverter。

💡 核心心法:三段式是"事件 → 多视角状态"的标准模式。一段缓存是"原始真相",二段 active 子集是"派生视图"(查询优化),三段 converter 是"另一套派生数据"(用于开平转换逻辑)。同一份事件,被三种消费者各取所需。这种"事件驱动 + 多重派生"的设计,使得加新功能时不必改 gateway,只需在 OmsEngine 多挂一个订阅。

模式三:订单/报价的对称实现

process_quote_event(engine.py:450-460)与 process_order_event 几乎一字不差,只是把 order 换成 quote:

450     def process_quote_event(self, event: Event) -> None:
452         quote: QuoteData = event.data
453         self.quotes[quote.vt_quoteid] = quote
455         if quote.is_active():
456             self.active_quotes[quote.vt_quoteid] = quote
458         elif quote.vt_quoteid in self.active_quotes:
459             self.active_quotes.pop(quote.vt_quoteid)
​

报价(做市商的双边报价)与订单是对称概念:报价也会"挂起→成交/撤掉",所以也维护 active_quotes 子集。唯一区别:quote 不喂 OffsetConverter——因为报价的开平逻辑与普通订单不同,不需要转换。

五、process_contract_event 的懒初始化

engine.py:441-448:

441     def process_contract_event(self, event: Event) -> None:
443         contract: ContractData = event.data
444         self.contracts[contract.vt_symbol] = contract
446         if contract.gateway_name not in self.offset_converters:
447             self.offset_converters[contract.gateway_name] = OffsetConverter(self)
​

第一行同模式一——缓存。第二行是关键:每个 gateway 第一次发合约事件时,为它 new 一个 OffsetConverter,存入 offset_converters 字典,key 是 gateway_name。

为什么在这里懒初始化?三个原因:

1. 合约事件先于订单/持仓事件:gateway 登录后会先批量推送合约(几千个),之后才有订单/持仓。所以"合约到了"是初始化 converter 的最早时机。
2. OffsetConverter 需要合约信息:OffsetConverter(self) 把整个 MainEngine 传进去,它内部会调 main_engine.get_contract(...) 查合约的开平模式(开平/平今/平昨),需要 contracts 字典已填充。所以必须先缓存合约,再 new converter——这正是"先 444 行再 446 行"的顺序。
3. 避免空对象:如果 gateway 没发任何订单/持仓,但策略要查 converter,直接 self.offset_converters.get(name) 返回 None——调用方知道"该 gateway 还没准备好"。

⚠️ 重要说明:OffsetConverter(self) 把 MainEngine 自己传进去——这看似循环引用(OmsEngine 持 main_engine,OffsetConverter 又持 main_engine),但 Python 的引用计数能处理(只要 MainEngine close 时所有子引擎释放,链就被打断)。这种"反向回调"模式让 OffsetConverter 能查到 OmsEngine 缓存的合约,而不用复制一份。

六、get_xxx 单查 + get_all_xxx 全查

查询方法分两组。

1. 单查(7 个)

get_tick 为例(engine.py:462-466):

462     def get_tick(self, vt_symbol: str) -> TickData | None:
466         return self.ticks.get(vt_symbol, None)
​

7 个单查方法(get_tick/get_order/get_trade/get_position/get_account/get_contract/get_quote)模板完全一致:return self.xxx.get(key, None)。找不到返回 None,不抛异常——策略层用 if tick := main_engine.get_tick(sym): 这种安全模式。

2. 全查(7 个全量 + 2 个 active)

get_all_ticks 为例(engine.py:504-508):

504     def get_all_ticks(self) -> list[TickData]:
508         return list(self.ticks.values())
​

注意 list(...) 包一层——返回字典 values 的浅拷贝列表。如果不包,返回的是 dict_values 视图对象,后续字典变动会影响视图,且视图不能索引。包成 list 既隔离了内部字典,又支持索引/迭代。

get_all_active_orders / get_all_active_quotes 两个针对 active 子集的方法实现完全相同,只是查的字典不同。这就是上一节 Facade 提升的 19 个方法——MainEngine 把它们挂成自己的属性,外部无感。

七、三个 OffsetConverter 相关方法

最后三个方法(engine.py:558-587)与开平转换直接相关。

1. update_order_request

558     def update_order_request(self, req: OrderRequest, vt_orderid: str, gateway_name: str) -> None:
562         converter = self.offset_converters.get(gateway_name, None)
563         if converter:
564             converter.update_order_request(req, vt_orderid)
​

订单真正下单时调用——记录"这条 OrderRequest 对应哪个 vt_orderid"。因为 OffsetConverter 推断持仓方向需要"原始请求意图"(用户想开多还是开空),而 gateway 推送回来的 OrderData 可能已经改过 Offset(如上期所会把"平"改成"平今")。所以两边都要记一份:OrderData 走 process_order_event,OrderRequest 走 update_order_request。

2. convert_order_request

566     def convert_order_request(
567         self, req: OrderRequest, gateway_name: str,
568         lock: bool, net: bool = False
569     ) -> list[OrderRequest]:
576         converter = self.offset_converters.get(gateway_name, None)
577         if not converter:
578             return [req]                              # 没有 converter,原样返回
579
580         reqs: list[OrderRequest] = converter.convert_order_request(req, lock, net)
581         return reqs
​

策略层下单前的请求转换。lock(锁仓转换)和 net(净持仓模式)两个开关决定转换算法——这是第 7 章的核心内容,本节只看接口形状。注意返回的是列表——因为一笔逻辑订单可能被拆成多笔实际订单(比如"锁仓"模式下,为了不开新仓,可能要先平旧仓再开反向)。

没有 converter 时返回 [req]——单元素列表,表示"不转换,原样下"。这种兜底让代码在 gateway 还没初始化 converter 时也不会崩。

3. get_converter

583     def get_converter(self, gateway_name: str) -> OffsetConverter | None:
587         return self.offset_converters.get(gateway_name, None)
​

直接返回 converter 对象本身(不是包装一层)。供高级策略或 UI 直接调 converter 的方法。找不到返回 None。

💡 核心心法:这三个方法体现了 OMS 的双重职责:既是缓存器(被动订阅事件),也是开平转换的中介(主动供策略查询)。OffsetConverter 是 VeighNa 处理国内期货"上期所平今/平昨"复杂规则的核心,但 OMS 本身不实现转换逻辑——它只持有 converter 字典并转发调用。真正的算法在 vnpy/trader/converter.py,第 7 章精讲。

本节要点回顾

1. 9 字典 + 2 子集 + 1 converter:七大业务全量缓存 + active_orders/active_quotes 派生子集 + offset_converters 按 gateway 懒初始化。
2. register_event 订阅 7 类事件:tick/order/trade/position/account/contract/quote,一一映射到 process_xxx。
3. process_xxx 三种模式:简单缓存型(tick/account)、三段式缓存+active+converter(order)、对称实现(quote 同 order)。
4. process_contract_event 懒初始化:每个 gateway 第一次发合约时 new OffsetConverter,先缓存合约再 new,因为 converter 要查合约的开平模式。
5. get_xxx / get_all_xxx:单查返回 None(不抛异常),全查用 list(...) 包浅拷贝。
6. 三个 converter 方法:update_order_request 记录原始请求,convert_order_request 转换并返回 list(可能拆单),get_converter 直接暴露对象。
7. 双重职责:OMS 既是缓存器也是开平转换中介,但转换算法在 OffsetConverter 里,本引擎只转发。

下一节,我们离开订单管理,看 VeighNa 的三大辅助引擎:LogEngine(把 EVENT_LOG 落盘到 loguru)、EmailEngine(懒启动的 SMTP 异步邮件)、WechatEngine(4.4 新增的节流合并微信推送,凭据持久化 + 会话过期自愈)。