监控、风控闸门与"熔断开关"

第一层：通俗理解

1.1 银行柜台的复核

银行柜员办理一笔大额转账，系统会自动检查：账户余额够不够、金额是否超过单笔限额、收款账户是否在黑名单上、今天是否已经转了很多笔。任何一项不通过，转账就办不了，柜员本人也绕不过去。

交易系统里的风控闸门就是这个"柜台复核"。它和策略是分开的：策略可能有 bug、可能被错误的数据误导，但风控闸门只做简单、确定、可审查的检查。

1.2 Knight Capital 事件（2012 年 8 月 1 日）

事后分析中被反复强调的几点：部署流程没有自动校验、没有对异常订单量的自动熔断、系统发出的告警邮件没有引起及时行动。

第二层：风控闸门检查什么

2.1 交易前检查清单

2.2 熔断开关

一个任何时候都能用、不依赖策略状态的开关：

• 立即拒绝所有新订单；
• 撤销所有挂单；
• 可选：按预设规则平掉全部或部分头寸。

它应该有多个触发方式：人工按钮、自动规则（亏损、订单量、错误率超阈值）、甚至外部的"死人开关"（监控系统失联就自动触发，见《安全监控与告警疲劳》）。

2.3 设计原则

第三层：代码：一个最小的风控闸门

3.1 实现

from dataclasses import dataclass, field
import time

@dataclass
class Order:
    symbol: str
    side: str        # "buy" / "sell"
    qty: float
    price: float

@dataclass
class RiskLimits:
    max_order_notional: float = 50_000
    max_position_notional: float = 180_000
    max_price_deviation: float = 0.05      # 偏离参考价 5% 以上拒绝（防"胖手指"）
    max_orders_per_sec: int = 5
    allowed_symbols: set = field(default_factory=lambda: {"BTC-USD", "ETH-USD"})

class PreTradeRisk:
    def __init__(self, limits, ref_price, positions):
        self.lim, self.ref, self.pos = limits, ref_price, positions
        self.recent, self.killed = [], False

    def kill(self):            # 熔断开关：立即停止一切新订单
        self.killed = True

    def check(self, o: Order):
        if self.killed:
            return False, "熔断开关已触发"
        if o.symbol not in self.lim.allowed_symbols:
            return False, "标的不在白名单"
        if o.qty <= 0 or o.price <= 0:
            return False, "数量或价格非法"
        notional = o.qty * o.price
        if notional > self.lim.max_order_notional:
            return False, f"单笔金额 {notional:,.0f} 超限"
        ref = self.ref[o.symbol]
        if abs(o.price / ref - 1) > self.lim.max_price_deviation:
            return False, f"价格偏离参考价 {o.price/ref-1:+.1%}"
        signed = o.qty if o.side == "buy" else -o.qty
        new_pos = (self.pos.get(o.symbol, 0) + signed) * ref
        if abs(new_pos) > self.lim.max_position_notional:
            return False, f"成交后持仓 {new_pos:,.0f} 超限"
        now = time.monotonic()
        self.recent = [t for t in self.recent if now - t < 1.0]
        if len(self.recent) >= self.lim.max_orders_per_sec:
            return False, "下单频率超限"
        self.recent.append(now)
        return True, "通过"

3.2 测试

risk = PreTradeRisk(RiskLimits(), {"BTC-USD": 60_000, "ETH-USD": 3_000}, {"BTC-USD": 2.5})
tests = [
    Order("BTC-USD", "buy", 0.5, 60_100),
    Order("BTC-USD", "buy", 0.5, 66_000),    # 价格偏离
    Order("ETH-USD", "buy", 20, 3_000),      # 单笔超限
    Order("BTC-USD", "buy", 0.8, 60_000),    # 持仓超限
    Order("DOGE-USD", "buy", 100, 0.1),      # 白名单
]
for o in tests:
    print(o.symbol, o.side, o.qty, o.price, "->", risk.check(o))
for i in range(6):
    ok, why = risk.check(Order("ETH-USD", "sell", 1, 3_000))
print("连续下单第 6 笔:", ok, why)
risk.kill(); print("熔断后:", risk.check(Order("ETH-USD", "sell", 1, 3_000)))

输出：

BTC-USD buy 0.5 60100 -> (True, '通过')
BTC-USD buy 0.5 66000 -> (False, '价格偏离参考价 +10.0%')
ETH-USD buy 20 3000 -> (False, '单笔金额 60,000 超限')
BTC-USD buy 0.8 60000 -> (False, '成交后持仓 198,000 超限')
DOGE-USD buy 100 0.1 -> (False, '标的不在白名单')
连续下单第 6 笔: False 下单频率超限
熔断后: (False, '熔断开关已触发')

3.3 生产环境还需要什么

这段代码只是骨架，真实系统还需要：

总结