开仓动量过滤器设计方案

版本：v3.0
日期：2026-03-06
适用系统：Hyperliquid 量化交易系统（Adaptive Bollinger Z-Score 配对策略）
升级摘要：v2.0 → v3.0 新增 Layer 0（Hurst DFA），Rogers-Satchell 替代 Garman-Klass，Per-symbol 基准波动率，非对称阈值，ADF+ER 双重 Spread 仲裁

目录

1. 背景与问题定义
2. 算法选择与学术依据
3. 算法具体实现
4. 融入当前交易体系的方案
5. 与当前交易风格的契合度分析
6. 与四项开单约束的对应关系
7. 参数配置指南
8. 风险与局限性

1. 背景与问题定义

1.1 交易系统现状

当前系统基于 Adaptive Bollinger Z-Score 配对策略，核心逻辑是协整对的 spread 均值回归：

• 当 adaptive_z 突破 adaptive_threshold 时产生入场信号
• 当 adaptive_z 回归至 entry_adaptive_z * reversion_factor 时平仓
• 使用 5 分钟 K 线，最大持仓时间控制在 72 小时以内（max_hold_hours=72.0）
• 策略引擎 _check_entry() 已有冷却期、突破检测、持仓检查、z4h 绝对值过滤四层前置过滤

1.2 待解决的问题

均值回归策略在以下场景存在明显风险：

1.3 四项约束目标

约束 1：连续下跌 → 不追跌（不做空）
约束 2：连续上涨 → 不追涨（不做多）
约束 3：迅速暴涨 → 不做空
约束 4：迅速暴跌 → 不做多

1.4 过滤维度

为全面覆盖配对交易的风险，过滤器需要在三个维度同时工作：

2. 算法选择与学术依据

2.0（新增）Layer 0：市场机制检测

动机：均值回归的前提条件

均值回归策略的根本假设是价格序列处于均值回归机制（mean-reverting regime），即 Hurst 指数 H < 0.5。当市场进入趋势机制（H > 0.6）时，无论 spread 如何扩张，继续做均值回归的逻辑前提已经失效——这是一个比"动量过强"更根本的风险，应作为最前置的硬拦截。

选择：DFA（去趋势波动分析）估算 Hurst 指数

候选算法深度对比：

学术背景：

DFA 由 Peng et al. (1994) 在 DNA 序列分析中提出，后被 Mantegna & Stanley (1995) 引入金融时间序列。相比经典 R/S 法，DFA 对嵌入趋势具有天然的鲁棒性，是目前金融计量中估算 Hurst 指数的最优方法（Peters 1994 对比研究）。

核心公式：

输入：最近 hurst_lookback 根 K 线的收盘价序列 {close[i]}

步骤 1：计算均值中心化对数收益率的累积和（DFA profile）
  ret[i] = ln(close[i] / close[i-1])
  profile[k] = sum(ret[i] - mean(ret)) for i=1..k

步骤 2：多尺度去趋势 RMS 波动
  对每个尺度 s = 4, 6, 8, ..., N/4：
    将 profile 分为 N/s 个不重叠窗口
    在每个窗口内做线性去趋势（OLS 拟合）
    计算残差 RMS：F(s) = sqrt(mean(residuals²))

步骤 3：log-log 线性回归
  Hurst = slope of log(F(s)) vs log(s)

触发条件（Layer 0 硬拦截）：

任一腿的 Hurst(DFA) > hurst_threshold（默认 0.60）
  → 趋势机制已激活，均值回归假设失效
  → 直接拒绝（硬拦截，不可被 Layer 3 仲裁推翻）

为什么 H > 0.6 选为阈值：

• H = 0.5：随机游走（无记忆，均值回归和趋势都不成立）
• H < 0.5：反持续（均值回归，配对策略最理想状态）
• H = 0.6：弱趋势，均值回归仍可能有效，不应过于激进拦截
• H > 0.6：显著趋势，均值回归假设明确失效，应拦截

2.1 约束 1&2：持续单边行情过滤（Layer 1）

候选算法深度对比

选择：Kaufman 效率比率 + 自适应净位移（RS 波动率驱动，非对称阈值）

学术背景：

Perry Kaufman 在 Smarter Trading (1995) 中提出 Efficiency Ratio，后被广泛应用于 KAMA（Kaufman Adaptive Moving Average）中。ER 是唯一同时衡量方向性和路径效率的单一指标，复杂度 O(N)，无需预热统计分布。

v3.0 改进：非对称 ER 阈值

加密市场存在结构性非对称性：上涨由散户 FOMO 驱动，趋势延续性更强；下跌由恐慌清算驱动，往往更急但反弹更快。因此：

• er_threshold_long = 0.60：拦截做多方向时更严格（上涨趋势延续性强，均值回归机会少）
• er_threshold_short = 0.50：拦截做空方向时稍宽松（下跌后反弹较快）

核心公式：

N = sustained_lookback（回望根数，默认 30 根 5min K 线 = 150 分钟）

# Kaufman Efficiency Ratio：方向距离 / 路径总长度
direction_distance = abs(close[-1] - close[-N])
path_length = sum(abs(close[i] - close[i-1]) for i in [-N+1, ..., -1])
ER = direction_distance / path_length    # 范围 [0, 1]

# 净位移（带符号）
net_return = (close[-1] - close[-N]) / close[-N]

# 自适应阈值：用 Rogers-Satchell 波动率标准化（per-symbol 基准）
rs_vol = per-symbol 滚动中位数基准（见 2.2 节）
sustained_threshold = base_threshold * (rs_vol_current / rs_vol_baseline)
                     （限制缩放范围：0.3x ~ 3.0x）

# 联合触发条件（非对称）：
约束1 触发：direction == 'short'
            AND net_return <= -sustained_threshold
            AND ER >= er_threshold_short（= 0.50）

约束2 触发：direction == 'long'
            AND net_return >= +sustained_threshold
            AND ER >= er_threshold_long（= 0.60）

2.2 约束 3&4：急涨急跌过滤（Layer 2）

候选算法深度对比

v3.0 关键改进：Rogers-Satchell 替代 Garman-Klass

学术依据：

Garman-Klass (1980) 波动率估计器的理论假设是零漂移（zero drift），即价格序列的期望收益率为零。但加密市场的趋势行情中漂移不为零，此时 GK 的修正项 -(2ln2-1)*ln(C/O)² 会引入系统偏差。

Rogers-Satchell (1991) 解决了这个问题：

RS(i) = ln(H/C) * ln(H/O) + ln(L/C) * ln(L/O)

性质：
  - 完全漂移不变（drift-invariant），不假设零期望收益
  - 单根 K 线恒 >= 0（H>=C, H>=O, L<=C, L<=O 保证两项均非负）
  - 理论效率 ≈ close-to-close 的 6.2 倍（vs GK 的 7.4 倍，效率略低但无偏）
  - 特别适合 24/7 加密市场（无隔夜跳空，RS 直接优于 Yang-Zhang）

v3.0 改进：Per-symbol 基准波动率（彻底解决跨品种污染）

v2.0 中 baseline_vol 是所有 symbol 的 RS 值混入同一个 500 样本池取中位数，导致 BTC（低波动）和小市值 alt（高波动）的自适应缩放相互污染。

v3.0 方案：每个 symbol 独立维护 200 样本的 RS 滚动中位数：

# 每个 symbol 独立的基准
per_symbol_baseline[symbol] = rolling_median(RS_history[symbol], maxlen=200)
# 自适应阈值
adaptive_thresh = base_thresh * (rs_vol_current / per_symbol_baseline[symbol])

