Beta 体制自适应过滤器设计方案

1. 问题背景

1.1 市场现象

当锚定物（BTC）回暖后，Alt 资产的 Beta 值会持续飙升：

正常态:   β ≈ 0.3 ~ 0.5   （BTC 涨 1%，Alt 涨 0.3~0.5%）
扩张态:   β ≈ 4 ~ 10       （BTC 涨 1%，Alt 涨 4~10%）

这不是价格的普通上涨，而是价差的持续过大 — 协整关系的阶段性破裂。

1.2 当前系统弱点

1.3 故障场景还原

时间线（无 Beta 体制过滤器）:
T+0h    BTC 开始回暖，β 从 0.5 开始上升
T+12h   β 实际已达 2.0，但 OLS(100期) β̂ 仍 ≈ 0.6
        → spread = log(alt) - 0.6×log(btc) 持续偏大
        → z4h 持续走正，adaptive_z 突破阈值
        → 系统产生做空信号（误判为均值回归机会）

T+24h   β 实际达 4.0，OLS β̂ ≈ 0.8
        → z4h 继续走高，系统再次尝试入场
        → 开仓即被 β 持续扩张吞噬利润

T+7d    β 实际达 8.0，OLS β̂ ≈ 2.0
        → 仍在追赶，z4h 仍然极端
        → 协整假设彻底失效，均值回归策略不成立

核心问题：系统用滞后 β 计算的 z4h 产生入场信号，实际上这些信号反映的是 β 正在变化，而非 spread 正在回归。

1.4 新方案对比时间线

同一故障场景下，Beta 体制过滤器如何介入：

时间线（有 Beta 体制过滤器）:
T+0h    BTC 开始回暖，β 从 0.5 开始上升
        β_short(20期)=0.55, β_long(100期)=0.50
        β_divergence=0.10 → STABLE, 正常交易

T+8h    β_short 开始偏离（短窗口更快响应）
        β_short=0.85, β_long=0.52
        β_divergence=0.63 → STABLE, 但 divergence 上升中

T+12h   β_short 明显领先 β_long
        β_short=1.20, β_long=0.55
        β_divergence=1.18, mean_div 近期均值≈0.90
        → EXPANDING, scale=1.10, 阈值 3.0→3.3
        → 系统产生做空信号，但因阈值提高被拦截 ✅

T+16h   β_divergence 持续走高
        β_short=1.80, β_long=0.58
        β_divergence=2.10, mean_div≈1.50
        → EXPANDING, scale=1.58, 阈值 3.0→4.75
        → 继续拦截入场 ✅

T+24h   β_divergence 超过硬拦截线
        β_short=3.50, β_long=0.65
        β_divergence=4.38, mean_div≈2.80
        → EXPANDING, hard_block=True
        → 完全禁止入场 ✅

T+7d+   β_long 逐渐跟上（100期窗口适应）
        β_short=8.20, β_long=6.50
        β_divergence=0.26, mean_div≈0.35
        → STABLE, 恢复正常交易

关键改善：T+12h 即检测到 β 偏离并提高入场门槛，比旧系统（无保护，T+12h 误入场）早 0h 拦截。整个 β 飙升期间零入场，避免了旧系统在 T+12h~T+7d 间的持续亏损。

2. 解决方案概述

2.1 核心思路

在分析层直接追踪 β 的多窗口偏离度，作为 β 体制切换的一手信号：

β_long  = OLS(100期) β     ← 已有，稳定但迟钝
β_short = OLS(20期) β      ← 新增，敏感快速响应
β_divergence = max(0, β_short - β_long) / max(|β_long|, 0.1)   ← β 上行偏离度

当 β_divergence 持续偏高 → β 正在快速上升 → 协整关系不可靠 → 拦截入场。

设计要点：

• 仅检测上行偏离：max(0, β_short - β_long) 确保 β 下降（Alt 回落、协整恢复中）不触发拦截
• 分母下限保护：max(|β_long|, 0.1) 防止低 β 资产对（如 β_long=0.05）时 divergence 爆炸

2.2 方案优势

2.3 架构集成

