手把手带你编写miniQMT实盘量化执行程序(2) | 主程序入口
由bqft6vju创建，最终由bqft6vju 被浏览 10 用户
大家好，我是策略老李。五一小长假即将收尾，不知大家是否已经按照上期教程搭建好了自己的本地代码库呢？今天老李将为大家奉上《手把手编写miniQMT实盘量化执行程序》的第2篇-主程序入口。废话不多说直接上代码：
main.py
"""
主程序入口
"""
from quant_trade_platform.config.config import config
from quant_trade_platform.utils.logger import sys_logger
mode = config.get('mode')
if __name__ == "__main__":
if mode == 'livetest':
from main_livetest import main
sys_logger.info("量化交易平台启动, 当前模式为 livetest")
main()
elif mode == 'functest':
from main_functest import main
sys_logger.info("量化交易平台启动, 当前模式为 functest")
main()
elif mode == 'backtest':
from main_backtest import main
sys_logger.info("量化交易平台启动, 当前模式为 backtest")
main()
elif mode == 'optimize':
from main_optimize import main
sys_logger.info("量化交易平台启动, 当前模式为 optimize")
main()
else:
from main_live import main
sys_logger.info("量化交易平台启动, 当前模式为 live")
main()
main.py根据config配置中的"mode"参数，自动选择对应的功能模式：
- live：实盘交易模式，真枪实弹，直连券商交易API，每次操作真实影响账户资金。
- optimize：参数优化模式，通过网格搜索寻找最优参数组合。
- backtest：历史回测模式，使用实盘量化框架完成策略回测验证。
- functest：功能测试模式，验证实盘量化框架各个功能组件功能。
- livetest：交易测试模式，验证miniQMT订单接口的功能逻辑。
这几个功能组件并不是老李在一开始就想好要做这几个功能的，而是在实盘执行程序开发过程中不断被BUG折磨，不断被策略捶打而逐步开发出来的。比如这个backtest模式，我做回测使用的是Jupyter Notebook，基本上20行代码就可以初步完成一个策略的回测评估，但是放到实盘平台里面执行的时候，由于策略代码不一致，导致我不能正确评估策略是否正确执行，因此我开发这个历史回测模块，目的是在完全不修改实盘策略代码的前提下完成实盘交易与历史回测。再比如为了摸透miniQMT的订单接口而开发了的这个livetest模式，就是想看看miniQMT对订单处理时间，订单下单，订单撤单，订单回调函数等等的处理逻辑，并在无厘头大规模随机下单的情况下，实盘执行程序能否通过压力测试。
main_live.py
"""
主程序入口
"""
import sys
import ctypes
import os
import signal
from typing import NoReturn
from datetime import datetime
from quant_trade_platform.config.config import config
from quant_trade_platform.live.livesystem import LiveSystem
from quant_trade_platform.utils.logger import sys_logger,close_logger
logger = sys_logger.getChild('Main')
# 全局退出标志
_exit_flag = False
# 定义Windows API常量
ES_CONTINUOUS = 0x80000000
ES_SYSTEM_REQUIRED = 0x00000001
ES_DISPLAY_REQUIRED = 0x00000002
def prevent_sleep():
"""启用防休眠模式"""
logger.info("启用防休眠模式")
ctypes.windll.kernel32.SetThreadExecutionState(
ES_CONTINUOUS | ES_SYSTEM_REQUIRED | ES_DISPLAY_REQUIRED
)
def allow_sleep():
"""恢复系统默认休眠策略"""
logger.info("恢复系统默认休眠策略")
ctypes.windll.kernel32.SetThreadExecutionState(ES_CONTINUOUS)
def handle_signal(signum, frame) -> None:
"""处理系统信号实现优雅退出
Args:
signum: 信号编号 (e.g. SIGINT=2, SIGTERM=15)
frame: 当前堆栈帧对象
"""
global _exit_flag
logger.warning("接收到系统信号 %d，启动优雅退出流程...", signum)
_exit_flag = True
def main() -> NoReturn:
"""主程序入口函数
实现功能：
1. 系统信号注册
2. 交易系统初始化
3. 主循环运行
4. 异常安全处理
"""
try:
start_time = datetime.now()
# 注册信号处理
signal.signal(signal.SIGINT, handle_signal) # Ctrl+C
signal.signal(signal.SIGTERM, handle_signal) # kill命令
# 系统启动信息记录
logger.info("=" * 60)
logger.info("量化交易平台启动")
logger.info("启动时间: %s", datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
logger.info("Python 版本: %s", sys.version.replace('\n', ''))
logger.info("当前工作目录: %s", os.getcwd())
logger.info("QMT工作目录: %s", config.get('xt.plugin_path'))
logger.info("=" * 60)
# 初始化交易系统
with LiveSystem() as system:
logger.info("交易系统初始化完成")
try:
# 启用防休眠模式
prevent_sleep()
# 设置终端标题
system.set_terminal_title(f"QMT LiveSystem - {config.get('account.account_id')}")
# 系统预热（连接交易所、加载策略等）
system.start()
# 主循环
logger.info("进入主运行循环")
while not _exit_flag:
try:
system.run()
except KeyboardInterrupt:
logger.warning("检测到键盘中断信号")
break
except Exception as e:
logger.critical("主循环运行异常: %s", str(e), exc_info=True)
raise
except Exception as e:
logger.critical("交易系统运行时异常: %s", str(e), exc_info=True)
raise # 传递异常到外层处理
logger.info("交易系统已安全关闭")
except Exception as e:
logger.critical("系统级异常导致终止: %s", str(e), exc_info=True)
sys.exit(255)
finally:
# 恢复系统默认休眠策略
allow_sleep()
logger.info("=" * 60)
logger.info("系统退出时间: %s", datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
logger.info("运行时长: %.2f秒", (datetime.now() - start_time).total_seconds())
logger.info("=" * 60)
close_logger()
if __name__ == "__main__":
main()
sys.exit(0)
main_live.py代码解读：
1. 防休眠逻辑（类比手机常亮模式）
ctypes.windll.kernel32.SetThreadExecutionState(...)
这段代码就像给电脑打了"强心针"，让实盘执行程序不休眠。你能想到当我某天晚上运行起实盘执行程序，第2天早上满怀期待的想看看程序执行结果的时候却发现程序卡住了，而当我一动鼠标他又开始继续执行时的那个让人崩溃的场景么，这个功能就是解决这个问题的！
2. 安全退出机制
def handle_signal(): global _exit_flag # 举起"正在降落"的指示灯
这个问题也纠结了我很久，在spyder IDE执行程序的时候使用ctrl+c终止程序，程序可以完美的终止，而在我使用windows PowerShell真正运行实盘执行程序的时候，在ctrl+c后PowerShell会弹出一堆dll错误，导致程序没办法在ctrl+c之后保存程序最后执行的快照文件，OK，现在这个解决方案也送给大家。
3. 交易系统核心
with LiveSystem() as system: # 启动交易引擎 system.set_terminal_title() # 给引擎贴标签 system.start() # 预热引擎 while not _exit_flag: # 持续巡航 system.run() # 执行交易策略
with语句确保引擎故障时会自动安全关闭，下一篇文章会详细分析LiveSystem的代码实现。
4. 异常处理
try...except...finally # 多层安全防护
整个代码被多层异常处理包裹，在我最开始写代码的时候也没有写这么多try，我看到很多开源代码也没有写try，而现在你们在看我后面的代码时，会发现无处不在的try except块，这是为什么呢？都是被BUG逼出来。那些无处不在的try-except不是懦弱的盾牌，而是身经百战后的智慧结晶。每次优雅的异常处理，都是在为系统注入一剂疫苗。当我们终于能在凌晨三点安睡时，就会明白这些防御代码不是累赘，而是写给未来自己最温柔的情书——毕竟，能伤害你代码的，永远是你未曾预料的那次意外。
【特别承诺】：
老李在这里向各位读者保证，每一个关注了公众号(策略老李)的同学，无需修改任何内容即可正确运行与老李实盘一模一样的实盘量化执行程序，无任何套路，无需任何打赏。对代码内容有任何不清楚的都可以公众号私信，或评论区留言。
【往期回顾】：
手把手带你编写miniQMT实盘量化执行程序(1) | 完整源码结构篇
脑子进水了？为什么我要独立开发一个miniQMT实盘执行程序
策略实盘上线公告（miniQMT）｜致谢合作伙伴与开发历程回顾
版权声明：
文中miniQMT实盘量化执行程序著作权归策略老李所有，本程序仅供个人学习研究使用，允许在注明作者的前提下进行技术交流与非商业性传播，使用者应对其使用行为自行承担风险。