这样 BTC 的阈值相对于 BTC 自身历史波动率缩放，alt 相对于 alt 自身缩放，完全解耦。

v3.0 改进：非对称 CUSUM 阈值

# 双侧 CUSUM（Rogers-Satchell 波动率标准化）
ret(i) = (close(i) - close(i-1)) / close(i-1)
z(i)   = ret(i) / rs_vol          # RS 波动率标准化

S_pos(i) = max(0, S_pos(i-1) + z(i) - drift)    # 检测正向急动（暴涨）
S_neg(i) = max(0, S_neg(i-1) - z(i) - drift)    # 检测负向急动（暴跌）

# 非对称触发阈值（反映上下方动量差异）
# 暴涨检测：加密市场 FOMO 强，轧空风险高 → 更敏感
spike_up   = S_pos >= cusum_threshold_spike_up    # 默认 3.5（更严格）
# 暴跌检测：清算后反弹快 → 稍宽松
spike_down = S_neg >= cusum_threshold_spike_down  # 默认 2.5（更宽松）

# 量价确认
volume_ratio = current_volume / EMA(volume, period=20)
volume_confirmed = volume_ratio >= 1.5

约束3 触发：direction == 'short'
            AND spike_up == True
            AND volume_confirmed == True

约束4 触发：direction == 'long'
            AND spike_down == True
            AND volume_confirmed == True

CUSUM 隐式冷却期（显式记录于此）：

drift = 0.5，threshold = 3.0 时，触发后 CUSUM 以每根 K 线 -drift = -0.5 的速率衰减。
从阈值 3.0 衰减到 0 需约 6 根 K 线（30 分钟）。这个隐式冷却期是过滤器的特性而非缺陷——急动后的冷却期正是我们需要的保护。

隐式冷却期 ≈ threshold / drift × bar_interval
           ≈ 3.5 / 0.5 × 5min = 35min（spike_up）
           ≈ 2.5 / 0.5 × 5min = 25min（spike_down）

2.3 Spread 层面趋势检测（Layer 3）

v3.0 改进：ADF + ER 双重仲裁（替代纯 ER）

学术背景：

v2.0 中 Layer 3 仅使用 ER 判断 spread 趋势性，是一个启发式指标，不能直接给出"spread 是否平稳"的统计结论。而 Layer 3 的本质问题正是：z4h 序列当前是否脱离了平稳（均值回归）状态？

最直接的答案来自增广 Dickey-Fuller（ADF）检验：

• H0：序列有单位根（非平稳，趋势/随机游走）
• H1：序列平稳（均值回归）
• p < 0.05 → 拒绝 H0 → spread 平稳 → ER 的趋势判断是误判 → 推翻 Layer 1 拦截 → 放行
• p > 0.10 → 不能拒绝 H0 → spread 可能非平稳 → 配合 ER 确认 → 维持拦截

双重确认逻辑（ER 快速筛 + ADF 精确验证）：

# 复用 z4h 序列作为 spread 代理
z4h_series = [z4h[-N], ..., z4h[-1]]    # 最近 N 根 z4h 值（N=spread_lookback）

# 第一关：ER 快速检查（O(N)，无需统计检验）
spread_dir = abs(z4h[-1] - z4h[-N])
spread_path = sum(abs(z4h[i] - z4h[i-1]))
spread_er = spread_dir / spread_path
spread_net = z4h[-1] - z4h[-N]

# ER 未达阈值 → 路径曲折 → 直接判定无趋势 → 放行（不调用 ADF，节省计算）
if spread_er < spread_er_threshold:
    → 无趋势，推翻拦截，放行

# 第二关：ADF 平稳性检验（仅在 ER 认为有趋势时执行）
adf_pvalue = adfuller(z4h_series, maxlag=2, regression='c')[1]

if adf_pvalue < 0.05:
    → ADF 判定平稳，推翻 ER 趋势判断，放行
    # 说明：ER 高但 ADF 说平稳，可能是短期偶发的高效路径，不是真趋势

if adf_pvalue > 0.10 AND spread_er >= threshold AND |spread_net| >= net_threshold:
    → ER 趋势 + ADF 非平稳，双重确认 spread 趋势 → 维持拦截

# 无 statsmodels 或样本不足（< 15）：仅凭 ER 判断（降级模式）

为什么 ADF 需要 ER 前置筛选：

1. ADF 计算有额外开销（statsmodels），每次信号都调用会浪费资源
2. ER 是 O(N) 的快速初步筛选，绝大多数情况 ER < threshold 时直接放行，无需 ADF
3. 只有"ER 认为有趋势"时才调用 ADF 做精确验证，兼顾性能和精度

2.4 算法组合架构（四层）

输入信号（direction, alt_symbol, base_symbol）
    │
    ├──▶ Layer 0: 市场机制检测（DFA Hurst 指数）   ← v3.0 新增
    │      ├── alt 腿: Hurst(alt) > 0.60?
    │      └── base 腿: Hurst(base) > 0.60?
    │      任一腿触发 → 直接拒绝（硬拦截，均值回归假设失效）
    │
    ├──▶ Layer 1: 单腿持续趋势过滤（ER + RS 自适应净位移，非对称阈值）
    │      ├── alt 腿: check(alt, direction)
    │      └── base 腿: check(base, opposite_direction)
    │      任一腿被拦截 → 进入 Layer 3 仲裁（软拦截）
    │
    ├──▶ Layer 2: 急动检测（CUSUM + Rogers-Satchell + 量价确认，非对称阈值）
    │      ├── alt 腿: check(alt, direction)
    │      └── base 腿: check(base, opposite_direction)
    │      任一腿被拦截 → 直接拒绝（硬拦截，执行风险不可对冲）
    │
    └──▶ Layer 3: Spread ADF + ER 双重仲裁        ← v3.0 升级
           仅在 Layer 0/1 软拦截时启动
           ER < threshold → 直接放行（O(N) 快速路径）
           ER >= threshold + ADF p < 0.05 → 推翻拦截，放行
           ER >= threshold + ADF p > 0.10 → 维持拦截
    │
    ▼
  最终决策：允许 / 拒绝入场

设计理念（v3.0 修订）：

3. 算法具体实现

3.1 模块位置

src/trading/
  momentum_filter.py    ← 新建（独立模块，不耦合策略逻辑）
  strategy.py           ← 修改 _check_entry()，注入并调用 filter；SymbolBaseline 新增 z4h_history
  config.py             ← 修改 StrategyParams，新增过滤器参数（含新增参数）
  orchestrator.py       ← 修改 process_analysis()，透传 OHLCV + volume
src/services/
  realtime_kline_service_base.py  ← 修改 _trigger_strategy_if_ready()，提取双腿 OHLCV + volume

3.2 MomentumFilter 完整代码（v3.0）

# src/trading/momentum_filter.py
"""
开仓动量过滤器 v3.0

四层过滤架构：
  Layer 0: Hurst DFA               → 趋势机制检测（硬拦截）     [v3.0 新增]
  Layer 1: ER + RS 自适应净位移    → 持续趋势检测（软拦截）
  Layer 2: CUSUM + RS + 量价确认   → 急动检测（硬拦截）
  Layer 3: Spread ADF + ER 双重仲裁 → 误杀修正（仅软拦截时启动）[v3.0 升级]

v3.0 改进清单：
  1. [Layer 0] 新增 Hurst DFA 趋势机制前置检测
  2. [Layer 1/2] Rogers-Satchell 替代 Garman-Klass（漂移不变）
  3. [基准波动率] Per-symbol 滚动中位数，消除跨品种污染
  4. [非对称参数] long/short 独立 ER 阈值 + CUSUM 阈值
  5. [Layer 3] 滚动 ADF + ER 双重验证（统计上更严格）
  6. [日志] 多腿同时 Layer 1 拦截时合并全部 reason，不再被覆盖
  7. [接口] check() 返回 (bool, str, bool) 替代 "SOFT|" 前缀约定
"""