┌──────────────────────────────────────────────────────────────┐
│                    analysis_core.py                           │
│  calculate_cointegration_params_dual_window()                │
│  ├─ β_long  = OLS(beta_window=100)        [已有]              │
│  ├─ β_short = OLS(beta_short_window=20)   [新增]              │
│  └─ β_divergence = max(0, β_short-β_long)                    │
│                    / max(|β_long|, 0.1)    [新增]             │
└────────────────────────┬─────────────────────────────────────┘
                         │ β_divergence (4h 结果，每根新 4h K线更新一次)
                         ▼
┌──────────────────────────────────────────────────────────────┐
│             orchestrator.py                                   │
│  process_analysis()                                          │
│  └─ 从 multi_period_result 提取 β_divergence，传给 strategy   │
└────────────────────────┬─────────────────────────────────────┘
                         │ beta_divergence: float
                         ▼
┌──────────────────────────────────────────────────────────────┐
│             strategy.py                                       │
│  _BetaRegimeTracker                                          │
│  ├─ update(): 接收 β_divergence，去重后入缓冲                  │
│  │   └─ 仅当值变化 >0.001 时入队（4h 级去重）                  │
│  ├─ check():  判定体制状态 + 输出阈值缩放因子                   │
│  │   ├─ STABLE:    scale=1.0，正常交易                        │
│  │   └─ EXPANDING: scale>1 或硬拦截                           │
│  │                                                           │
│  _check_entry() 改动:                                        │
│  ├─ [新增] Beta 体制检查 → 硬拦截 / 动态缩放                   │
│  └─ threshold = adaptive_threshold × beta_regime_scale        │
└──────────────────────────────────────────────────────────────┘

3. 详细设计

3.1 分析层：β_short 计算

文件：src/config.py

BETA_WINDOW: int = 100        # [已有] OLS 长期 β 窗口
BETA_SHORT_WINDOW: int = 20   # [新增] OLS 短期 β 窗口（20×4h≈3.3天）

窗口选择依据：

• 20 期 × 4h = 80h ≈ 3.3 天，足以捕捉 β 快速变化
• 比 ZSCORE_WINDOW=30 短，确保对结构性变化更敏感
• 最少 20 个数据点满足 OLS 统计显著性要求

文件：src/utils/analysis/analysis_core.py

改动函数：calculate_cointegration_params_dual_window()

在现有 OLS(beta_window=100) 之后，新增短窗口 OLS：

def calculate_cointegration_params_dual_window(
    base_klines, alt_klines,
    beta_window=None, zscore_window=None,
    beta_short_window=None,        # [新增] 短窗口大小
):
    # ... 现有逻辑不变 ...
    # beta_ols = 现有的 β_long

    # ═══ 新增：短窗口 β 估计 ═══
    if beta_short_window is None:
        beta_short_window = BETA_SHORT_WINDOW

    beta_short = beta_ols          # 默认等于长期 β
    beta_divergence = 0.0

    if len(aligned) >= beta_short_window + 1:
        short_base = np.log(aligned['base'].iloc[-beta_short_window:])
        short_alt  = np.log(aligned['alt'].iloc[-beta_short_window:])
        X_short = sm.add_constant(short_base)
        model_short = sm.OLS(short_alt, X_short).fit()
        beta_short = model_short.params.iloc[1]

        # β 上行偏离度 = max(0, β_short - β_long) / max(|β_long|, 0.1)
        # 仅检测 β 上升（Alt 跑赢 BTC），β 下降不拦截
        # 分母下限 0.1 防止低 β 资产对的 divergence 爆炸
        beta_divergence = max(0.0, beta_short - beta_ols) / max(abs(beta_ols), 0.1)

    return {
        # ... 现有字段不变 ...
        'beta_short': beta_short,
        'beta_divergence': beta_divergence,
    }

计算开销：短窗口 OLS 仅用 20 个数据点，相比现有 100 期 OLS 可忽略不计。

β_divergence 语义

注意：

• β_divergence 是瞬时值，单次高值可能是噪声，需要在策略层跟踪其时间序列
• β_divergence ≥ 0（设计保证），β 下降时为 0，不会误拦截协整恢复中的交易机会

