🛡️ RiskGuard

给你的交易系统,配一个 7×24、永不情绪化的风控官。

"Rule No.1:永远不要亏钱。Rule No.2:永远不要忘记第一条。" —— 沃伦·巴菲特

简体中文 · English

量化交易的五层积木——数据 → 研究 → 回测 → 风控 → 执行——数据有 OpenBB、研究有 Qlib、回测有 backtesting.py/vectorbt、执行有 Alpaca,唯独风控层没有一个成熟的开源标准件。RiskGuard 就是来补这块空白的。

风控的本质不是技术,而是提前写死的纪律。RiskGuard 把那几条能挡住绝大多数爆仓惨剧的规则,做成一个核心零依赖、可组合的库,让系统自动执行——因为等你真的亏钱上头那一刻,意志力是这世界上最靠不住的东西。

• 看它拦下一次爆仓(30 秒)
• 为什么不自己写几个 if 判断就好?
• 一套风控,这些能力
• 60 秒上手
• 动态下注
• 实时哨兵(kill-switch)
• 防篡改审计
• 状态持久化
• 每日体检 + 压力测试
• 接进你的回测框架
• 三档预设 + 命令行
• 架构
• 适合谁 / 设计原则 / 边界 / 路线图

🔥 看它拦下一次爆仓(30 秒)

同一个"满仓做多"信号,喂给同一段 −45% 的崩盘行情。一个不设风控,一个套上 RiskGuard(单笔 10% 上限 + 15% 回撤熔断)。看最大回撤的差距:

同一段崩盘,风控把最大回撤从 45% 压到 5.6%。
纪律不是让你多赚,是让你别在最坏的一天把本金亏光。

这段数字是 examples/06_with_vs_without.py 的真实输出,是"仓位封顶"这条纪律的机械结果、可一键复现——不是收益承诺。git clone 下来自己跑一遍就知道。

但纪律不是免费的

同一档配置,换成 +200% 的单边牛市,风控账户只吃到裸奔涨幅的 10%——120,000 vs 300,000。仓位上限截断的是分布的两端:既拦暴跌,也拦暴涨。这是真实的保费,不是隐藏条款(见 examples/09)。值不值,取决于你更怕错过暴涨,还是更怕活不到下一次机会。

🤔 为什么不自己写几个 if 判断就好?

你当然可以写。但风控真正难的不是"能不能判断",而是分析质量和决策纪律——尤其在你亏钱上头、最不该动手的那一刻。

这不是"能不能判断"的问题,而是"纪律能不能被系统强制执行"的问题。 RiskGuard 把这七件事全部焊死进代码。

🧱 一套风控,这些能力

两层严格分离:Sizer 只决定"下多大注",Rule 决定"能不能下、要不要缩、该不该全停"。AI 可以做研究、写代码、挑毛病,但每一笔真实指令都必须先过这几条写死的风控规则。

⚡ 60 秒上手

⚠️ 尚未发布到 PyPI——pip install riskguard 现在装不上,请用下面的 git 安装方式。 发布到 PyPI 后这里会第一时间更新(进度见路线图)。

pip install git+https://github.com/SilentFleetKK/riskguard.git              # 核心,零依赖
pip install "riskguard[alpaca] @ git+https://github.com/SilentFleetKK/riskguard.git"  # 需要 Alpaca 适配器时

from riskguard import RiskEngine, RiskConfig, Order, Side, PaperBroker

# 1) 一个带滑点 + 手续费的纸面模拟盘(呼应"回测和实盘是两个世界")
broker = PaperBroker(cash=100_000, slippage_bps=2, commission_bps=1,
                     marks={"AAPL": 200.0})

# 2) 把纪律写进配置:单笔 ≤ 10%、回撤 15% 熔断
engine = RiskEngine(RiskConfig(max_position_pct=0.10, max_drawdown_pct=0.15),
                    broker=broker)

# 3) 想买 1000 股 = 20 万 = 占比 200%,风控自动缩到上限
decision = engine.check(Order("AAPL", Side.BUY, 1000), broker.get_portfolio())
print(decision.decision)          # Decision.RESIZE
print(decision.order.quantity)    # 50.0  ← 缩到 10% 权益
print(decision.reasons())
# AAPL capped to 10.00% of equity (qty 1000 -> 50); gross exposure capped ... (qty 1000 -> 500)
# ↑ 多条规则可能同时报告,引擎取最保守的那个(min = 50)