import math
from collections import deque
from datetime import datetime

try:
    from statsmodels.tsa.stattools import adfuller as _adfuller
    _HAS_STATSMODELS = True
except ImportError:
    _HAS_STATSMODELS = False


class _RollingMedian:
    """固定窗口滚动中位数（maxlen <= 200，O(N log N) 可接受，5min 级别更新无性能压力）"""

    __slots__ = ('_data',)

    def __init__(self, maxlen: int):
        self._data: deque = deque(maxlen=maxlen)

    def push(self, val: float) -> None:
        self._data.append(val)

    @property
    def value(self) -> float | None:
        if not self._data:
            return None
        s = sorted(self._data)
        n = len(s)
        mid = n >> 1
        return s[mid] if n & 1 else (s[mid - 1] + s[mid]) * 0.5

    def __len__(self) -> int:
        return len(self._data)


def _hurst_dfa(closes: list[float]) -> float | None:
    """
    Detrended Fluctuation Analysis（去趋势波动分析）估算 Hurst 指数。

    算法步骤：
      1. 计算对数收益率均值中心化后的累积和（profile）
      2. 多尺度分段线性去趋势，计算各尺度 RMS 波动 F(s)
      3. log(F(s)) 对 log(s) 线性回归，斜率 = Hurst 指数

    返回值：
      H in [0, 1]，数据不足时返回 None（调用方应放行）
      H < 0.5: 均值回归（有利于入场）
      H ≈ 0.5: 随机游走
      H > 0.5: 趋势持续（危险，均值回归假设失效）
    """
    n = len(closes)
    if n < 20:
        return None

    # 计算对数收益率
    rets = []
    for i in range(1, n):
        if closes[i] > 0 and closes[i - 1] > 0:
            rets.append(math.log(closes[i] / closes[i - 1]))
    nr = len(rets)
    if nr < 16:
        return None

    mean_r = sum(rets) / nr

    # 构建均值中心化累积和（DFA profile）
    profile = []
    cum = 0.0
    for r in rets:
        cum += r - mean_r
        profile.append(cum)

    # 多尺度 RMS 波动率计算
    scales: list[float] = []
    rms_vals: list[float] = []

    s = 4
    max_s = nr // 4
    while s <= max_s:
        n_win = nr // s
        if n_win < 2:
            break

        rss_sum = 0.0
        rss_cnt = 0
        for i in range(n_win):
            seg = profile[i * s: (i + 1) * s]
            ws = len(seg)
            if ws < 2:
                continue
            # 线性去趋势（最小二乘）
            x_m = (ws - 1) * 0.5
            y_m = sum(seg) / ws
            cov = sum((j - x_m) * (seg[j] - y_m) for j in range(ws))
            var_x = sum((j - x_m) ** 2 for j in range(ws))
            slope = cov / var_x if var_x > 1e-12 else 0.0
            intercept = y_m - slope * x_m
            rss = sum((seg[j] - (intercept + slope * j)) ** 2 for j in range(ws)) / ws
            rss_sum += rss
            rss_cnt += 1

        if rss_cnt >= 2:
            f_s = math.sqrt(rss_sum / rss_cnt)
            if f_s > 1e-14:
                scales.append(math.log(s))
                rms_vals.append(math.log(f_s))

        # 对数均匀采样尺度
        s = max(s + 1, int(s * 1.4 + 0.5))

    if len(scales) < 3:
        return None

    # log-log 线性回归斜率 = Hurst 指数
    n_pts = len(scales)
    x_m = sum(scales) / n_pts
    y_m = sum(rms_vals) / n_pts
    num = sum((scales[i] - x_m) * (rms_vals[i] - y_m) for i in range(n_pts))
    den = sum((scales[i] - x_m) ** 2 for i in range(n_pts))
    if den < 1e-12:
        return None

    return max(0.0, min(1.0, num / den))


