程序员量化交易实战 39:把失败检查转换成处理动作
原创 · 约 10 分钟阅读 · 阅读 --
Last updated on

程序员量化交易实战 39:把失败检查转换成处理动作

作者: Alex Xiang


程序员量化交易实战 39:把失败检查转换成处理动作

古董级程序员,大厂出来后一直在创业公司,现在仍活跃在一线做 AI 相关的开发。这个专栏会把一个 A 股量化平台从 0 到 1 拆开写:数据、策略、回测、模拟盘、提醒和生产化,尽量用真实代码和真实运行结果说话。更完整的更新也会同步到微信公众号「字与码」。

第 37 篇已经能把失败检查项保存在 DailyRunResult.failed_checks 里。

第 39 篇继续把失败变成动作。系统不应该只说“失败了”,还应该告诉操作者下一步该等、该补数据,还是该看健康报告。

ZiCode 工程师处理模拟盘失败动作

失败动作不是自动修复

这里的动作不是直接帮你修复生产问题。

第一版只做路由:把失败检查项映射成处理建议和严重级别。

失败检查动作严重级别
run_windowwait_next_windowinfo
history_readyinspect_archivewarning
data_gapsrepair_market_datablocker
run_healthinspect_run_healthblocker
未知检查项manual_reviewwarning

这张表很小,但它让日报、命令行和后续自动化都能说同一种语言。

失败动作对象

第 39 章新增 app/failure_policy.py

@dataclass(frozen=True)
class FailureAction:
    check_name: str
    action: str
    severity: str

check_name 保留原始检查项,action 给出处理方向,severity 用于决定汇总状态。

映射策略

策略表直接写成常量。

_ACTION_BY_CHECK = {
    "run_window": ("wait_next_window", "info"),
    "history_ready": ("inspect_archive", "warning"),
    "data_gaps": ("repair_market_data", "blocker"),
    "run_health": ("inspect_run_health", "blocker"),
}

未知检查项默认进入人工复核。

action, severity = _ACTION_BY_CHECK.get(
    check_name,
    ("manual_review", "warning"),
)

这样后续新增检查项时,即使忘了更新策略,也不会让系统直接崩掉。

动作汇总

汇总规则也保持简单:

if not actions:
    return "no_action_required"
if any(action.severity == "blocker" for action in actions):
    return "blocker"
return "warning"

只要有 blocker,整次失败动作就是 blocker。没有动作说明不需要处理。

当前联动运行结果

paper-run-plan 里故意让价格源缺少 600519.SH,因此运行结果的失败检查项是 data_gaps

uv run python -m scripts.chapter_examples paper-run-plan

paper-run-plan 命令生成的失败动作映射

系统把 data_gaps 映射成 repair_market_data,严重级别是 blocker。这比只输出 “blocked” 更有用,因为操作者能直接知道下一步是补行情数据,而不是去看运行窗口或通知通道。

测试动作映射

运行本章测试:

uv run pytest tests/test_failure_policy.py tests/test_run_result.py

关键断言:

assert [(item.check_name, item.action, item.severity) for item in actions] == [
    ("run_window", "wait_next_window", "info"),
    ("data_gaps", "repair_market_data", "blocker"),
    ("run_health", "inspect_run_health", "blocker"),
]
assert failure_action_summary(actions) == "blocker"

本章更新与代码仓库

本章更新内容:

  • 新增 app/failure_policy.py
  • 实现 FailureAction
  • 把失败检查项映射成处理动作。
  • 支持未知检查项的人工复核兜底。
  • 实现 failure_action_summary()
  • 增加 paper-run-plan 联动示例,展示 data_gapsrepair_market_data 的映射。
  • 补充 blocked 状态和处理动作之间的差异。
  • 新增 tests/test_failure_policy.py,覆盖已知检查、未知检查和空动作。

代码仓库:

https://github.com/ax2/zi-quant-platform

本章代码:

git clone https://github.com/ax2/zi-quant-platform.git
cd zi-quant-platform
git checkout chapter-39
uv sync --extra dev
uv run pytest tests/test_failure_policy.py tests/test_run_result.py

第 39 章提交为 c05f77f,tag 为 chapter-39

本篇小结

失败动作让 blocked 状态更可操作。

第 39 篇把失败检查项转换成处理建议和严重级别。下一篇会把请求、检查清单、运行结果和失败动作合成一个每日运行计划,完成这一组的收口。