自动运行的交易策略，什么时候需要人工干预

小象实战讲义 · 进阶量化交易实战：迭代式的量化策略研发 本讲义仅用于编程与量化方法教学，不构成任何投资建议。

全自动交易系统并非永动机，完全放任不管与随意伸手乱改都是灾难的根源。在什么明确的边界条件下量化交易员必须介入？在什么情况下绝对不能插手？本节深入剖析人工干预的标准化准则、干预分级机制以及紧急一键平仓的触发红线。

💡 核心导读

• 量化实盘人工干预四大明确准则：
  • 准则 1（严禁干预）：处于历史回撤范围内：策略浮亏在正常回撤区间内，坚决不允许基于主观情绪干预；
  • 准则 2（一级干预）：外部不可抗力通知：交易所临时变更交易时间、提前休市或拔网线，手动暂停开仓；
  • 准则 3（二级干预）：数据与接口严重异常：行情源发生大面积中断或持续跳变，挂起策略并撤销未成交挂单；
  • 准则 4（三级干预）：系统致命死循环或逻辑穿仓：发生重复下单、资金透支等恶性 Bug，秒级启动 Emergency Stop 一键全平。

┌─────────────────────────────────────────────────────────────┐
│                    人工干预分级决策树拓扑图                 │
└──────────────────────────────┬──────────────────────────────┘
                               │
 ┌─────────────────────────────┼─────────────────────────────┐
 ▼                             ▼                             ▼
┌─────────────────────────┐   ┌─────────────────────────┐   ┌─────────────────────────┐
│ 正常波动 (严禁干预)     │   │ 外部异常 (一至二级挂起) │   │ 致命故障 (三级紧急全平) │
├─────────────────────────┤   ├─────────────────────────┤   ├─────────────────────────┤
│ • 浮亏 < 历史最大回撤   │──►│ • 数据断流 / 交易所通知 │──►│ • 死循环重复报单        │
│ • 遵守大数定律统计纪律  │   │ • 撤销挂单，暂停新开仓  │   │ • 一键强平，断电拔网线  │
└─────────────────────────┘   └─────────────────────────┘   └─────────────────────────┘

一、人工干预分级状态机数学逻辑

干预状态机分为 4 个层级： $$\text{InterventionLevel} = \begin{cases} 0 (\text{None}), & \text{Normal trading within risk limits} \\ 1 (\text{Pause}), & \text{External scheduled halt} \\ 2 (\text{Freeze}), & \text{Data feed anomaly or connection loss} \\ 3 (\text{KillSwitch}), & \text{Abnormal high-freq loops or capital deficit} \end{cases}$$

import pandas as pd
import numpy as np

class InterventionDecisionEngine:
    """
    人工干预分级决策引擎（合成教学样例，实际阈值需根据风控规则动态调整）
    """
    def evaluate_situation(self, is_normal_loss: bool, is_feed_lost: bool, is_order_looping: bool, is_external_halt: bool = False) -> str:
        if is_order_looping:
            return "LEVEL_3_KILL_SWITCH_LIQUIDATE"
        elif is_feed_lost:
            return "LEVEL_2_FREEZE_NEW_ORDERS"
        elif is_external_halt:
            return "LEVEL_1_PAUSE_NEW_ORDERS"
        elif is_normal_loss:
            return "LEVEL_0_STRICT_NO_INTERVENTION"
        return "LEVEL_0_NORMAL"

# 合成测试场景：模拟不同故障与正常情况
engine = InterventionDecisionEngine()
print("===== 人工干预决策引擎测试 =====")
print(f"场景1-正常回撤: {engine.evaluate_situation(is_normal_loss=True, is_feed_lost=False, is_order_looping=False)}")
print(f"场景2-数据中断: {engine.evaluate_situation(is_normal_loss=False, is_feed_lost=True, is_order_looping=False)}")
print(f"场景3-报单死循环: {engine.evaluate_situation(is_normal_loss=False, is_feed_lost=False, is_order_looping=True)}")
print(f"场景4-外部暂停: {engine.evaluate_situation(is_normal_loss=False, is_feed_lost=False, is_order_looping=False, is_external_halt=True)}")
print(f"场景5-运行正常: {engine.evaluate_situation(is_normal_loss=False, is_feed_lost=False, is_order_looping=False)}")

===== 人工干预决策引擎测试 =====
场景1-正常回撤: LEVEL_0_STRICT_NO_INTERVENTION
场景2-数据中断: LEVEL_2_FREEZE_NEW_ORDERS
场景3-报单死循环: LEVEL_3_KILL_SWITCH_LIQUIDATE
场景4-外部暂停: LEVEL_1_PAUSE_NEW_ORDERS
场景5-运行正常: LEVEL_0_NORMAL

二、一键全平与挂单清除 SOP 动作

当判定为 Level 3 故障时，系统必须在单次原子调用中完成全量撤单与市价清仓。

流动性口径提示：市价一键全平适用于高流动性期货主力合约；在 A 股一字跌停或期货冷门合约流动性真空时，市价单可能被拒单或承受极端滑点，工业级做法是改以跌停板价（买入平仓时用涨停板价）挂限价单（如 LIMIT_DOWN_CLOSE），既保证进入交易所撮合队列，又锁定最差可接受成交价。

import pandas as pd
import numpy as np