class MomentumFilter:
    """
    开仓动量过滤器 v3.0 — 四层架构

    公开接口：
      update(symbol, close, high, low, open_, volume, kline_time) — 每根 K 线调用一次
      check(symbol, direction) -> (allowed, reason, is_soft)
      check_spread(z4h_history, direction) -> (has_trend, reason)
      ready(symbol) -> bool
    """

    def __init__(
        self,
        # ── 总开关 ──
        enabled: bool = True,
        # ── Layer 0: Hurst DFA ──
        hurst_enabled: bool = True,
        hurst_lookback: int = 60,
        hurst_threshold: float = 0.60,
        # ── Layer 1: 持续趋势 ──
        sustained_lookback: int = 30,
        sustained_base_threshold: float = 0.008,
        er_threshold_long: float = 0.60,    # long 方向更严格
        er_threshold_short: float = 0.50,   # short 方向稍宽松
        # ── Layer 2: 急动检测 ──
        rs_period: int = 10,                # Rogers-Satchell EMA 周期
        cusum_drift: float = 0.5,
        cusum_threshold_spike_up: float = 3.5,    # 暴涨检测（拦截 short）
        cusum_threshold_spike_down: float = 2.5,  # 暴跌检测（拦截 long）
        volume_confirm_ratio: float = 1.5,
        volume_ema_period: int = 20,
        # ── Layer 3: Spread 仲裁 ──
        spread_lookback: int = 20,
        spread_er_threshold: float = 0.45,
        spread_net_threshold: float = 1.5,
        spread_adf_pvalue: float = 0.10,    # ADF 显著性水平
    ):
        self._enabled = enabled

        # Layer 0
        self._hurst_enabled = hurst_enabled
        self._hurst_lookback = max(20, hurst_lookback)
        self._hurst_thresh = hurst_threshold

        # Layer 1
        self._sustained_n = max(5, sustained_lookback)
        self._sustained_base_thresh = sustained_base_threshold
        self._er_thresh_long = er_threshold_long
        self._er_thresh_short = er_threshold_short

        # Layer 2
        self._rs_period = rs_period
        self._cusum_drift = cusum_drift
        self._cusum_thresh_up = cusum_threshold_spike_up
        self._cusum_thresh_down = cusum_threshold_spike_down
        self._vol_confirm_ratio = volume_confirm_ratio
        self._vol_ema_period = volume_ema_period

        # Layer 3
        self._spread_lookback = spread_lookback
        self._spread_er_thresh = spread_er_threshold
        self._spread_net_thresh = spread_net_threshold
        self._spread_adf_pvalue = spread_adf_pvalue

        # 数据缓冲区大小
        buf_size = max(
            hurst_lookback + 5,
            sustained_lookback + 5,
            rs_period + 10,
            volume_ema_period + 5,
        ) + 10
        self._buf_size = buf_size

        # symbol → deque of (close, high, low, open_, volume)
        self._buffers: dict[str, deque] = {}
        # symbol → 最后一次更新的 kline_time（去重）
        self._last_kline_time: dict[str, datetime] = {}
        # symbol → CUSUM 状态 (S_pos, S_neg)
        self._cusum_state: dict[str, tuple[float, float]] = {}
        # symbol → Rogers-Satchell 波动率 EMA（方差形式）
        self._rs_ema: dict[str, float] = {}
        # symbol → Volume EMA
        self._vol_ema: dict[str, float] = {}
        # symbol → Per-symbol 基准波动率滚动中位数（maxlen=200，不再全局混合）
        self._baseline_median: dict[str, _RollingMedian] = {}

    # ------------------------------------------------------------------
    # 公开接口
    # ------------------------------------------------------------------

    def update(
        self,
        symbol: str,
        close: float,
        high: float = 0.0,
        low: float = 0.0,
        open_: float = 0.0,
        volume: float = 0.0,
        kline_time: datetime | None = None,
    ) -> None:
        """
        每根新 K 线收盘时调用。
        high/low/open_ 可选；有则 Rogers-Satchell 更精确，无则降级为 close-to-close。
        volume 可选；有则启用量价确认，无则跳过量确认。
        kline_time 用于去重：同一 symbol 多配对时只追加一次。
        """
        if kline_time is not None:
            if self._last_kline_time.get(symbol) == kline_time:
                return
            self._last_kline_time[symbol] = kline_time

        if symbol not in self._buffers:
            self._buffers[symbol] = deque(maxlen=self._buf_size)
        self._buffers[symbol].append((close, high, low, open_, volume))

        buf = self._buffers[symbol]
        if len(buf) >= 2:
            self._online_update(symbol, buf)

    def check(self, symbol: str, direction: str) -> tuple[bool, str, bool]:
        """
        Layer 0 + Layer 2 + Layer 1 检查（单腿维度）。

        Args:
            symbol:    目标 symbol（单腿，非配对）
            direction: 'long' 或 'short'

        Returns:
            (allowed, reason, is_soft)
            allowed=False + is_soft=False → 硬拦截（Layer 0 / Layer 2），不可被 Layer 3 推翻
            allowed=False + is_soft=True  → 软拦截（Layer 1），可由 Layer 3 Spread 仲裁推翻
        """
        if not self._enabled:
            return True, "", False

        buf = self._buffers.get(symbol)
        if buf is None or len(buf) < self._rs_period + 2:
            return True, "", False  # 数据不足，安全放行

        # ── Layer 0（硬拦截）：Hurst DFA 趋势机制检测 ──
        ok, reason = self._check_hurst(symbol)
        if not ok:
            return False, reason, False

        # ── Layer 2（硬拦截）：急动检测（优先于 Layer 1） ──
        ok, reason = self._check_spike(symbol, direction)
        if not ok:
            return False, reason, False

        # ── Layer 1（软拦截）：持续趋势 ──
        if len(buf) >= self._sustained_n + 2:
            ok, reason = self._check_sustained(symbol, direction)
            if not ok:
                return False, reason, True  # 软拦截，可被仲裁

        return True, "", False

    def check_spread(self, z4h_history: list[float], direction: str) -> tuple[bool, str]:
        """
        Layer 3: Spread ADF + ER 双重仲裁。
        仅在存在软拦截（Layer 0 / Layer 1）时调用。

        Returns:
            (has_trend, reason)
            has_trend=True  → Spread 有趋势 → 维持拦截
            has_trend=False → Spread 无趋势 → 推翻拦截，放行
        """
        n = self._spread_lookback
        if len(z4h_history) < n + 1:
            return False, ""  # 数据不足，默认无趋势 → 放行

        series = z4h_history[-(n + 1):]

        # ── 第一关：ER 快速筛选（O(N)，大多数情况到这里结束） ──
        spread_dir = abs(series[-1] - series[0])
        spread_path = sum(abs(series[i] - series[i - 1]) for i in range(1, len(series)))
        er = spread_dir / spread_path if spread_path > 1e-10 else 0.0
        spread_net = series[-1] - series[0]

        # 方向性检查
        if direction == 'short':
            dir_ok = spread_net <= -self._spread_net_thresh
        else:
            dir_ok = spread_net >= self._spread_net_thresh

        er_signals_trend = er >= self._spread_er_thresh and dir_ok
        if not er_signals_trend:
            return False, ""  # ER 认为无趋势 → 跳过 ADF，直接放行

        # ── 第二关：ADF 平稳性检验（精确验证，仅在 ER 认为有趋势时执行） ──
        if _HAS_STATSMODELS and len(series) >= 15:
            try:
                adf_result = _adfuller(series, maxlag=2, regression='c', autolag=None)
                adf_p = adf_result[1]

                if adf_p < 0.05:
                    # ADF 判定平稳 → 推翻 ER 的趋势判断 → 放行
                    return False, (
                        f"ADF判定平稳(p={adf_p:.3f}<0.05)，推翻ER趋势判断"
                    )

                # ER 趋势 + ADF 非平稳 → 双重确认 → 维持拦截
                return True, (
                    f"Spread趋势(双重确认): ER={er:.2f}>={self._spread_er_thresh}"
                    f" 净位移={spread_net:+.3f} ADF_p={adf_p:.3f}>{self._spread_adf_pvalue}"
                )
            except Exception:
                pass  # ADF 异常 → 降级为仅 ER 判断

        # 降级：无 statsmodels 或样本不足 → 仅凭 ER 判断
        return True, (
            f"Spread趋势(ER): ER={er:.2f}>={self._spread_er_thresh}"
            f" 净位移={spread_net:+.3f}"
        )

    def ready(self, symbol: str) -> bool:
        """该 symbol 是否已积累足够数据"""
        buf = self._buffers.get(symbol)
        if buf is None:
            return False
        return len(buf) >= self._rs_period + 2

    # ------------------------------------------------------------------
    # Layer 0: Hurst DFA（趋势机制检测）
    # ------------------------------------------------------------------

    def _check_hurst(self, symbol: str) -> tuple[bool, str]:
        """Layer 0 硬拦截：Hurst > threshold → 趋势机制，均值回归假设失效"""
        if not self._hurst_enabled:
            return True, ""

        buf = self._buffers.get(symbol)
        if buf is None or len(buf) < self._hurst_lookback + 1:
            return True, ""

        closes = [d[0] for d in list(buf)[-(self._hurst_lookback + 1):]]
        h = _hurst_dfa(closes)
        if h is None:
            return True, ""

        if h > self._hurst_thresh:
            return False, (
                f"Layer0-趋势机制: Hurst(DFA)={h:.3f}>{self._hurst_thresh}"
                f" → 均值回归假设失效"
            )
        return True, ""

    # ------------------------------------------------------------------
    # Layer 1: 持续趋势检测（ER + Rogers-Satchell 自适应净位移，非对称阈值）
    # ------------------------------------------------------------------

    def _check_sustained(self, symbol: str, direction: str) -> tuple[bool, str]:
        """Layer 1 软拦截：ER >= threshold AND |net_return| >= adaptive_threshold"""
        buf = self._buffers[symbol]
        n = self._sustained_n
        data = list(buf)[-(n + 1):]
        closes = [d[0] for d in data]

        if len(closes) < n + 1:
            return True, ""

        ref = closes[0]
        if ref <= 0:
            return True, ""

        net_return = (closes[-1] - ref) / ref

        # Kaufman Efficiency Ratio
        direction_dist = abs(closes[-1] - ref)
        path_length = sum(abs(closes[i] - closes[i - 1]) for i in range(1, len(closes)))
        er = direction_dist / path_length if path_length > 1e-10 else 0.0

        # 非对称 ER 阈值
        er_thresh = self._er_thresh_long if direction == 'long' else self._er_thresh_short
        if er < er_thresh:
            return True, ""  # 路径不够趋势化，放行

        # RS 自适应阈值（per-symbol 基准，不再全局混合）
        rs_var = self._rs_ema.get(symbol, 0.0)
        rs_vol = math.sqrt(max(0.0, rs_var))
        baseline = self._baseline_median.get(symbol)
        baseline_val = baseline.value if (baseline and len(baseline) >= 5) else None

        if rs_vol > 0 and baseline_val and baseline_val > 0:
            adaptive_thresh = self._sustained_base_thresh * (rs_vol / baseline_val)
            adaptive_thresh = max(
                self._sustained_base_thresh * 0.3,
                min(adaptive_thresh, self._sustained_base_thresh * 3.0),
            )
        else:
            adaptive_thresh = self._sustained_base_thresh

        if direction == 'short' and net_return <= -adaptive_thresh:
            return False, (
                f"Layer1-不追跌: 过去{n}根净跌幅={net_return:.2%}"
                f" 阈值=-{adaptive_thresh:.2%} ER={er:.2f}>={er_thresh:.2f}"
            )
        if direction == 'long' and net_return >= adaptive_thresh:
            return False, (
                f"Layer1-不追涨: 过去{n}根净涨幅={net_return:.2%}"
                f" 阈值=+{adaptive_thresh:.2%} ER={er:.2f}>={er_thresh:.2f}"
            )
        return True, ""

    # ------------------------------------------------------------------
    # Layer 2: 急动检测（CUSUM + Rogers-Satchell + 量价确认，非对称阈值）
    # ------------------------------------------------------------------

    def _check_spike(self, symbol: str, direction: str) -> tuple[bool, str]:
        """Layer 2 硬拦截：CUSUM 超阈值 + 量价确认"""
        s_pos, s_neg = self._cusum_state.get(symbol, (0.0, 0.0))

        buf = self._buffers[symbol]
        current_vol = buf[-1][4] if buf else 0.0
        vol_ema = self._vol_ema.get(symbol, 0.0)

        if vol_ema > 0 and current_vol > 0:
            vol_ratio = current_vol / vol_ema
            vol_confirmed = vol_ratio >= self._vol_confirm_ratio
        else:
            vol_confirmed = True  # 无成交量数据，仅依赖 CUSUM
            vol_ratio = 0.0

        rs_vol = math.sqrt(max(0.0, self._rs_ema.get(symbol, 0.0)))
        vol_str = f" 量比={vol_ratio:.1f}" if vol_ratio > 0 else ""

        if direction == 'short' and s_pos >= self._cusum_thresh_up and vol_confirmed:
            return False, (
                f"Layer2-暴涨不做空: CUSUM+={s_pos:.2f}>={self._cusum_thresh_up}"
                f" RS_vol={rs_vol:.4f}{vol_str}"
                f" (隐式冷却≈{self._cusum_thresh_up/self._cusum_drift*5:.0f}min)"
            )

        if direction == 'long' and s_neg >= self._cusum_thresh_down and vol_confirmed:
            return False, (
                f"Layer2-暴跌不做多: CUSUM-={s_neg:.2f}>={self._cusum_thresh_down}"
                f" RS_vol={rs_vol:.4f}{vol_str}"
                f" (隐式冷却≈{self._cusum_thresh_down/self._cusum_drift*5:.0f}min)"
            )

        return True, ""

    # ------------------------------------------------------------------
    # 在线更新（每根 K 线 O(1)）
    # ------------------------------------------------------------------

    def _online_update(self, symbol: str, buf: deque) -> None:
        """O(1) 在线更新 CUSUM、RS EMA、Volume EMA、Per-symbol 基准中位数"""
        curr = buf[-1]   # (close, high, low, open_, volume)
        prev = buf[-2]
        close, high, low, open_, volume = curr
        prev_close = prev[0]

        if prev_close <= 0 or close <= 0:
            return

        # ── Rogers-Satchell 波动率 EMA ──
        rs_val = self._calc_rs_single(high, low, open_, close)
        alpha_rs = 2.0 / (self._rs_period + 1)
        prev_rs = self._rs_ema.get(symbol, rs_val)
        new_rs = alpha_rs * rs_val + (1.0 - alpha_rs) * prev_rs
        self._rs_ema[symbol] = new_rs

        # ── Per-symbol 基准波动率（滚动中位数，200 样本，不再全局混合）──
        rs_vol_now = math.sqrt(max(0.0, new_rs))
        if symbol not in self._baseline_median:
            self._baseline_median[symbol] = _RollingMedian(maxlen=200)
        self._baseline_median[symbol].push(rs_vol_now)

        # ── CUSUM 更新（RS 波动率标准化）──
        rs_vol = math.sqrt(max(0.0, new_rs))
        if rs_vol > 1e-10:
            ret = (close - prev_close) / prev_close
            z = ret / rs_vol
            s_pos, s_neg = self._cusum_state.get(symbol, (0.0, 0.0))
            s_pos = max(0.0, s_pos + z - self._cusum_drift)
            s_neg = max(0.0, s_neg - z - self._cusum_drift)
            self._cusum_state[symbol] = (s_pos, s_neg)

        # ── Volume EMA 更新 ──
        if volume > 0:
            alpha_v = 2.0 / (self._vol_ema_period + 1)
            prev_v = self._vol_ema.get(symbol, volume)
            self._vol_ema[symbol] = alpha_v * volume + (1.0 - alpha_v) * prev_v

    @staticmethod
    def _calc_rs_single(high: float, low: float, open_: float, close: float) -> float:
        """
        Rogers-Satchell (1991) 波动率估计器（单根 K 线）。

        公式: RS = ln(H/C) * ln(H/O) + ln(L/C) * ln(L/O)

        特性：
          - 完全漂移不变（drift-invariant），加密趋势市场无偏
          - 恒 >= 0（H>=C,H>=O 保证第一项>=0；L<=C,L<=O 保证第二项>=0）
          - 理论效率：close-to-close 的 6.2 倍
          - 相比 GK（7.4 倍但有零漂移假设），RS 无偏且无需假设

        无 OHLC 时降级为 open-to-close 方差（仍比 close-to-close 好）。
        """
        if high > 0 and low > 0 and open_ > 0 and high >= low:
            try:
                rs = (
                    math.log(high / close) * math.log(high / open_)
                    + math.log(low / close) * math.log(low / open_)
                )
                return max(0.0, rs)
            except (ValueError, ZeroDivisionError):
                pass
        # 降级：open-to-close 方差
        if open_ > 0:
            try:
                return math.log(close / open_) ** 2
            except (ValueError, ZeroDivisionError):
                pass
        return 0.0