低 β 资产对的分母保护

向上传播

analyze_pair_advanced()：从 coint_new 结果中提取 beta_short 和 beta_divergence，放入 cointegration_new dict。

analyze_multi_period()：从 4h/60d 周期的 detail 中提取 beta_divergence，放入返回值：

return {
    # ... 现有字段 ...
    'beta_divergence': details.get(('4h', '60d'), {}).get('cointegration_new', {}).get('beta_divergence', 0.0),
}

只用 4h 周期的 β_divergence，因为：

• 4h 是协整分析和交易信号的主时间框架
• 5m/1h 周期的 β 波动大、噪声多，不适合体制判定

3.2 编排层：传递 β_divergence

文件：src/trading/orchestrator.py

改动函数：process_analysis()

从 multi_period_result 提取 beta_divergence，传给 strategy.process_tick()：

# 在调用 strategy.process_tick() 之前
beta_divergence = multi_period_result.get('beta_divergence')

entry_signal, exit_signal = self._strategy.process_tick(
    symbol, base_symbol, z4h, timestamp,
    kline_time=kline_time,
    latest_price=price_for_log,
    alt_ohlcv=alt_ohlcv,
    base_ohlcv=base_ohlcv,
    beta_divergence=beta_divergence,   # [新增]
)

无需改 process_analysis() 的方法签名 — multi_period_result 已经作为参数传入。

3.3 策略层：BetaRegimeTracker

文件：src/trading/config.py

StrategyParams 新增字段：

@dataclass(frozen=True)
class StrategyParams:
    # ... 现有字段 ...

    # Beta 体制自适应过滤器
    beta_regime_enabled: bool = True
    beta_regime_divergence_threshold: float = 0.8  # β_divergence 均值超此值 → 扩张
    beta_regime_hard_block_threshold: float = 2.0  # β_divergence 均值超此值 → 硬拦截
    beta_regime_scale_max: float = 2.0             # 阈值最大缩放倍数

环境变量约定：

TRADING_STRATEGY_BETA_REGIME_ENABLED=true
TRADING_STRATEGY_BETA_REGIME_DIVERGENCE_THRESHOLD=0.8
TRADING_STRATEGY_BETA_REGIME_HARD_BLOCK_THRESHOLD=2.0
TRADING_STRATEGY_BETA_REGIME_SCALE_MAX=2.0

# 币种级覆盖 (PURR)
TRADING_STRATEGY_PURR_BETA_REGIME_DIVERGENCE_THRESHOLD=1.0

文件：src/trading/strategy.py

3.3.1 _BetaRegimeState 数据类

@dataclass
class _BetaRegimeState:
    """Beta 体制检测结果"""
    regime: str              # 'stable' | 'expanding'
    mean_divergence: float   # β_divergence 滑动均值
    max_divergence: float    # β_divergence 近期最大值
    threshold_scale: float   # 阈值缩放因子 (>=1.0)
    hard_block: bool         # True = 双重确认硬拦截
    reason: str              # 诊断信息

3.3.2 _BetaRegimeTracker 类