# 4) 放行则真正下单(这里进模拟盘)
engine.submit(Order("AAPL", Side.BUY, 40), broker.get_portfolio())

熔断触发后必须人工复盘再 engine.reset_breaker()——不把原因想明白,不许重启。

🎲 动态下注:让公式算,不让情绪算

from riskguard import KellySizer, Signal, Side

engine = RiskEngine(RiskConfig(kelly_fraction=0.5), sizer=KellySizer(), broker=broker)

# Kelly 需要胜率与盈亏比;f = kelly_fraction × (p − q/b)
sig = Signal("AAPL", Side.BUY, price=200.0, win_probability=0.55, payoff_ratio=1.5)
engine.size_and_submit(sig, broker.get_portfolio())   # 无正期望时返回 None,自动不下注

• FixedFractionalSizer —— 固定比例,最难自欺。
• KellySizer —— 分数 Kelly,满 Kelly 太颠簸,默认取半。
• VolatilityTargetSizer —— 目标波动率 / 实现波动率,给每个头寸等量风险预算。

🚨 实时哨兵(kill-switch)

光把纪律写进配置还不够——你睡觉、开会、没信号的时候,市场不会为了配合你的作息暂停。RiskMonitor 是一条永不闭眼的后台线程,周期性盯着账户,触线自动踩刹车:

from riskguard import RiskMonitor

with RiskMonitor(engine, broker, interval=5.0, auto_liquidate=True):
    ...   # 后台线程:周期观测权益 → 触及回撤红线自动熔断并平仓

📿 防篡改审计

from riskguard import RiskEngine, JsonlAuditSink

with JsonlAuditSink("audit.jsonl", hmac_key="放在日志之外的密钥") as audit:
    engine = RiskEngine(broker=broker, audit=audit)
    ...

JsonlAuditSink.verify("audit.jsonl", hmac_key="...", expected_count=42)   # 验真伪

诚实说清边界:不带密钥的纯哈希链只能防"改/重排中间记录",挡不住尾部截断或整体重写。要真正防伪,传 hmac_key(密钥存在日志之外)+ verify(expected_count=N) 用外部锚点核对条数。

🔒 状态持久化:堵住"重启绕过熔断"的后门

RiskState 默认纯内存——进程一重启,高水位和熔断标记全部归零,等于给亏红眼的操作者留了一条"重启一下继续交易"的隐藏后路。这与本库的立库初衷直接矛盾。接一个 state_store,熔断就真的扛得住重启(examples/10):

from riskguard import RiskEngine, SqliteStateStore

store = SqliteStateStore("risk_state.db")
engine = RiskEngine(config, broker=broker, state_store=store)
# 构造时自动从存档恢复高水位/熔断/策略入役时间;此后每次状态变更自动写透。
# 存档读取失败会直接抛出——宁可拒绝启动,也不能悄悄当成"一切正常"重新开始。

CLI 同理支持跨调用持久化(否则脚本反复调用本身就是一种"重启绕过"),多个策略/标的共用一个数据库文件时用 --state-key 隔离:

$ riskguard check --state-db risk_state.db --state-key strategy_a \
    --equity 80000 --side buy --qty 10 --price 100

诚实说清边界:这堵住的是"重启进程"这条最常见的绕过路径,不是万能防绕过—— 一个决心要绕的人仍可以删掉存档文件、改配置、或者干脆改代码。它提高的是绕过成本, 不是消灭绕过本身;个人量化的风控终究是尤利西斯之约,不是机构级的权责分离。

一个 (存档文件, key) 只能有一个活跃写者:两个引擎共用同一个 key 会被乐观锁 (版本号 CAS)检测到并报错,而不是静默互相覆盖——多策略/多账户请务必用不同 key。

🌙 每日体检 + 压力测试:AI 帮你盯盘的那一层