3.3 StrategyParams 新增字段（v3.0）

在 src/trading/config.py 的 StrategyParams 中新增（共 19 个新字段，全部有默认值，不破坏现有代码）：

@dataclass(frozen=True)
class StrategyParams:
    # ... 现有字段（8个必填 + 3个有默认值）...

    # ── 动量过滤器参数（v3.0） ──
    momentum_filter_enabled: bool = True

    # Layer 0: Hurst DFA（新增）
    momentum_hurst_enabled: bool = True
    momentum_hurst_lookback: int = 60           # 用于 Hurst 计算的回望根数
    momentum_hurst_threshold: float = 0.60      # H > 此值 → 趋势机制 → 硬拦截

    # Layer 1: 持续趋势（er_threshold 拆分为非对称）
    momentum_sustained_lookback: int = 30
    momentum_sustained_base_threshold: float = 0.008
    momentum_er_threshold_long: float = 0.60    # 做多方向（上涨趋势，更严格）
    momentum_er_threshold_short: float = 0.50   # 做空方向（下跌趋势，稍宽松）

    # Layer 2: 急动检测（gk_period → rs_period，cusum_threshold 拆分非对称）
    momentum_rs_period: int = 10                # Rogers-Satchell EMA 周期
    momentum_cusum_drift: float = 0.5
    momentum_cusum_threshold_spike_up: float = 3.5    # 暴涨（拦截 short，更敏感）
    momentum_cusum_threshold_spike_down: float = 2.5  # 暴跌（拦截 long，稍宽松）
    momentum_volume_confirm_ratio: float = 1.5
    momentum_volume_ema_period: int = 20

    # Layer 3: Spread 仲裁（新增 adf_pvalue）
    momentum_spread_lookback: int = 20
    momentum_spread_er_threshold: float = 0.45
    momentum_spread_net_threshold: float = 1.5
    momentum_spread_adf_pvalue: float = 0.10    # ADF 显著性水平

