第 4 章 · 01 MainEngine 的装配容器与委托式 API

第 4 章 · 01 MainEngine 的装配容器与委托式 API

本节摘要:本节精读 VeighNa 的装配中枢——MainEngine。它本身不实现任何交易逻辑,只持有四个 dict(gateways/engines/apps/exchanges),通过 add_gateway/add_app/add_engine 三件套动态装配。所有对外 API(connect/subscribe/send_order...)都是委托式透传:取 gateway → 写日志 → 调 gateway 同名方法。本节最精彩的是 init_engines 里的 Facade 技巧——OmsEngine 的 19 个查询/转换方法被一行行"提升"为 MainEngine 的属性,外部调用 main_engine.get_tick(...) 完全意识不到背后是 OmsEngine。读完本节,你理解了"核心 + 插件"架构怎么用 200 行搭起来。

内容来源:原项目源码 vnpy/trader/engine.py(BaseEngine L59-78、MainEngine L81-322),精读并套用体系化模板。

学习目标

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

1. 解释 BaseEngine 抽象基类三个引用字段与 close() 默认实现的意义。
2. 逐行读懂 MainEngine 的 __init__:四个 dict、os.chdir、init_engines。
3. 区分 add_engine/add_gateway/add_app 三件套各自的入参和返回。
4. 读懂 init_engines 的 Facade 提升——19 个方法为何要"挂"到 MainEngine 上。
5. 看清委托式交易 API 的统一三段式模式。
6. 澄清 add_gateway 不用 importlib,接收的是已导入的类对象。

一、BaseEngine 抽象基类

所有"功能引擎"(LogEngine/OmsEngine/EmailEngine/WechatEngine)的共同父类(engine.py:59-78):

59  class BaseEngine(ABC):
64      @abstractmethod
65      def __init__(
66          self,
67          main_engine: "MainEngine",
68          event_engine: EventEngine,
69          engine_name: str,
70      ) -> None:
71          self.main_engine: MainEngine = main_engine
72          self.event_engine: EventEngine = event_engine
73          self.engine_name: str = engine_name
74
76      def close(self) -> None:
77          return
​

三个引用字段,都是"反向指回宿主":

注意 __init__ 是 @abstractmethod,强制子类必须自己写构造器并 super().__init__(...) 把这三个引用挂上;而 close() 不是抽象方法,默认空实现,子引擎按需覆写(如 EmailEngine 要停线程,LogEngine 不需要)。

💡 核心心法:BaseEngine 注入两个引擎而非一个,体现了 VeighNa 的双总线设计——main_engine 是"组件总线"(查 gateway、写日志、发通知),event_engine 是"数据总线"(行情、订单、定时心跳)。子引擎同时挂在两条总线上,既能在组件之间互相调用,也能订阅异步事件。

二、MainEngine.init:四个 dict + 两条副作用

核心初始化(engine.py:86-100):

86      def __init__(self, event_engine: EventEngine | None = None) -> None:
88          if event_engine:
89              self.event_engine: EventEngine = event_engine
90          else:
91              self.event_engine = EventEngine()
92          self.event_engine.start()
93
94          self.gateways: dict[str, BaseGateway] = {}
95          self.engines: dict[str, BaseEngine] = {}
96          self.apps: dict[str, BaseApp] = {}
97          self.exchanges: list[Exchange] = []
98
99          os.chdir(TRADER_DIR)    # Change working directory
100         self.init_engines()     # Initialize function engines
​

逐行拆解:

1. event_engine: EventEngine | None = None:可以外部传一个已有的 EventEngine,不传就 new 一个。这是依赖注入模式——MainEngine 不强求自己创建事件引擎,允许策略层共享同一个事件总线。
2. self.event_engine.start():无论新建还是复用,MainEngine 都负责启动它。注意这里 start 会拉起事件主循环和定时器两条线程(第 2 章讲过)。如果传进来的是已经启动过的 EventEngine,Thread.start() 会抛 RuntimeError——所以实践中要么传未启动的,要么让 MainEngine 自己创建。
3. 四个容器:

4. os.chdir(TRADER_DIR):把工作目录切到 ~/.vntrader/。VeighNa 的配置文件、日志、SQLite 数据库都用相对路径写入(如 "vt_setting.json"、"database.db"),所以必须先 chdir 到统一目录,否则这些文件会散落到启动进程的 cwd 里。
5. init_engines():装配四大内置引擎,详见第四节。

三、三件套:add_engine / add_gateway / add_app

1. add_engine:最底层

engine.py:102-108:

102     def add_engine(self, engine_class: type[EngineType]) -> EngineType:
106         engine: EngineType = engine_class(self, self.event_engine)
107         self.engines[engine.engine_name] = engine
108         return engine
​

