4、闪崩行情应对策略：暂停报价机制、撤单保护、价格保护带、熔断逻辑实现
闪崩行情，做市商最怕的东西。
我入行第三年就碰上过一次。那会儿还在用老一套的风控逻辑，结果一秒亏掉三个月的利润。嗯，从那以后，我对闪崩的敬畏心就刻进骨子里了。
说白了，闪崩就是价格在极短时间内出现剧烈、无序的波动。可能是乌龙指，可能是程序化踩踏，也可能是消息面冲击。不管什么原因，做市引擎必须能扛住。
这一章，我把自己踩过的坑和验证过的方案，拆成四个核心策略来讲。
4.1 暂停报价机制：别在瀑布里接飞刀
行情开始崩的时候，最忌讳什么？
还在那儿傻傻地挂单。
你想想看，价格每秒跳几十个tick，你的报价模型根本来不及反应。挂出去的单子，大概率被吃掉，而且吃在对你最不利的位置。
所以，第一道防线就是——暂停报价。
我个人习惯用两个维度来触发暂停：
- 价格变化率： 比如1秒内价格变动超过0.5%
- 波动加速度： 连续3个tick的变动幅度都在递增
代码实现上，我一般这样写：
class QuotePauseManager:
def __init__(self, price_change_threshold=0.005, tick_count=3):
self.threshold = price_change_threshold
self.tick_count = tick_count
self.price_history = []
self.is_paused = False
def check_flash_crash(self, new_price):
self.price_history.append(new_price)
if len(self.price_history) > self.tick_count:
self.price_history.pop(0)
if len(self.price_history) < self.tick_count:
return False
# 计算连续tick的变化率
changes = []
for i in range(1, len(self.price_history)):
change = abs(self.price_history[i] - self.price_history[i-1]) / self.price_history[i-1]
changes.append(change)
# 如果所有变化都超过阈值，触发暂停
if all(c > self.threshold for c in changes):
self.is_paused = True
return True
return False
def resume_quoting(self):
# 恢复条件：价格稳定一段时间
self.is_paused = False
self.price_history.clear()4.2 撤单保护：别让旧单子害了你
暂停报价只是第一步。更关键的是——把已经挂出去的单子撤回来。
为什么？因为你的报价是基于几毫秒前的行情算出来的。在闪崩行情里，这几毫秒足以让价格跑出几个档位。你挂的单子，可能已经严重偏离当前合理价格。
撤单保护的核心就一句话：发现异常，先撤为敬。
我一般会做两层撤单：
- 主动撤单： 引擎检测到异常后，立即撤销所有活跃订单
- 被动撤单： 如果某个订单长时间未成交，且市场波动加剧，也主动撤掉
这里有个坑，我曾经踩过——撤单指令本身也有延迟。在极端行情下，交易所的撤单通道可能拥堵。所以，撤单要带重试机制。
def cancel_all_orders_with_retry(exchange, order_ids, max_retries=3):
for order_id in order_ids:
for attempt in range(max_retries):
try:
exchange.cancel_order(order_id)
break # 成功就跳出
except Exception as e:
if attempt == max_retries - 1:
log.error(f"撤单失败: {order_id}, 错误: {e}")
else:
time.sleep(0.1) # 重试前等100ms4.3 价格保护带：给报价上个保险
暂停和撤单都是事后处理。有没有办法在报价生成时就避免问题？
有——价格保护带。
说白了，就是给报价设定一个「安全区间」。任何超出这个区间的报价，引擎直接拒绝，不往外发。
保护带的设定方式，我常用两种：
| 类型 | 计算方式 | 适用场景 |
|---|---|---|
| 固定偏移 | 基于参考价 ± 固定点数 | 波动较小的品种 |
| 动态偏移 | 基于近期波动率 × 倍数 | 波动较大的品种 |
我个人更推荐动态偏移。因为固定偏移在波动率放大时，保护效果会打折扣。
class PriceProtectionBand:
def __init__(self, volatility_window=20, multiplier=3):
self.window = volatility_window
self.multiplier = multiplier
self.price_list = []
def update_price(self, price):
self.price_list.append(price)
if len(self.price_list) > self.window:
self.price_list.pop(0)
def get_band(self, reference_price):
if len(self.price_list) < 2:
return (reference_price * 0.99, reference_price * 1.01)
# 计算近期波动率
returns = []
for i in range(1, len(self.price_list)):
r = (self.price_list[i] - self.price_list[i-1]) / self.price_list[i-1]
returns.append(r)
volatility = np.std(returns)
band_width = volatility * self.multiplier
lower = reference_price * (1 - band_width)
upper = reference_price * (1 + band_width)
return (lower, upper)
def is_price_safe(self, price, reference_price):
lower, upper = self.get_band(reference_price)
return lower <= price <= upper4.4 熔断逻辑实现：最后的保险丝
如果暂停、撤单、保护带都失效了怎么办？
那就上熔断。
熔断是最后一道防线。它的逻辑很简单：当损失达到某个阈值时，强制停止所有做市活动。
熔断的触发条件，我一般设三个级别：
- 一级熔断： 单笔亏损超过阈值 → 暂停该品种报价
- 二级熔断： 累计亏损超过阈值 → 暂停所有品种报价
- 三级熔断： 总资产回撤超过阈值 → 完全停止引擎，人工介入
实现上，我习惯用一个状态机来管理：
class CircuitBreaker:
def __init__(self, thresholds):
self.thresholds = thresholds # {1: 单笔, 2: 累计, 3: 总回撤}
self.state = 0 # 0: 正常, 1: 一级, 2: 二级, 3: 三级
self.daily_pnl = 0
self.total_asset = 0
def check_and_trigger(self, trade_pnl):
self.daily_pnl += trade_pnl
# 检查三级熔断
drawdown = abs(self.daily_pnl) / self.total_asset
if drawdown > self.thresholds[3]:
self.state = 3
self.trigger_shutdown()
return
# 检查二级熔断
if abs(self.daily_pnl) > self.thresholds[2]:
self.state = 2
self.pause_all_quoting()
return
# 检查一级熔断
if abs(trade_pnl) > self.thresholds[1]:
self.state = 1
self.pause_symbol_quoting()
return
def reset(self):
# 熔断恢复需要人工确认
pass4.5 四道防线的联动逻辑
这四个策略不是孤立的。它们应该形成一个递进式的防御体系。
我画了一张流程图，帮你理解它们之间的关系：
你看，整个流程是串行的。每一层都在过滤风险。如果某一层没通过，就直接进入对应的处理逻辑，不再往下走。
这样做的好处是——早发现，早处理。不用等到熔断才动手，前面几层已经把大部分风险挡在外面了。
嗯，这些策略我都在生产环境验证过。不能说100%防住所有闪崩，但至少能让你在极端行情下，活着走出来。