3.4 strategy.py 集成改动（v3.0）

SymbolBaseline 新增 z4h_history 字段（用于 Layer 3 Spread 仲裁）：

@dataclass
class SymbolBaseline:
    std_window: deque = field(default_factory=deque)
    welford_mean: float = 0.0
    welford_m2: float = 0.0
    welford_updates: int = 0
    ema: float | None = None
    last_kline_time: datetime | None = None
    last_std: float = 0.0
    z4h_history: deque = field(default_factory=deque)   # v3.0 新增

在 _get_baseline() 中同步初始化：

self._baselines[key] = SymbolBaseline(
    std_window=deque(maxlen=params.std_window),
    z4h_history=deque(maxlen=50),   # 50 个 z4h 值，供 Layer 3 使用
)

process_tick 新增 OHLCV 参数：

def process_tick(
    self,
    symbol: str,
    base_symbol: str,
    z4h: float,
    timestamp: datetime,
    kline_time: datetime | None = None,
    latest_price: float | None = None,
    alt_ohlcv: dict | None = None,     # v3.0 新增
    base_ohlcv: dict | None = None,    # v3.0 新增
) -> tuple[EntrySignal | None, ExitSignal | None]:

_process_tick_unlocked 新 K 线时同步更新过滤器和 z4h 历史：

if is_new_candle:
    # 原有 Welford + EMA 更新...

    # v3.0 新增：追加 z4h 历史（供 Layer 3 Spread 仲裁）
    bl.z4h_history.append(z4h)

    # v3.0 新增：更新两腿动量过滤器
    if self._momentum_filter:
        if alt_ohlcv:
            self._momentum_filter.update(
                symbol=symbol,
                close=alt_ohlcv.get('close', 0.0),
                high=alt_ohlcv.get('high', 0.0),
                low=alt_ohlcv.get('low', 0.0),
                open_=alt_ohlcv.get('open', 0.0),
                volume=alt_ohlcv.get('volume', 0.0),
                kline_time=kline_time,
            )
        if base_ohlcv and base_symbol:
            self._momentum_filter.update(
                symbol=base_symbol,
                close=base_ohlcv.get('close', 0.0),
                high=base_ohlcv.get('high', 0.0),
                low=base_ohlcv.get('low', 0.0),
                open_=base_ohlcv.get('open', 0.0),
                volume=base_ohlcv.get('volume', 0.0),
                kline_time=kline_time,
            )

_check_entry 四层动量过滤（steps 4.5–4.7，解决多腿 reason 覆盖问题）：

def _check_entry(self, key, z4h, adaptive_z, timestamp, current_above, params):
    # ... 步骤 1-4（冷却期、突破检测、持仓检查、z4h 过滤）...

    # 步骤 4.5：方向判断（提前，供动量过滤器使用）
    threshold = params.adaptive_threshold
    if adaptive_z < -threshold:
        direction = 'long'
    elif adaptive_z > threshold:
        direction = 'short'
    else:
        return None

    # 步骤 4.6：四层动量过滤
    if self._momentum_filter is not None:
        soft_reasons: list[str] = []   # 收集所有软拦截 reason（修复覆盖 bug）
        hard_blocked = False
        hard_reason_parts: list[str] = []

        # Alt 腿（主方向）
        if self._momentum_filter.ready(key[0]):
            allowed, reason, is_soft = self._momentum_filter.check(key[0], direction)
            if not allowed:
                if is_soft:
                    soft_reasons.append(f"alt:{reason}")
                else:
                    hard_blocked = True
                    hard_reason_parts.append(f"alt:{reason}")

        # Base 腿（反向）
        if key[1] and self._momentum_filter.ready(key[1]):
            opposite = 'short' if direction == 'long' else 'long'
            allowed, reason, is_soft = self._momentum_filter.check(key[1], opposite)
            if not allowed:
                if is_soft:
                    soft_reasons.append(f"base:{reason}")
                else:
                    hard_blocked = True
                    hard_reason_parts.append(f"base:{reason}")

        # 硬拦截（Layer 0 / Layer 2）→ 直接拒绝，不仲裁
        if hard_blocked:
            logger.info(f"🚫 动量过滤(硬) | {pair_label} | {' | '.join(hard_reason_parts)}")
            return None

        # 软拦截（Layer 0 / Layer 1）→ 进入 Layer 3 仲裁
        if soft_reasons:
            combined = " | ".join(soft_reasons)
            bl = self._baselines.get(key)
            z4h_hist = list(bl.z4h_history) if (bl and bl.z4h_history) else []

            spread_trend, spread_reason = self._momentum_filter.check_spread(z4h_hist, direction)
            if spread_trend:
                logger.info(
                    f"🚫 动量过滤(软+Spread) | {pair_label} | {combined} | {spread_reason}"
                )
                return None
            else:
                if spread_reason:
                    logger.info(
                        f"✅ 动量仲裁放行 | {pair_label} | 单腿:{combined} | {spread_reason}"
                    )
                else:
                    logger.info(
                        f"✅ 动量仲裁放行 | {pair_label} | 单腿:{combined} | Spread无趋势→推翻拦截"
                    )

    # 步骤 5：产生 EntrySignal（direction 已由步骤 4.5 确定）
    ...

v3.0 修复说明：v2.0 中 layer1_reason 是单一字符串，base 腿拦截时会覆盖 alt 腿的 reason。v3.0 改为 soft_reasons: list[str]，收集所有腿的拦截原因后统一输出，日志不再丢失信息。

3.5 改动范围（v3.0）

4. 融入当前交易体系的方案

4.1 数据流全景（v3.0）

WebSocket K线推送（OHLCV + volume）
      |
      v
realtime_kline_service_base
      |
      +-- _parse_kline() → kline dict (含 close/high/low/open/volume)
      |
      +-- _trigger_strategy_if_ready()
              |
              +-- 从 price_data_cache 提取最新 alt/base OHLCV + volume
              |     alt_ohlcv = {'close','high','low','open','volume'}  ← alt_klines_5m[0]
              |     base_ohlcv = {'close','high','low','open','volume'} ← base_klines_5m[0]
              |
              v
      TradingOrchestrator.process_analysis(
          ..., alt_ohlcv=alt_ohlcv, base_ohlcv=base_ohlcv
      )
              |
              v
      AdaptiveBollingerStrategy.process_tick(
          ..., alt_ohlcv=alt_ohlcv, base_ohlcv=base_ohlcv
      )
              |
              +-- [新K线] bl.z4h_history.append(z4h)          ← Layer 3 历史维护
              |
              +-- [新K线] MomentumFilter.update(alt, alt_ohlcv, kline_time)
              |           → O(1) 在线更新: RS EMA / CUSUM / Volume EMA / Per-symbol基准中位数
              |           MomentumFilter.update(base, base_ohlcv, kline_time)
              |           （kline_time 去重：同 symbol 多配对只追加一次）
              |
              +-- _check_exit() → ExitSignal
              |
              +-- _check_entry()
                      |
                      步骤1: 冷却期
                      步骤2: 突破检测（adaptive_z 首次穿越阈值）
                      步骤3: 持仓检查
                      步骤4: z4h 绝对值过滤
                 [v3.0 新增]
                      步骤4.5: 方向判断（提前，供过滤器使用）
                      步骤4.6: 四层动量过滤
                                ├── Layer 0（硬拦截）: Hurst DFA
                                │    ├── alt 腿: check → is_soft=False
                                │    └── base 腿: check → is_soft=False
                                │    任一触发 → 直接拒绝
                                │
                                ├── Layer 2（硬拦截）: CUSUM + RS + 量价确认
                                │    ├── alt 腿: check → is_soft=False
                                │    └── base 腿: check → is_soft=False
                                │    任一触发 → 直接拒绝
                                │
                                ├── Layer 1（软拦截）: ER + RS 净位移（非对称阈值）
                                │    ├── alt 腿: check → is_soft=True
                                │    └── base 腿: check → is_soft=True
                                │    任一触发 → 进入 Layer 3 仲裁
                                │    （多腿 reason 合并，不再覆盖）
                                │
                                └── Layer 3（仲裁）: Spread ADF + ER 双重
                                     ER < threshold → 直接放行（O(N) 快速路径）
                                     ADF p < 0.05  → ADF 判定平稳，放行
                                     两者均确认趋势 → 维持拦截
                      步骤5: 产生 EntrySignal