三步:实例化(注入 main_engine + event_engine)→ 用 engine.engine_name 作 key 存入 self.engines → 返回引擎实例。返回实例很重要——init_engines 里正是靠返回值拿到 OmsEngine 才能做 Facade 提升。

注意 engine_class 是类对象而非字符串,调用方负责先 import 类。返回类型用了泛型 EngineType,这样 add_engine(OmsEngine) 的返回值在类型检查器眼里就是 OmsEngine(而不是基类 BaseEngine),IDE 能补全 OmsEngine 的方法。

2. add_gateway:接收类 + 合并交易所

engine.py:110-126:

110     def add_gateway(self, gateway_class: type[BaseGateway], gateway_name: str = "") -> BaseGateway:
114         if not gateway_name:
115             gateway_name = gateway_class.default_name
117         gateway: BaseGateway = gateway_class(self.event_engine, gateway_name)
118         self.gateways[gateway_name] = gateway
120         for exchange in gateway.exchanges:
121             if exchange not in self.exchanges:
122                 self.exchanges.append(exchange)
125         return gateway
​

四个要点:

1. gateway_class 是已导入的类(不是字符串)。同一个底层接口类可以注册多个实例——比如跑两个 CTP 账户,就 add_gateway(CtpGateway, "CTP_A") 和 add_gateway(CtpGateway, "CTP_B"),靠 gateway_name 区分。
2. gateway_name 缺省走 default_name:类属性 CtpGateway.default_name = "CTP",这是常见用法。
3. gateway 构造只传 event_engine 和 gateway_name,不传 main_engine——gateway 是"数据生产者",只需要事件总线 push 行情/订单,不需要回头调 MainEngine。这与子引擎的注入签名明显不同。
4. 合并交易所:每个 gateway 类声明它支持的 exchanges(如 CTP 支持 SHFE/DCE/CZCE/CFFEX/INE),MainEngine 把所有 gateway 的交易所去重合并到 self.exchanges,供 UI 的合约查询用。

⚠️ 重要说明:add_gateway 不用 importlib。它接收的是已经 import 进来的类对象,而非字符串路径。VeighNa 里真正用 importlib 动态加载的地方在另外三处:database.py(按 SETTINGS 加载数据库驱动)、datafeed.py(加载数据源服务)、ui/mainwindow.py(扫描 gateways/apps 目录批量导入插件类)。也就是说——importlib 负责"发现",add_gateway 负责"登记",两步分离。这样 MainEngine 完全可以脱离 importlib 单测,只要传入已导入的类即可。

3. add_app:复用 add_engine

engine.py:128-136:

128     def add_app(self, app_class: type[BaseApp]) -> BaseEngine:
132         app: BaseApp = app_class()
133         self.apps[app.app_name] = app
135         engine: BaseEngine = self.add_engine(app.engine_class)
136         return engine
​

App(BaseApp 子类)是个"元对象",它本身不做事,只持有元信息——其中 engine_class 指向真正干活的引擎类。add_app 三步:实例化 app 存入 self.apps → 从 app 取出 engine_class → 委托给 add_engine 装配。

💡 核心心法:add_app 是 add_engine 的语法糖。区别只在 add_app 多登记一份 app 元信息(供 UI 列出已加载插件、查询其配置)。app 与 engine 的关系是"声明与实现"——CtaStrategyApp.app_name = "CtaStrategy",CtaStrategyApp.engine_class = CtaEngine。最终 engine 都进 self.engines,app 都进 self.apps,两条线分开记。

四、init_engines:装配四大内置引擎 + Facade 提升

本节最核心的方法(engine.py:138-166):

138     def init_engines(self) -> None:
142         self.add_engine(LogEngine)
143
144         oms_engine: OmsEngine = self.add_engine(OmsEngine)
145         self.get_tick: Callable = oms_engine.get_tick
146         self.get_order: Callable = oms_engine.get_order
147         self.get_trade: Callable = oms_engine.get_trade
148         self.get_position: Callable = oms_engine.get_position
149         self.get_account: Callable = oms_engine.get_account
150         self.get_contract: Callable = oms_engine.get_contract
151         self.get_quote: Callable = oms_engine.get_quote
152         self.get_all_ticks: Callable = oms_engine.get_all_ticks
153         self.get_all_orders: Callable = oms_engine.get_all_orders
154         self.get_all_trades: Callable = oms_engine.get_all_trades
155         self.get_all_positions: Callable = oms_engine.get_all_positions
156         self.get_all_accounts: Callable = oms_engine.get_all_accounts
157         self.get_all_contracts: Callable = oms_engine.get_all_contracts
158         self.get_all_quotes: Callable = oms_engine.get_all_quotes
159         self.get_all_active_orders: Callable = oms_engine.get_all_active_orders
160         self.get_all_active_quotes: Callable = oms_engine.get_all_active_quotes
161         self.update_order_request: Callable = oms_engine.update_order_request
162         self.convert_order_request: Callable = oms_engine.convert_order_request
163         self.get_converter: Callable = oms_engine.get_converter
164
165         self.add_engine(EmailEngine)
166         self.add_engine(WechatEngine)
​

