黑天鹅演练与断线预案

黑天鹅在交易系统里的四种面孔

「黑天鹅」这个词容易让人往宏观事件上想，但对一个自动交易系统来说，无论外部世界发生什么，最终传到你系统里的只有四类症状。抓住症状而不是原因，预案才写得出来。

第一类是交易所宕机或维护：下单接口返回错误、行情停止推送、网页打不开，但你的持仓还在。第二类是稳定币脱锚：报价的分母出问题，USDT 或某个抵押品短时偏离 1 美元，所有以它计价的策略同时失真。

第三类是API 限频与鉴权失效：请求被 429 拒绝或密钥被风控临时冻结，系统还在运行但指令发不出去，这是最容易被低估的一类。第四类是极端插针：行情在数秒内出现远超正常范围的偏离，止损单被以极差价格成交，然后价格立刻回归。

逐类拆解应对动作

预案必须写到「按哪个键」的粒度。下面这张表是可以直接搬进运维文档的形式，每一行都对应一个明确的判定条件和一个明确的动作。

表里有两条值得单独强调。「禁止动作」列往往比「立即动作」列更重要：宕机时的重试风暴会在交易所恢复的瞬间造成重复成交，脱锚时的抄底会把一次可控事件变成本金归零事件。事故中大部分额外损失来自应激反应，而不是事故本身。

另一条是持仓快照。系统在检测到异常的第一时间就应把当前持仓、挂单、余额写入本地文件并打上时间戳。恢复时你需要拿这份快照和交易所的真实状态做逐笔比对——差异就是事故期间发生的所有事情。

断线自动降险：心跳与超时保护

断线预案的核心机制只有两个：心跳检测判断连接是否还活着，超时保护决定失去连接后系统自己做什么。二者缺一，自动化就形同虚设。

心跳的关键是双向。只监听交易所的推送不够，因为 TCP 连接可能保持而数据早已停止；必须主动发 ping 或定期查询一个轻量接口，并记录最后一次成功响应的时间戳。超时判定用这个时间戳，而不是用「有没有报错」。

超时保护的默认动作必须是降险，而不是等待。很多人把断线处理写成「重连直到成功」，这在持有杠杆仓位时是危险的默认值——你在完全看不见行情的情况下继续承担风险。合理设计是分级：短时超时只停开仓，长时超时主动减仓，超长时间无连接则尝试所有可用通道清仓。

import time

class Watchdog:
    """心跳检测 + 分级超时保护（教学用伪代码，需接入真实交易接口）。"""

    def __init__(self, warn_s=15, derisk_s=45, flatten_s=120):
        self.warn_s = warn_s        # 停止新开仓
        self.derisk_s = derisk_s    # 敞口减半
        self.flatten_s = flatten_s  # 尝试清仓
        self.last_ok = time.time()
        self.state = 'RUNNING'

    def heartbeat(self):
        """每次成功收到行情或接口响应时调用。"""
        self.last_ok = time.time()
        if self.state != 'RUNNING':
            snapshot_and_reconcile()   # 恢复前必须先对账，不可直接续跑
            self.state = 'RUNNING'

    def check(self):
        gap = time.time() - self.last_ok

        if gap < self.warn_s:
            return self.state

        if gap >= self.flatten_s and self.state != 'FLATTEN':
            self.state = 'FLATTEN'
            save_snapshot('flatten')
            for channel in ('rest_primary', 'rest_backup', 'manual_alert'):
                if close_all_positions(via=channel):
                    break
        elif gap >= self.derisk_s and self.state == 'HALT_NEW':
            self.state = 'DERISK'
            save_snapshot('derisk')
            reduce_exposure(ratio=0.5)
        elif self.state == 'RUNNING':
            self.state = 'HALT_NEW'
            save_snapshot('halt')
            cancel_all_open_orders()   # 撤单优先于减仓：先止血再处理

        return self.state


# 主循环：check() 必须独立于行情线程运行，
# 否则行情断了检测逻辑也一起停了——这是最常见的实现错误。
def risk_loop(dog):
    while True:
        dog.check()
        time.sleep(1)

代码里有三个容易踩的坑。第一，看门狗必须跑在独立线程或独立进程；写在行情回调里的检测逻辑会随行情一起停止。第二，撤单排在减仓之前，因为断线期间残留的挂单可能在恢复瞬间以极差价格成交。第三，恢复时必须先对账再续跑，直接把状态改回 RUNNING 会让程序基于错误的持仓认知继续下单。

python

演练：没跑过的预案不算预案

预案写在文档里的价值接近于零。真正有效的验证方式是主动制造故障，在可控条件下跑一遍完整流程。这件事在小资金阶段成本最低，也最该做。

演练要覆盖三个层次：连接层（拔网线、关 WiFi、把交易所域名指向黑洞）、接口层（用错误密钥、故意超频触发 429、模拟返回 5xx）、行情层（回放一段历史插针数据，看策略与风控的反应）。第三层可以在回测环境做，前两层必须在实盘环境用最小仓位做。

最后两项常被忽略但价值最高。进程崩溃重启的对账逻辑是重复下单事故的唯一防线；人工介入通道则是全部自动化失效后的兜底——包括在你的电脑断电、机房故障、账号被风控时，你手上有没有一份写清了「登录哪里、点哪里、清哪个仓」的纸面流程。相关的风控层级设计可以配合三层止损体系一起看。

告警：谁在事故发生的第一分钟知道

自动降险处理的是系统能自己判断的情况，但有一类事故系统判断不出来：它运行得好好的，只是基于错误的数据在做决策。这类事故只能靠告警加人工介入，因此告警链路本身就是预案的一部分。

告警的设计有三个要点。第一是告警必须走独立通道。如果告警依赖的网络或服务器与交易系统是同一套，那么最需要告警的时候它正好也挂了。可行做法是用手机端推送服务，并且定期发送心跳消息——收不到心跳本身就是一种告警。

第二是分级而不是全部推送。把告警分成三级：信息级只写日志，警告级推送但不打扰（如策略暂停），紧急级必须响铃唤醒（如账户回撤触线、自动清仓已执行）。不分级的结果是告警疲劳，几十条无关消息之后你会关掉通知，于是真正的紧急告警也被一起屏蔽。

第三是告警要带上下文与建议动作。「策略 A 已暂停」这条消息价值有限，「策略 A 因连亏 8 笔暂停，当前持仓 0.3 BTC，账户回撤 6.2%，建议核对最近成交记录」才能让你在半夜被叫醒时立刻做出判断。写告警文案的时间投入很小，回报却在事故当时全部体现。

把事故成本提前计入策略评估

回测几乎从不包含事故成本，这让高频与低容错策略在纸面上被系统性高估。一个务实的做法是给策略加一项事故折扣：按历史频率估算每年遇到几次影响交易的事故，每次造成多少损失，从回测收益里直接扣掉。

举例说明估算方式：假设某策略每年遇到 4 次接口不可用（平均每次损失 0.5% 净值）、1 次极端插针（损失 2%）、1 次稳定币波动导致的暂停（错失收益 1%），合计约 5% 的年化折扣。如果策略的回测年化本来就在 8% 附近，扣掉之后剩下的部分是否还值得承担运维复杂度，答案就清楚了。以上均为示例数字，基于历史事件频率的粗略假设，不构成收益预测。

这个视角还会改变策略选择。对断线越敏感的策略，事故折扣越大：需要持续在线做市的策略折扣最高，日频调仓的趋势策略折扣很低，纯现货低频策略几乎为零。把这一项纳入比较，很多人会发现自己被高频策略的回测数字误导了。策略容错性的评估方法在量化学院里有更完整的展开。