class _BetaRegimeTracker:
    """Beta 体制自适应跟踪器

    跟踪 β_divergence 时间序列，检测协整 β 持续飙升。

    数据策略：
    - 每个 5m tick 接收一个 β_divergence 值
    - 去重：β_divergence 基于 4h K线计算，同一根 4h 内值不变
      通过比较与上次入队值的差异（>_DEDUP_EPSILON）实现去重
    - 实际入队频率 ≈ 每 4h 一次
    - 缓冲区 maxlen=50 → 50 根 4h K线 ≈ 8.3 天的独立观测

    判定逻辑：
    - mean(β_divergence) > hard_block_threshold  → 硬拦截
    - mean(β_divergence) > divergence_threshold  → 阈值缩放
    - 否则 → 正常交易
    """

    _BUFFER_SIZE = 50         # 50 × 4h = ~8.3天（去重后的独立观测）
    _MIN_SAMPLES = 3          # 至少 3 根 4h K线（12h）才开始判定
    _RECENT_WINDOW = 6        # 近期窗口 6 × 4h = 24h（用于检测加速）
    _DEDUP_EPSILON = 0.001    # β_divergence 变化小于此值视为同一根 4h K线

    def __init__(self, enabled: bool = True):
        self._enabled = enabled
        self._buffers: dict[PairKey, deque] = {}       # 每配对缓冲
        self._last_values: dict[PairKey, float] = {}   # 去重用：上次入队值

    def update(self, key: PairKey, beta_divergence: float) -> None:
        """每 5m tick 调用，内部去重确保仅新 4h 值入队"""
        if not self._enabled or beta_divergence is None:
            return

        # 去重：β_divergence 基于 4h OLS，同一根 4h 内值不变
        last = self._last_values.get(key)
        if last is not None and abs(beta_divergence - last) < self._DEDUP_EPSILON:
            return  # 同一根 4h K线，跳过

        if key not in self._buffers:
            self._buffers[key] = deque(maxlen=self._BUFFER_SIZE)
        self._buffers[key].append(beta_divergence)
        self._last_values[key] = beta_divergence

    def check(
        self,
        key: PairKey,
        divergence_threshold: float,
        hard_block_threshold: float,
        scale_max: float,
    ) -> _BetaRegimeState:
        """检测当前 Beta 体制状态

        Args:
            key: 配对标识
            divergence_threshold: β_divergence 均值扩张阈值
            hard_block_threshold: β_divergence 均值硬拦截阈值
            scale_max: 最大阈值缩放倍数

        Returns:
            _BetaRegimeState
        """
        if not self._enabled:
            return _BetaRegimeState('stable', 0.0, 0.0, 1.0, False, "")

        buf = self._buffers.get(key)
        if buf is None or len(buf) < self._MIN_SAMPLES:
            return _BetaRegimeState('stable', 0.0, 0.0, 1.0, False, "数据不足")

        data = list(buf)

        # 全窗口均值
        mean_div = sum(data) / len(data)
        max_div = max(data)

        # 近期窗口均值（检测加速）
        recent = data[-self._RECENT_WINDOW:] if len(data) >= self._RECENT_WINDOW else data
        recent_mean = sum(recent) / len(recent)

        # 取 max(全窗口均值, 近期均值) 作为判定值
        # 这样即使长期均值还不高，但近期急剧升高也能触发
        effective_div = max(mean_div, recent_mean)

        # ── 判定 ──
        if effective_div >= hard_block_threshold:
            return _BetaRegimeState(
                'expanding', mean_div, max_div, scale_max, True,
                f"Beta硬拦截: mean_div={mean_div:.3f} recent={recent_mean:.3f} "
                f"max={max_div:.3f} >={hard_block_threshold}"
            )

        if effective_div >= divergence_threshold:
            # 线性插值缩放：divergence_threshold → 1.0, hard_block_threshold → scale_max
            t = (effective_div - divergence_threshold) / max(
                hard_block_threshold - divergence_threshold, 1e-6
            )
            t = min(1.0, max(0.0, t))
            scale = 1.0 + t * (scale_max - 1.0)
            return _BetaRegimeState(
                'expanding', mean_div, max_div, scale, False,
                f"Beta缩放: mean_div={mean_div:.3f} recent={recent_mean:.3f} "
                f"max={max_div:.3f} scale={scale:.2f}"
            )

        return _BetaRegimeState(
            'stable', mean_div, max_div, 1.0, False,
            f"Beta稳定: mean_div={mean_div:.3f} recent={recent_mean:.3f}"
        )

    def cleanup_pair(self, key: PairKey) -> None:
        self._buffers.pop(key, None)
        self._last_values.pop(key, None)

关键设计决策：

1. 4h 级去重：

   • β_divergence 基于 4h K线 OLS 计算，5m tick 内值不变
   • update() 通过 _DEDUP_EPSILON=0.001 检测值变化，仅在新 4h K线时入队
   • 避免 buffer 被 48 个相同值填满（4h/5m=48）
   • buffer 中每个元素对应一根独立的 4h K线，统计计算有效

2. 双窗口判定：max(全窗口均值, 近期24h均值)

   • 防止"短期急升但长期均值还低"的漏检
   • 近期窗口(6×4h=24h)对应 β 开始急速飙升的典型时间尺度