装配顺序固定:LogEngine → OmsEngine → EmailEngine → WechatEngine。

• 先装 LogEngine 是因为后续装配过程若出问题,需要它把日志写出来。
• OmsEngine 第二 是因为它要订阅事件,装配越早缓存越全。
• EmailEngine/WechatEngine 最后,它们是"被动通知渠道",与核心交易流无关。

Facade 技巧:19 个方法提升

最精彩的是 145-163 行——把 OmsEngine 的 19 个方法直接赋值给 MainEngine 实例属性:

self.get_tick = oms_engine.get_tick
​

这一行干了一件看似魔法的事:main_engine.get_tick 不是 MainEngine 类里定义的方法,而是一个指向 oms_engine.get_tick 函数对象的实例属性。Python 中,self.foo = some_func 之后,self.foo(...) 等价于 some_func(...)(注意没有 self 自动绑定,因为 some_func 已经是绑定方法了)。

效果是:外部代码看不到 OmsEngine 的存在。它写:

main_engine = MainEngine()
tick = main_engine.get_tick("rb2501.SHFE")           # 背后是 oms_engine.get_tick
orders = main_engine.get_all_active_orders()          # 背后是 oms_engine.get_all_active_orders
​

而不是:

oms = main_engine.get_engine("oms")
tick = oms.get_tick("rb2501.SHFE")
​

这就是 Facade(外观)模式——把子系统的复杂接口"扁平化"到顶层入口。19 个方法分三类:

💡 核心心法:为什么不用继承(class MainEngine(OmsEngine, ...))?因为 Python 多继承的 MRO 容易冲突,且 MainEngine 与 OmsEngine 是组合关系而非is-a关系。Facade 技巧用赋值实现"组合 + 接口扁平化",既保持了组件解耦(可以单独测 OmsEngine),又给了外部一个干净的入口。代价是 MainEngine 类的 IDE 自动补全看不出这 19 个方法(因为它们是运行时挂上去的),VeighNa 用类型注解 Callable[[str], TickData | None] 部分补偿了可读性。

五、write_log 与 send_notification

1. write_log:不直接写,只发事件

engine.py:168-173:

168     def write_log(self, msg: str, source: str = "MainEngine") -> None:
172         log: LogData = LogData(msg=msg, gateway_name=source)
173         event: Event = Event(EVENT_LOG, log)
174         self.event_engine.put(event)
​

注意:write_log 本身不写日志文件,也不调 loguru。它只是构造一个 LogData,包装成 EVENT_LOG 事件,put 到事件总线。真正消费这个事件、调 loguru 落盘的是 LogEngine(第 03 节详讲)。

这种"只发不写"的设计带来三个好处:

• 统一入口:整个 VeighNa 只有 EVENT_LOG 一种日志事件,谁都能发(MainEngine/gateway/engine/策略),消费方也只有 LogEngine 一处。
• 异步:put 立即返回,不阻塞调用线程(下单线程绝不被磁盘 IO 卡住)。
• 可关停:LogEngine 启动时读 SETTINGS["log.active"],False 时它收到事件也不落盘——只需改配置,不用改业务代码。

source 默认 "MainEngine",但子引擎/gateway 调用时传入自己的名字(如 self.main_engine.write_log(msg, "EmailEngine")),最终 loguru 会用这个 gateway_name 做 bind,日志里能区分来源。

2. send_notification:双渠道推送

engine.py:176-187:

176     def send_notification(self, content: str, subject: str | None = None) -> None:
180         if subject is None:
181             subject = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
183         email_engine: EmailEngine = cast(EmailEngine, self.get_engine("email"))
184         email_engine.send_email(subject, content)
185
186         wechat_engine: WechatEngine = cast(WechatEngine, self.get_engine("wechat"))
187         wechat_engine.send_wechat(f"{subject}\n{content}")
​

subject 缺省用当前时间(发邮件/微信总得有个标题)。然后从 self.engines 用名字取出 EmailEngine 和 WechatEngine,两个渠道并行投递——注意这里的 send_email/send_wechat 都是"入队即返回"的异步方法(详第 03 节),所以这个调用几乎零开销。

