金融科技【免费下载链接】rqalphaA extendable, replaceable Python algorithmic backtest trading framework supporting multiple securities项目地址https://gitcode.com/gh_mirrors/rq/rqalpha点击查看免费下载RQAlpha 是一个可扩展、可替换的 Python 量化回测与交易框架其日志系统默认采用logbook实现。本文聚焦如何在 RQAlpha 中通过编写 Mod 为 logger 添加自定义 handler将策略运行期间的日志含用户日志与用户系统日志实时转发到任意目标文件、数据库、消息队列等并从源码层面剖析user_log、user_system_log与日志处理器的工作机理帮助读者掌握一套可复用的策略日志收集方案。读完本文你将能够理解 RQAlpha 的日志体系与三类核心 logger 的职责实现一个自定义LogHandler并将其挂载到 Mod 生命周期中结合Environment.calendar_dt为每条日志打上回测时间戳从而把日志收集能力无缝嵌入自己的量化工作流。RQAlpha 的日志体系概述RQAlpha 在 rqalpha/utils/logger.py 中基于 logbook 构建了完整的日志基础设施核心内容包括默认日志模块采用logbook文档中声明为默认日志模块并在模块加载时调用logbook.set_datetime_format(local)将日志时间格式统一为本地时间同时将 WARNING 级别的名称 patch 为WARN。三类核心 logger定义于 rqalpha/utils/logger.pyuser_log用户策略代码如init、handle_bar中打印的日志产生的日志user_system_log面向用户展示的系统日志例如 API 参数校验告警、g/context持久化失败提示等见 rqalpha/apis/api_base.py、rqalpha/core/global_var.pysystem_logRQAlpha 内部运行日志例如 Mod 加载与启动过程见 rqalpha/mod/init.py。默认输出init_logger()将三个 logger 的 handlers 均设置为StderrHandler(bubbleTrue)即默认向标准错误输出。文档强调开发者可以通过在 Mod 中为 logger 添加 handler 来实现自定义的日志收集。这与 logbook 的 Handler 机制一致——每个 logger 维护一组 handler日志记录会依次经过这些 handler 处理format、emit从而可以实现转发、持久化等自定义行为。值得留意的是user_log与user_system_log同时被注册进user_log_group一个logbook.LoggerGroup并通过user_log_processor处理器在每次记录时把Environment中的calendar_dt写入record.time见 rqalpha/utils/logger.py。这意味着日志记录天然携带了回测的日历时间信息为后续按时间序列收集日志提供了基础。前置知识Mod 的生命周期与加载约定日志收集 Mod 本质上是一个标准 RQAlpha Mod因此需要先了解 Mod 的两个关键约定必须实现load_mod()函数返回一个继承自AbstractMod的实例见 rqalpha/mod/init.pystart_up(env, mod_config)在系统启动时被调用tear_down(code, exceptionNone)在系统退出前被调用接口定义见 rqalpha/interface.py。在start_up中挂载 handler正是文档给出的推荐时机。RQAlpha 在加载 Mod 时按config.mod中启用的模块依次调用load_mod()、start_up()系统退出时按逆序调用tear_down()见 rqalpha/mod/init.py。若 Mod 带自定义配置如目标地址、日志级别等可在包内定义__config__字典RQAlpha 会将其与用户配置合并后以mod_config形式传入start_up见 rqalpha/mod/init.py。自定义日志收集 Handler 的实现文档给出了一个完整的LogHandler实现其核心思路是继承 logbook 的Handler并混入StringFormatterHandlerMixin在emit阶段将格式化后的日志消息连同Environment.calendar_dt与日志级别一起回调给外部函数从而将日志“送出”RQAlpha。from logbook.handlers import Handler, StringFormatterHandlerMixin from logbook.base import NOTSET from rqalpha.environment import Environment class LogHandler(Handler, StringFormatterHandlerMixin): def __init__(self, send_log_handler, levelNOTSET, format_stringNone, filterNone, bubbleFalse): Handler.__init__(self, level, filter, bubble) StringFormatterHandlerMixin.__init__(self, format_string) self.send_log_handler send_log_handler def _write(self, level_name, item): dt Environment.get_instance().calendar_dt self.send_log_handler(dt, item, level_name) def emit(self, record): msg self.format(record) self.lock.acquire() try: self._write(record.level_name, msg) finally: self.lock.release()各参数与方法的职责如下组成说明Handler.__init__(level, filter, bubble)继承 logbook 基础 Handlerlevel控制该 handler 接收的最低日志级别默认NOTSET表示接收全部filter用于按记录过滤bubble表示处理完本条记录后是否继续向父级/其他 handler 冒泡StringFormatterHandlerMixin.__init__(format_string)提供字符串格式化能力format_string缺省时使用 logbook 默认格式_write(level_name, item)获取当前回测日历时间Environment.get_instance().calendar_dt连同日志文本与级别名一起回调send_log_handleremit(record)先format(record)得到格式化文本在加锁保护下调用_write保证多线程场景下输出不交错其中Environment.get_instance()是 RQAlpha 的单例访问入口见 rqalpha/environment.pycalendar_dt在环境初始化时被设为config.base.start_date的零点并在回测推进过程中不断更新因此这里取到的是“当前正在回测的自然日时间”非常适合日志按时间归档。编写日志收集 Mod文档给出了对应的 Mod 骨架初始化LogHandler并挂载到user_log与user_system_log的 handlers 列表同时实现_send_log将日志送往目的地。from rqalpha.interface import AbstractMod from rqalpha.utils.logger import user_system_log, user_log class CustomLogHandlerMod(AbstractMod): def _send_log(self, dt, text, log_tag): # TODO pass def start_up(self, env, mod_config): user_log.handlers.append(LogHandler(self._send_log, bubbleTrue)) user_system_log.handlers.append(LogHandler(self._send_log, bubbleTrue)) def tear_down(self, code, exceptionNone): pass def load_mod(): return CustomLogHandlerMod()代码要点bubbleTrue是关键默认情况下 logbook 的 handler 处理完记录后若不设置 bubble会中断记录传播设置为True表示当前 handler 处理完毕后记录继续冒泡从而不影响 RQAlpha 原有的标准错误输出日志收集与默认输出并行共存挂载目标文档明确要求同时挂载user_log策略日志与user_system_log系统给用户的提示日志这样策略内print/logger.info与框架层面的告警都能被统一收集_send_log(dt, text, log_tag)三个参数分别对应回测时间、格式化日志文本、级别名如INFO、WARN、ERROR在此处实现写入文件、上报消息队列等业务逻辑即可包结构按 Mod 发布约定该文件应置于rqalpha_mod_xxx命名的包中并声明load_mod()可参考仓库内的 rqalpha/examples/extend_api/rqalpha_mod_extend_api_demo.py 中__config__与load_mod()的写法为 Mod 增加可配置项。实战增强按天落盘并携带级别与时间在上述骨架基础上一个可直接运行的“按交易日落盘”实现如下落盘逻辑置于_send_log中import os from datetime import datetime from logbook.handlers import Handler, StringFormatterHandlerMixin from logbook.base import NOTSET from rqalpha.interface import AbstractMod from rqalpha.environment import Environment from rqalpha.utils.logger import user_log, user_system_log class LogHandler(Handler, StringFormatterHandlerMixin): def __init__(self, send_log_handler, levelNOTSET, format_stringNone, filterNone, bubbleFalse): Handler.__init__(self, level, filter, bubble) StringFormatterHandlerMixin.__init__(self, format_string) self.send_log_handler send_log_handler def _write(self, level_name, item): dt Environment.get_instance().calendar_dt self.send_log_handler(dt, item, level_name) def emit(self, record): msg self.format(record) self.lock.acquire() try: self._write(record.level_name, msg) finally: self.lock.release() class FileLogMod(AbstractMod): def __init__(self): self._log_dir None def _send_log(self, dt, text, log_tag): if not self._log_dir: return # 按自然日生成日志文件避免单个文件无限膨胀 filename os.path.join(self._log_dir, {}.log.format(dt.strftime(%Y%m%d))) with open(filename, a, encodingutf-8) as f: f.write([{}] [{}] {}\n.format(dt.strftime(%Y-%m-%d %H:%M:%S.%f), log_tag, text)) def start_up(self, env, mod_config): self._log_dir mod_config.log_dir os.makedirs(self._log_dir, exist_okTrue) user_log.handlers.append(LogHandler(self._send_log, bubbleTrue)) user_system_log.handlers.append(LogHandler(self._send_log, bubbleTrue)) def tear_down(self, code, exceptionNone): pass def load_mod(): return FileLogMod()若希望日志目录可配置可在包内补充__config__ { log_dir: ./logs, }随后通过rqalpha mod enable mod名启用Mod 管理与启停命令的实现可参考 rqalpha/cmds/mod.py。日志流向的源码级解读为了更准确地理解上述方案可以对照 RQAlpha 源码确认日志的产生与流转路径策略日志的产生RQAlpha 在运行时将用户代码中的print重定向为user_print后者内部调用user_log.info(message)见 rqalpha/utils/logger.py因此用户打印的内容都会进入user_log的 handler 链系统提示日志的产生框架内部大量使用user_system_log.warn(...)输出对用户可见的告警例如参数被弃用时rqalpha/apis/api_base.py、g对象无法 pickle 时rqalpha/core/global_var.py、策略before_trading使用了 deprecated 参数时rqalpha/core/strategy.py等。将这些提示一并收集有助于完整还原回测过程中的每一次告警handler 链的执行logbook 的Logger.handle(record)会依次调用其 handlers 的emit自定义 handler 通过bubbleTrue保持记录继续传播因此默认的 stderr 输出与自定义收集互不干扰时间信息user_log_group的user_log_processor在记录产生时即写入Environment.calendar_dtrqalpha/utils/logger.py自定义 handler 中再次读取同一时间戳保证与回测节奏严格一致。此外仓库还提供了另一个可借鉴的 handler 范式rqalpha/utils/log_capture.py中的CaptureHandler将所有记录缓存到内存列表LogCapture上下文管理器可临时替换某个 logger 的 handlers 并在退出时恢复rqalpha/utils/log_capture.py。这一模式常用于测试场景下的日志断言也证明了“为 logger 增删 handler”是 RQAlpha 日志体系的标准扩展点。注意事项与最佳实践不要移除默认 handler追加自定义 handler 时应保留原有 handler或确保新 handlerbubbleTrue避免日志只在自定义通道输出、控制台无任何反馈影响调试体验级别控制LogHandler的level参数可设为logbook.INFO、logbook.WARNING等按需过滤低频告警与高频调试信息防止日志文件被刷爆线程安全文档示例在emit中使用self.lock加锁多线程/多账户场景下应保留该写法在start_up中完成挂载start_up在回测正式开始前被调用此时挂载可保证从第一条策略日志起即可被收集tear_down中可执行 flush、关闭连接等收尾工作按天切分文件结合calendar_dt的日期部分生成文件名既符合回测语义又便于后续按日检索是文档示例的自然延伸。小结本文围绕 RQAlpha 官方文档《收集策略日志》展开完整实现了基于 logbook 的自定义日志收集方案先厘清user_log/user_system_log与 Handler 的关系再实现LogHandler与日志收集 Mod最后结合 rqalpha/utils/logger.py、rqalpha/mod/init.py 等源码确认了日志的产生与传播路径。开发者只需将_send_log替换为写入文件、上报消息队列或接入日志平台的具体实现即可在不影响 RQAlpha 默认输出的前提下将回测期间的策略日志与系统提示完整、按时间有序地收集起来为策略迭代、异常排查与量化研究提供可靠的日志支撑。赞分享金融科技【免费下载链接】rqalphaA extendable, replaceable Python algorithmic backtest trading framework supporting multiple securities项目地址https://gitcode.com/gh_mirrors/rq/rqalpha点击查看免费下载相关推荐Aptos 集群日志收集实战基于 Helm 与 Vector DaemonSet 的 Kubernetes 日志管道搭建指南Aptos 集群日志收集实战基于 Helm 与 Vector DaemonSet 的 Kubernetes 日志管道搭建指南 导读 本文面向在 Kuberne区块链Web3Kubernetes Handbook 应用日志收集实战基于 Filebeat Sidecar 与自有 ELK 集群的轻量级日志采集方案Kubernetes Handbook 应用日志收集实战基于 Filebeat Sidecar 与自有 ELK 集群的轻量级日志采集方案 在 Kubernet教程云原生容器编排Rust 固定投影还要手写 unsafe 吗pin-project 用一行属性宏生成全部样板Rust 固定投影还要手写 unsafe 吗pin project 用一行属性宏生成全部样板 给自引用 Future 手写投影时unsafe 块、泛型透传和网页爬虫工作流自动化上一篇RustFS 放置与修复不变式对象到集哈希规则、Quorum 边界与 Scanner/Heal 准入契约下一篇CopilotKit v2 × Express用单路由端点Single Route在 Node 侧部署 Agent 运行时创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考