3. 线性插值缩放：

   divergence_threshold(0.8) → scale=1.0（刚开始偏离，不缩放）
   hard_block_threshold(2.0) → scale=2.0（最大缩放）
   中间值按线性插值
   

   避免硬切换，平滑过渡。

4. 硬拦截 vs 缩放：

   • effective_div >= 2.0（默认）→ 硬拦截（β 已剧烈偏移，协整假设彻底失效）
   • 0.8 <= effective_div < 2.0 → 阈值缩放（β 在变化但可能是暂时的，提高门槛）
   • effective_div < 0.8 → 正常交易

5. scale_max=2.0 的选择（而非 3.0）：

   • scale_max=3.0 时阈值从 3.0→9.0，adaptive_z 几乎不可能达到，整个缩放区间实质等同硬拦截
   • scale_max=2.0 时阈值从 3.0→6.0，z4h=5~6 在极端行情下仍可能出现
   • 缩放区间（0.8~2.0）中的中间值真正有意义：门槛提高但不封死

3.3.3 集成到 AdaptiveBollingerStrategy

__init__：

self._beta_regime = _BetaRegimeTracker(
    enabled=default_params.beta_regime_enabled,
)

process_tick() / _process_tick_unlocked() 签名新增：

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

在 _process_tick_unlocked 的新 K 线更新块中：

if is_new_candle:
    # ... 现有: z4h_history, momentum_filter update ...
    # [新增] Beta 体制跟踪（内部自动去重，仅新 4h 值入队）
    if beta_divergence is not None:
        self._beta_regime.update(key, beta_divergence)

_check_entry() 改动（在 z4h 过滤之后、方向判断之前）：

# ── 现有步骤 3: z4h 绝对值过滤 ──
if abs(z4h) < params.min_zscore_abs:
    ...
    return None

# ── [新增] 步骤 4: Beta 体制检查 ──
if params.beta_regime_enabled:
    beta_state = self._beta_regime.check(
        key,
        divergence_threshold=params.beta_regime_divergence_threshold,
        hard_block_threshold=params.beta_regime_hard_block_threshold,
        scale_max=params.beta_regime_scale_max,
    )
    if beta_state.hard_block:
        logger.info(
            f"🛡️ Beta体制硬拦截 | {pair_label} | {beta_state.reason} | "
            f"az={adaptive_z:+.4f} z4h={z4h:+.4f}"
        )
        return None
    threshold_scale = beta_state.threshold_scale
else:
    beta_state = None
    threshold_scale = 1.0

# ── 步骤 5: 方向判断（应用 Beta 缩放）──
threshold = params.adaptive_threshold * threshold_scale
if adaptive_z < -threshold:
    direction = 'long'
elif adaptive_z > threshold:
    direction = 'short'
else:
    if threshold_scale > 1.01:
        logger.info(
            f"🛡️ Beta体制缩放拦截 | {pair_label} | "
            f"az={adaptive_z:+.4f} 有效阈值={threshold:.2f} "
            f"(原始={params.adaptive_threshold} ×{threshold_scale:.2f}) | "
            f"{beta_state.reason if beta_state else ''}"
        )
    return None

cleanup_pair()：

self._beta_regime.cleanup_pair(key)

4. 参数调优指南

4.1 默认参数

4.2 参数语义示例

beta_regime_divergence_threshold = 0.8：

• β_long=0.5, β_short=0.9 → divergence=(0.9-0.5)/0.5=0.8 → 开始缩放
• β_long=1.0, β_short=1.8 → divergence=(1.8-1.0)/1.0=0.8 → 开始缩放
• 含义：短期 β 比长期 β 上升 80% 时开始介入

beta_regime_hard_block_threshold = 2.0：

• β_long=0.5, β_short=1.5 → divergence=(1.5-0.5)/0.5=2.0 → 硬拦截
• 含义：短期 β 是长期 β 的 3 倍时完全禁止入场

beta_regime_scale_max = 2.0：