cast(EmailEngine, ...) 是 typing 的纯类型提示函数,运行时无副作用,纯粹帮类型检查器确认 get_engine 返回的 BaseEngine 真的是 EmailEngine。

六、委托式交易 API:统一三段式

engine.py:234-308 一共 7 个方法(connect/subscribe/send_order/cancel_order/send_quote/cancel_quote/query_history),结构高度一致。看最典型的 send_order(engine.py:254-264):

254     def send_order(self, req: OrderRequest, gateway_name: str) -> str:
258         gateway: BaseGateway | None = self.get_gateway(gateway_name)
259         if gateway:
260             self.write_log(_("委托下单 -> {}：{}").format(gateway_name, req))
261             return gateway.send_order(req)
262         else:
263             return ""
​

统一三段式:

1. get_gateway(gateway_name):按名字从 self.gateways 取,取不到返回 None(顺便 write_log 报错)。
2. write_log:写一条审计日志(委托下单/撤单/订阅是关键操作,必须留痕)。
3. 透传给 gateway 的同名方法:返回值原样返回。

7 个方法只有返回值的细节差异:

⚠️ 重要说明:send_order 失败返回 "",这是 VeighNa 的失败约定。策略层拿到空 vt_orderid 就知道下单没成功(可能是 gateway 名字写错,也可能是 gateway 内部失败)。send_quote(报价下单,做市商用)同理返回 vt_quoteid 或 ""。query_history(查 K 线)失败返回空列表。这三种返回约定贯穿整个 gateway 体系,后续 BaseGateway 的实现都遵守。

注意 _("委托下单 -> {}：{}")——_() 是国际化函数(第 03 章讲过 vnpy.trader.locale),把字符串标记为可翻译。这里中文是默认文案,实际可被替换为英文。{} 是占位符,format 时填入 gateway_name 和 req。注意第二个冒号是中文全角"："——VeighNa 在国际化字符串里用了中文标点。

七、get_xxx 与 close

几个简单的查询方法(engine.py:189-232):

• get_gateway(name):取 gateway,找不到 write_log 报错。
• get_engine(name):取 engine,同上。
• get_default_setting(name):取某 gateway 的默认配置(供 UI 渲染登录窗口)。
• get_all_gateway_names()/get_all_apps()/get_all_exchanges():返回各 dict/list 的浅拷贝(用 list(...) 包一层,防止外部改 dict 影响 MainEngine)。

close 生命周期(engine.py:310-322):

310     def close(self) -> None:
316         self.event_engine.stop()            # 先停事件引擎,避免再来新定时事件
318         for engine in self.engines.values():
319             engine.close()                  # 各引擎收尾(EmailEngine 停线程等)
321         for gateway in self.gateways.values():
322             gateway.close()                 # 各 gateway 关连接
​

顺序重要:先停事件引擎(不再有新 EVENT_TIMER/事件入队),再 close 各引擎(EmailEngine/WechatEngine 的 worker 线程此时 join 才不会卡),最后关 gateway。如果反过来,worker 线程在 join 时可能还在 write_log 试图 put 事件,而事件引擎已经停了——逻辑混乱。

本节要点回顾

1. BaseEngine 三引用:main_engine(组件总线)、event_engine(数据总线)、engine_name(dict key),close() 默认空实现。
2. init 四容器:gateways/engines/apps/exchanges;外加 os.chdir(TRADER_DIR) 切到统一目录、init_engines() 装配内置引擎。
3. 三件套区别:add_engine(最底层,注入两个引擎)→ add_gateway(接收已导入类,合并交易所)→ add_app(app 是元对象,内部委托 add_engine)。
4. Facade 提升:init_engines 把 OmsEngine 的 19 个方法逐行赋值给 MainEngine 实例属性,外部无感调用。组合 + 接口扁平化,比多继承更干净。
5. write_log 只发不写:put 一个 EVENT_LOG 事件,真正落盘的是 LogEngine;send_notification 双渠道(EmailEngine + WechatEngine)异步投递。
6. 委托式 API 三段式:取 gateway → write_log → 透传同名方法。send_order 失败返回 "",是 VeighNa 的失败约定。
7. add_gateway 不用 importlib:接收已导入类对象;importlib 实际在 database.py/datafeed.py/ui/mainwindow.py 负责"发现"插件类。

下一节,我们钻进 MainEngine 的"大脑"——OmsEngine:9 个缓存字典 + 2 个 active 子集怎么管理订单全生命周期,以及它如何懒初始化 OffsetConverter,为第 7 章的开平转换打地基。