def execute_level3_emergency_actions(open_orders_count: int, positions_dict: dict) -> list:
    """
    Level 3 紧急干预动作流水线（合成教学样例，实盘需对接交易所API）
    """
    actions = []
    if open_orders_count > 0:
        actions.append("CANCEL_ALL_OPEN_ORDERS")
    for sym, vol in positions_dict.items():
        if vol != 0:
            # 高流动性主力合约可用市价 MARKET_CLOSE; 流动性真空(一字跌停/冷门合约)时应改挂
            # 跌停板限价单 LIMIT_DOWN_CLOSE(买入平仓用涨停价), 防拒单与极端滑点
            actions.append(f"MARKET_CLOSE_{sym}_{vol}")
    actions.append("SYSTEM_HALT_AND_LOG")
    return actions

# 合成持仓与挂单场景
positions = {'rb2410': 15, 'cu2408': -6, 'ag2412': 0}
open_orders = 4
emergency_plan = execute_level3_emergency_actions(open_orders_count=open_orders, positions_dict=positions)
print("===== Level 3 应急动作流水线 =====")
print(f"当前挂单数: {open_orders}")
print(f"当前持仓: {positions}")
print("应急动作序列:")
for i, action in enumerate(emergency_plan, 1):
    print(f"  {i}. {action}")

===== Level 3 应急动作流水线 =====
当前挂单数: 4
当前持仓: {'rb2410': 15, 'cu2408': -6, 'ag2412': 0}
应急动作序列:
  1. CANCEL_ALL_OPEN_ORDERS
  2. MARKET_CLOSE_rb2410_15
  3. MARKET_CLOSE_cu2408_-6
  4. SYSTEM_HALT_AND_LOG

📝 动手练一练

编写 Python 测试脚本，模拟不同故障情景下的人工干预决策与 Level 3 应急动作执行：

import pandas as pd
import numpy as np

class InterventionDecisionEngine:
    """
    人工干预分级决策引擎（合成教学样例，实际阈值需根据风控规则动态调整）
    """
    def evaluate_situation(self, is_normal_loss: bool, is_feed_lost: bool, is_order_looping: bool, is_external_halt: bool = False) -> str:
        if is_order_looping:
            return "LEVEL_3_KILL_SWITCH_LIQUIDATE"
        elif is_feed_lost:
            return "LEVEL_2_FREEZE_NEW_ORDERS"
        elif is_external_halt:
            return "LEVEL_1_PAUSE_NEW_ORDERS"
        elif is_normal_loss:
            return "LEVEL_0_STRICT_NO_INTERVENTION"
        return "LEVEL_0_NORMAL"

def execute_level3_emergency_actions(open_orders_count: int, positions_dict: dict) -> list:
    """
    Level 3 紧急干预动作流水线（合成教学样例，实盘需对接交易所API）
    """
    actions = []
    if open_orders_count > 0:
        actions.append("CANCEL_ALL_OPEN_ORDERS")
    for sym, vol in positions_dict.items():
        if vol != 0:
            actions.append(f"MARKET_CLOSE_{sym}_{vol}")
    actions.append("SYSTEM_HALT_AND_LOG")
    return actions

# 综合测试：决策引擎与应急响应
engine = InterventionDecisionEngine()
dec_normal = engine.evaluate_situation(is_normal_loss=True, is_feed_lost=False, is_order_looping=False)
dec_feed = engine.evaluate_situation(is_normal_loss=False, is_feed_lost=True, is_order_looping=False)
dec_fatal = engine.evaluate_situation(is_normal_loss=False, is_feed_lost=False, is_order_looping=True)
actions = execute_level3_emergency_actions(open_orders_count=3, positions_dict={'rb2410': 10, 'cu2408': -4})

print("================ 人工干预决策树与应急响应测试 ================")
print(f"正常回撤波动判定: {dec_normal} (严禁主观插手)")
print(f"数据中断故障判定: {dec_feed}")
print(f"死循环报单故障判定: {dec_fatal}")
print(f"Level 3 应急动作执行: {actions}")
assert dec_normal == "LEVEL_0_STRICT_NO_INTERVENTION", "正常回撤未能阻止主观干预！"
assert (3 == 0) or actions[0] == "CANCEL_ALL_OPEN_ORDERS", "紧急撤单未优先执行！"
print("✅ 人工干预分级准则与应急 SOP 验证 PASS")
print("============================================================")

运行结果输出：

================ 人工干预决策树与应急响应测试 ================
正常回撤波动判定: LEVEL_0_STRICT_NO_INTERVENTION (严禁主观插手)
数据中断故障判定: LEVEL_2_FREEZE_NEW_ORDERS
死循环报单故障判定: LEVEL_3_KILL_SWITCH_LIQUIDATE
Level 3 应急动作执行: ['CANCEL_ALL_OPEN_ORDERS', 'MARKET_CLOSE_rb2410_10', 'MARKET_CLOSE_cu2408_-4', 'SYSTEM_HALT_AND_LOG']
✅ 人工干预分级准则与应急 SOP 验证 PASS
============================================================

本章小结

1. 严禁在正常回撤中情绪化插手：恪守纪律，让策略在完整统计样本中展现其长期概率优势。
2. 建立分级干预机制：清晰界定挂起新开仓与一键全平的明确触发条件，拒绝模糊地带。
3. 熟练演练一键全平 SOP：确保在突发致命软件故障时具备秒级撤单清仓并自保的实战能力。

📋 行动清单 (Action Checklist)

• 运行人工干预测试代码，核查不同故障组合下的状态分发；
• 打印并张贴《量化实盘人工干预分级标准与红线守则》；
• 预习最后一节，掌握实盘应急预案准备与交易心理建设。

—— 小象教研组