• 阈值从 3.0 缩放到 6.0
• adaptive_z 需要 ±6.0 才能入场（极端行情下仍有可能，保留入场窗口）
• 对比 scale_max=3.0：阈值到 9.0 几乎不可能入场，缩放区间与硬拦截无实质差异

β 下降方向的处理：

• β_long=0.5, β_short=0.3 → divergence=max(0, 0.3-0.5)/0.5=0.0 → 不拦截
• 含义：β 下降意味着 Alt 回落，协整关系可能在恢复，不应阻止入场

4.3 场景模拟

场景 A：正常市场（β 稳定）

β_long=0.5, β_short=0.55
β_divergence=max(0, 0.05)/0.5=0.10
mean_divergence=0.08
→ regime=STABLE, scale=1.0
→ threshold=3.0（无变化）
→ 正常入场

场景 B：β 温和上升

β_long=0.5, β_short=0.85
β_divergence=max(0, 0.35)/0.5=0.70
mean_divergence=0.50
→ regime=STABLE, scale=1.0
→ 偏离度尚未超过 0.8 阈值
→ 正常入场，但需关注

场景 C：β 开始飙升（早期检测）

β_long=0.5, β_short=1.2
β_divergence=max(0, 0.7)/0.5=1.40
mean_divergence=0.90（近期均值=1.20）
effective_div=1.20（取较大值）
→ regime=EXPANDING, scale=1.0+(1.20-0.8)/(2.0-0.8)×1.0=1.33
→ threshold=3.0×1.33=4.0
→ 入场门槛提高，z4h<4.0 被拦截

场景 D：β 全面飙升（硬拦截）

β_long=0.5, β_short=3.0
β_divergence=max(0, 2.5)/0.5=5.0
mean_divergence=3.5（近期均值=4.2）
effective_div=4.2
→ regime=EXPANDING, hard_block=True
→ 完全禁止入场
→ 日志: "🛡️ Beta体制硬拦截 | PURR|HYPE | Beta硬拦截: mean_div=3.500 ..."

场景 E：β 飙升后企稳

β_long=2.0（已跟上），β_short=2.2
β_divergence=max(0, 0.2)/2.0=0.10
mean_divergence=0.30（历史缓冲中仍有较高值，但在衰减）
→ regime=STABLE, scale=1.0
→ threshold=3.0
→ β_long 已适应新体制，恢复正常交易

场景 F：β 下降（Alt 回落）

β_long=0.5, β_short=0.25（Alt 跑输 BTC）
β_divergence=max(0, -0.25)/0.5=0.0（方向性保护）
→ regime=STABLE, scale=1.0
→ 不拦截。β 下降 = 协整关系可能在恢复，允许入场

场景 G：低 β 资产对

β_long=0.05, β_short=0.10
β_divergence=max(0, 0.05)/max(0.05, 0.1)=0.05/0.1=0.50
→ regime=STABLE（分母保护避免了 divergence=1.0 的误触发）
→ 正常入场

5. 改动文件清单

不涉及改动的文件：

• momentum_filter.py — 不变
• position_manager.py — 不变（不涉及平仓逻辑，见 §7.1 风险评估）
• executor.py — 不变
• models.py — 不变（PairTradeSignal 暂不加字段）

6. 日志设计

6.1 新增日志

6.2 飞书告警集成

当 Beta 体制从 STABLE → EXPANDING 转换时，可在 orchestrator 层发送飞书告警（可选增强，不在本次范围内）。

7. 风险与边界条件

7.1 已识别风险

7.2 不在本次范围

• 退场逻辑调整（β 飙升时的持仓保护）— 建议作为 P0 后续任务
• 冷却期（EXPANDING → STABLE 的过渡保护）
• β_long 窗口自适应（根据 divergence 动态调短 β_long 窗口）
• 飞书告警集成
• PairTradeSignal 增加 beta_divergence 字段

8. 验证方案

8.1 单元测试

# test_beta_regime.py