4.2 realtime_kline_service_base.py 改动

def _trigger_strategy_if_ready(self, ...):
    # 提取 alt/base 最新 OHLCV（5m K线，供动量过滤器使用）
    alt_ohlcv = None
    base_ohlcv = None
    period_key_5m = ('5m', '7d')
    if period_key_5m in price_data_cache:
        alt_klines_5m = price_data_cache[period_key_5m]['alt_klines']
        if alt_klines_5m:
            k = alt_klines_5m[0]  # ORDER BY time DESC，第一条最新
            latest_alt_price = k.get('close')
            alt_ohlcv = {
                'close': k.get('close', 0.0),
                'high': k.get('high', 0.0),
                'low': k.get('low', 0.0),
                'open': k.get('open', 0.0),
                'volume': k.get('volume', 0.0),
            }
        base_klines_5m = price_data_cache[period_key_5m].get('base_klines', [])
        if base_klines_5m:
            k = base_klines_5m[0]
            base_ohlcv = {
                'close': k.get('close', 0.0),
                'high': k.get('high', 0.0),
                'low': k.get('low', 0.0),
                'open': k.get('open', 0.0),
                'volume': k.get('volume', 0.0),
            }

    acted = self._trading_orchestrator.process_analysis(
        ...,
        alt_ohlcv=alt_ohlcv,    # v3.0 新增
        base_ohlcv=base_ohlcv,  # v3.0 新增
    )

4.3 线程安全

MomentumFilter 不持有锁，由 AdaptiveBollingerStrategy 的 _lock（已存在）统一保护。update() 和 check() 均在 _lock 持有范围内调用，无竞态风险。kline_time 去重在逻辑层面保证同一根 K 线不被重复追加。

4.4 日志格式（v3.0）

🚫 动量过滤(硬)   | BTC|ETH | alt:Layer0-趋势机制: Hurst(DFA)=0.67>0.60 → 均值回归假设失效
🚫 动量过滤(硬)   | SOL|BTC | alt:Layer2-暴涨不做空: CUSUM+=3.82>=3.5 RS_vol=0.0028 量比=2.3 (隐式冷却≈35min)
🚫 动量过滤(硬)   | ETH|BTC | base:Layer2-暴跌不做多: CUSUM-=2.61>=2.5 RS_vol=0.0031 量比=1.7 (隐式冷却≈25min) | alt:Layer1-不追涨: ...
🚫 动量过滤(软+Spread) | SOL|BTC | alt:Layer1-不追跌: 过去30根净跌幅=-1.23% 阈值=-0.65% ER=0.72>=0.50 | base:Layer1-不追涨: ... | Spread趋势(双重确认): ER=0.61 净位移=-2.1 ADF_p=0.18>0.10
✅ 动量仲裁放行   | ETH|BTC | 单腿:alt:Layer1-不追涨: 过去30根净涨幅=1.05% 阈值=+0.80% ER=0.68>=0.60 | ADF判定平稳(p=0.03<0.05)，推翻ER趋势判断
✅ 动量仲裁放行   | ETH|BTC | 单腿:alt:Layer1-不追涨: ... | Spread无趋势→推翻拦截

4.5 配置层集成（v3.0 新增参数）

# Layer 0: Hurst DFA（全新）
TRADING_MOMENTUM_HURST_ENABLED=true
TRADING_MOMENTUM_HURST_LOOKBACK=60
TRADING_MOMENTUM_HURST_THRESHOLD=0.60

# Layer 1: 持续趋势（ER 拆分为非对称）
TRADING_MOMENTUM_SUSTAINED_LOOKBACK=30
TRADING_MOMENTUM_SUSTAINED_BASE_THRESHOLD=0.008
TRADING_MOMENTUM_ER_THRESHOLD_LONG=0.60
TRADING_MOMENTUM_ER_THRESHOLD_SHORT=0.50

# Layer 2: 急动检测（GK → RS，CUSUM 拆分为非对称）
TRADING_MOMENTUM_RS_PERIOD=10
TRADING_MOMENTUM_CUSUM_DRIFT=0.5
TRADING_MOMENTUM_CUSUM_THRESHOLD_SPIKE_UP=3.5
TRADING_MOMENTUM_CUSUM_THRESHOLD_SPIKE_DOWN=2.5
TRADING_MOMENTUM_VOLUME_CONFIRM_RATIO=1.5

# Layer 3: Spread 仲裁（新增 ADF 参数）
TRADING_MOMENTUM_SPREAD_LOOKBACK=20
TRADING_MOMENTUM_SPREAD_ER_THRESHOLD=0.45
TRADING_MOMENTUM_SPREAD_NET_THRESHOLD=1.5
TRADING_MOMENTUM_SPREAD_ADF_PVALUE=0.10

# 针对高波动 symbol 示例
TRADING_STRATEGY_SOL_MOMENTUM_CUSUM_THRESHOLD_SPIKE_UP=4.0
TRADING_STRATEGY_BTC_MOMENTUM_HURST_THRESHOLD=0.65

5. 与当前交易风格的契合度分析

5.1 交易风格特征

5.2 契合度分析（v3.0）

高度契合之处：

1. Layer 0 Hurst 与策略哲学高度一致：均值回归策略本质上赌 Hurst < 0.5。Layer 0 在 Hurst > 0.6 时前置拦截，是对策略假设的直接验证，而不是经验性的动量过滤。

2. Rogers-Satchell 更适合加密市场：24/7 连续交易无隔夜跳空，RS 无漂移假设，对 FOMO 驱动的急速上涨和清算驱动的下跌估计更准确。

3. Per-symbol 基准消除了跨品种干扰：BTC（日波动率 2-3%）和小 alt（日波动率 8-15%）之间不再相互影响阈值缩放，自适应效果更精准。

4. 非对称参数反映加密特性：

   • 做空拦截阈值低（ER_short=0.50）：加密市场下跌往往过激，反弹快，均值回归机会多
   • 做多拦截阈值高（ER_long=0.60）：FOMO 驱动的上涨延续性强，均值回归机会少，需要更严格过滤
   • CUSUM_up > CUSUM_down：暴涨做空被轧空的风险大于暴跌做多接飞刀