"AI 帮你 24 小时盯盘"这个刚需,拆开看是四层:实时监控、异常检测、硬编码熔断线、压力测试。硬编码熔断线是 RiskGuard 从第一行代码起就在做的事(上面几节都是);这里补上另外两层里属于本库使命范围的部分——每日体检(把"实时监控"摊开成一份可复算的摘要)和压力测试(把"如果……会怎样"摊开成一份推演)。

刻意不做的事:统计/AI 判定"这次不一样"式的异常检测,不属于这个库——那是数据科学能力(需要历史基线、新闻/情绪输入、模型打分),和 RiskGuard "确定性、fail-closed、绝不模糊判断" 的设计哲学是两种不同的产品。这一层该由外部的 AI agent 来做:agent 负责观察、判断、叙述,RiskGuard 只负责保证它拿到的事实是真的、以及在规则被触发时毫不犹豫地执行——这条线不能由 AI 决定。

from riskguard.reporting import build_digest, render_digest_text, run_stress_test, render_stress_text

# 每日体检:把 engine 当前状态(高水位/熔断/隔离策略)+ 一份持仓快照,组装成结构化摘要
report = build_digest(engine, portfolio)
print(render_digest_text(report))
report.to_dict()   # 结构化数据,交给 AI agent 去叙述、去推送提醒

# 压力测试:"如果我的持仓明天集体跌 20%,我扛得住吗"——绝对只读,零副作用
result = run_stress_test(engine, portfolio, shock_pct=-0.20)
print(render_stress_text(result))

CLI 同样支持,适合脚本化的"每天 check-in"和"随手问一句"(examples/11、examples/12):

$ riskguard digest --equity 95000 --position AAPL:80:190 --position TSLA:-40:250 --state-db risk.db
$ riskguard stress --equity 95000 --shock -0.20 --position AAPL:80:190 --position TSLA:-40:250

压力测试的只读承诺:不触发熔断、不写审计、不碰持久化——即便 --state-db 指向一个还不存在的文件,也不会在磁盘上留下任何新文件。它是纯粹的"假设推演",算完就完。

🔌 接进你的回测框架

把 RiskGuard 作为"风险叠加层"接进 backtesting.py / vectorbt,让策略研究和爆仓防护第一次在同一条流水线上(examples/07)。

# 框架无关的叠加层:目标持仓 → 风控批准的订单;还能一键跑"套风控 vs 不套"对比
from riskguard.backtest import RiskOverlay, compare

res = compare(prices, my_strategy, config=RiskConfig(max_position_pct=0.10))
print(res["naive"].max_drawdown, res["guarded"].max_drawdown)  # 例:−45% vs −5.6%

# backtesting.py:子类只写 signal() 返回目标权重,下单自动过风控
from riskguard.backtest import make_riskguard_strategy
class MyStrat(make_riskguard_strategy(RiskConfig(max_position_pct=0.10))):
    def signal(self): return 1.0 if bullish else 0.0

# vectorbt:把仓位按上限封顶(纯函数,不装 vectorbt 也能用)
from riskguard.backtest import risk_capped_weights, kelly_weights
size = risk_capped_weights(target_weights, RiskConfig(max_position_pct=0.10))

🎚️ 三档预设 + 命令行

不想逐个调参?挑一档起步——保守 / 均衡 / 激进:

from riskguard import get_preset, RiskEngine
engine = RiskEngine(get_preset("conservative"))   # 或 "balanced" / "aggressive"
# 再用 config.replace(...) 微调成你自己的口径

装好后还带一个零依赖命令行,一键验单、看预设、跑对比:

$ riskguard check --preset balanced --equity 100000 --side buy --qty 1000 --price 200
裁决:  RESIZE
放行:  BUY 50 ASSET  (占权益 10.0%)

$ riskguard presets                               # 三档参数对照表
$ riskguard replay --prices 100,96,90,82,75,70    # 套风控 vs 不套 的回撤对比

🗺️ 架构

        Signal ──Sizer──┐          RiskOverlay(回测叠加层) / reporting(digest · stress)
     Kelly·波动率·固定  │          都是同一个 RiskEngine 的不同调用方,只读或推演
                        ▼
                      Order
                        │
                        ▼
        ┌──────────────────────────────────┐
        │          RiskEn