def test_stable_regime():
    """β_divergence 稳定低值 → regime=STABLE, scale=1.0"""
    tracker = _BetaRegimeTracker(enabled=True)
    key = ("PURR/USDC:USDC", "HYPE/USDC:USDC")
    # 模拟 10 根 4h K线的低 divergence（每次值不同以通过去重）
    for i in range(10):
        tracker.update(key, 0.10 + i * 0.002)
    state = tracker.check(key, 0.8, 2.0, 2.0)
    assert state.regime == 'stable'
    assert state.threshold_scale == 1.0

def test_expanding_soft():
    """β_divergence 超过 divergence_threshold → 缩放"""
    tracker = _BetaRegimeTracker(enabled=True)
    key = ("PURR/USDC:USDC", "HYPE/USDC:USDC")
    for i in range(10):
        tracker.update(key, 1.20 + i * 0.002)
    state = tracker.check(key, 0.8, 2.0, 2.0)
    assert state.regime == 'expanding'
    assert 1.0 < state.threshold_scale <= 2.0
    assert not state.hard_block

def test_expanding_hard():
    """β_divergence 超过 hard_block_threshold → 硬拦截"""
    tracker = _BetaRegimeTracker(enabled=True)
    key = ("PURR/USDC:USDC", "HYPE/USDC:USDC")
    for i in range(10):
        tracker.update(key, 3.00 + i * 0.002)
    state = tracker.check(key, 0.8, 2.0, 2.0)
    assert state.regime == 'expanding'
    assert state.hard_block

def test_recent_spike():
    """长期低但近期急升 → 近期窗口触发"""
    tracker = _BetaRegimeTracker(enabled=True)
    key = ("PURR/USDC:USDC", "HYPE/USDC:USDC")
    # 30 根 4h K线低值
    for i in range(30):
        tracker.update(key, 0.10 + i * 0.002)
    # 10 根 4h K线急升
    for i in range(10):
        tracker.update(key, 1.50 + i * 0.002)
    state = tracker.check(key, 0.8, 2.0, 2.0)
    assert state.regime == 'expanding'  # 近期均值 ~1.5 > 0.8

def test_dedup():
    """相同 β_divergence 值不重复入队"""
    tracker = _BetaRegimeTracker(enabled=True)
    key = ("PURR/USDC:USDC", "HYPE/USDC:USDC")
    # 模拟同一根 4h K线的 48 个 5m tick（值相同）
    for _ in range(48):
        tracker.update(key, 1.5)
    # 应该只入队 1 次，不满足 MIN_SAMPLES=3
    state = tracker.check(key, 0.8, 2.0, 2.0)
    assert state.regime == 'stable'  # 数据不足
    assert state.reason == "数据不足"

def test_directional():
    """β 下降时 divergence=0，不触发拦截"""
    # β_divergence 由分析层计算，tracker 只接收值
    # 此测试验证分析层传入 0.0 时 tracker 行为正确
    tracker = _BetaRegimeTracker(enabled=True)
    key = ("PURR/USDC:USDC", "HYPE/USDC:USDC")
    for i in range(10):
        tracker.update(key, 0.0 + i * 0.0001)  # 微小变化通过去重
    state = tracker.check(key, 0.8, 2.0, 2.0)
    assert state.regime == 'stable'
    assert state.threshold_scale == 1.0

def test_disabled():
    """禁用时始终返回 stable"""
    tracker = _BetaRegimeTracker(enabled=False)
    key = ("PURR/USDC:USDC", "HYPE/USDC:USDC")
    for i in range(10):
        tracker.update(key, 5.0 + i * 0.002)
    state = tracker.check(key, 0.8, 2.0, 2.0)
    assert state.regime == 'stable'
    assert state.threshold_scale == 1.0

8.2 集成验证

1. 回测：在 β 飙升的历史区间（如 2024 年 BTC 减半后的 alt season）上回测：

   • 对比有无 Beta 体制过滤的入场次数和胜率
   • 预期：过滤后入场次数减少 50%+，但胜率提升

2. 实盘观察：

   • 监控 beta_divergence 日志，验证数值范围
   • 观察 🛡️ Beta体制硬拦截 和 🛡️ Beta体制缩放拦截 日志频率
   • 确认 β 下降场景（divergence=0.0）不产生拦截日志
   • 根据实际情况调整 divergence_threshold 和 hard_block_threshold