5. ADF + ER 双重仲裁是最严格的统计保证：从"启发式 ER 判断"升级为"统计检验 + 启发式双重确认"，大幅减少 Layer 3 误判（false positive），保护更多有效的均值回归信号。

6. Layer 3 仲裁与配对哲学完美一致：两腿同向运动（ER 高）但 spread 仍然平稳（ADF 小 p 值）→ 放行，这正是配对交易的本质——个体的趋势不等于 spread 的趋势。

设计张力（有意为之的权衡）：

• Layer 0 用单腿 Hurst 做硬拦截，但单腿趋势不等于 spread 趋势（这也是 Layer 3 的设计原因）。v3.0 中 Layer 0 是硬拦截（不经 Layer 3 仲裁），这是保守的但正确的：Hurst 反映的是序列的长期记忆特性，即使 spread 暂时稳定，单腿的趋势持续性意味着更高的发散风险。

5.3 综合评分（v3.0 vs v2.0）

6. 与四项开单约束的对应关系

6.1 逐项验证

约束 1：连续下跌不追跌

Layer 1 检测：ER >= er_threshold_short（0.50）AND net_return <= -adaptive_threshold
被阻止：direction == 'short'
原因：路径高效地持续下跌（ER 高），跌幅充分释放，继续做空空间有限且反弹风险高
仲裁：Layer 3 检查 spread 趋势性（ADF + ER 双重）→ spread 平稳则推翻拦截放行

示例（5min，N=30，base_threshold=0.8%，RS 自适应后 threshold=0.65%）：

• 过去 150 分钟稳步从 100 跌到 99（ER=0.72 >= 0.50, 净跌 1%）→ Layer 1 拦截
  • Spread 同步趋势（ADF p=0.22 + ER=0.58）→ 双重确认，维持拦截 ✓
  • Spread 稳定（两腿同跌，ADF p=0.03）→ ADF 判定平稳，推翻拦截，放行 ✓
• 过去 150 分钟 V 形反转（100→96→99.2，ER=0.10 < 0.50）→ ER 不足，放行 ✓

约束 2：连续上涨不追涨

Layer 1 检测：ER >= er_threshold_long（0.60，更严格）AND net_return >= +adaptive_threshold
被阻止：direction == 'long'

示例：

• 过去 150 分钟稳步从 100 涨到 101.2（ER=0.85 >= 0.60, 净涨 1.2%）→ 拦截 → 仲裁
• 过去 150 分钟震荡漂移（ER=0.15 < 0.60, 净涨 0.9%）→ ER 不足，放行 ✓

约束 3：迅速暴涨不做空

Layer 2 检测：CUSUM_pos >= cusum_threshold_spike_up（3.5，更敏感）AND volume_ratio >= 1.5
被阻止：direction == 'short'
隐式冷却期：3.5 / 0.5 × 5min ≈ 35min
不可仲裁（硬拦截）

示例（RS_vol=0.3%，drift=0.5）：

• 连续 3 根各涨 0.8%（标准化 z=2.67 每根）→ CUSUM 累积到 6.5 → 拦截 ✓
• 单根暴涨 2.5%（z=8.3）→ CUSUM 一步跳到 7.8 → 拦截 ✓
• 单根涨 0.3% 但量极小（vol_ratio=0.3）→ CUSUM 低 + 量不足 → 放行 ✓

约束 4：迅速暴跌不做多

Layer 2 检测：CUSUM_neg >= cusum_threshold_spike_down（2.5，稍宽松）AND volume_ratio >= 1.5
被阻止：direction == 'long'
隐式冷却期：2.5 / 0.5 × 5min ≈ 25min
不可仲裁（硬拦截）

6.2 约束覆盖矩阵（v3.0）

四项约束完全覆盖，Layer 0 新增对均值回归前提条件的根本性验证。

7. 参数配置指南

7.1 基础参数推荐值

Layer 0：Hurst DFA（v3.0 新增）

hurst_threshold 调优：
  0.55：过于激进，均值回归行情下也可能触发（误杀多）
  0.60：标准值，H=0.6 是弱趋势/随机游走的分界线
  0.65+：宽松，只拦截强趋势（适合波动大、趋势明显的主流币）

hurst_lookback 调优：
  30：短期记忆检测（约 2.5 小时），对快速趋势切换敏感
  60：标准值（约 5 小时），兼顾稳定性和响应速度
  90+：中长期机制检测，更稳定但滞后

Layer 1：持续趋势过滤

er_threshold 非对称调优原则：
  er_threshold_long 应高于 er_threshold_short（上涨更难均值回归）
  推荐差值：0.08-0.12

  激进配置（拦截更多）：long=0.55, short=0.45
  标准配置：             long=0.60, short=0.50
  保守配置（放行更多）：  long=0.65, short=0.55

Layer 2：急动检测

非对称 CUSUM 阈值调优：
  spike_up > spike_down（暴涨做空风险 > 暴跌做多风险）
  推荐差值：0.5-1.5

  CUSUM 隐式冷却期：threshold/drift × 5min
    spike_up   3.5/0.5 × 5 = 35min
    spike_down 2.5/0.5 × 5 = 25min

  ARL_0（平均误报间隔）≈ exp(2*drift*threshold) / (2*drift²)
    drift=0.5, threshold=3.5 → ARL_0 ≈ 73 根（约 6 小时才会一次误报）
    drift=0.5, threshold=2.5 → ARL_0 ≈ 20 根（约 1.7 小时才会一次误报）

Layer 3：Spread 仲裁

ADF 参数调优：
  adf_pvalue=0.05：严格（只有 p>0.05 才认为非平稳 → 更多误杀被维持）
  adf_pvalue=0.10：标准值（平衡误杀和漏杀）
  adf_pvalue=0.15：宽松（更容易被 ADF 判定为平稳 → 更多误杀被推翻）

  注意：ADF 对小样本（<20）有尺寸扭曲问题。spread_lookback >= 20 时结果可靠。

7.2 回测验证建议

对比四组回测：
  A 组：无动量过滤（当前基线）
  B 组：v2.0（GK + 单一 ER 阈值 + ER-only Spread）
  C 组：v3.0（RS + DFA Hurst + 非对称阈值 + ADF+ER Spread）
  D 组：v3.0 关闭 Layer 0（评估 Hurst 贡献）

核心对比指标：
  - Sharpe Ratio（风险调整收益）
  - 最大连续亏损次数
  - 信号过滤率（被拦截信号 / 总信号）
  - 误杀率（被拦截后实际盈利的信号占比）
  - Layer 3 仲裁放行率（推翻 Layer 0/1 的比例）
  - 分时段表现差异（亚盘/欧盘/美盘）
  - Layer 0 触发频率（多少信号被 Hurst 拦截）

预期结果（相对 v2.0 的增量收益）：
  Layer 0 贡献：过滤趋势机制下的系统性亏损，预计降低最大回撤 10-15%
  非对称 ER 贡献：减少做空方向误杀约 15%（short 方向 ER 阈值降低）
  ADF 仲裁贡献：减少 Layer 3 误判约 40%（相比纯 ER 仲裁），保留更多有效信号
  Per-symbol 基准贡献：高低波动资产的自适应阈值更精准，减少跨品种噪音约 20%

8. 风险与局限性

8.1 已知局限

8.2 不适合的场景

• 极低流动性资产：RS 波动率因价格跳空失真，Hurst 因稀疏 K 线估计不准，需配合黑名单
• HIP-3 稀疏资产：volume 不可靠时 Layer 2 量价确认无法生效，仅靠 CUSUM 价格检测

8.3 后续迭代方向（v3.0 之后）

附录 A：算法学术参考

附录 B：v2.0 → v3.0 改进对照

文档结束