开仓动量过滤器设计方案

版本：v1.0
日期：2026-03-06
适用系统：Hyperliquid 量化交易系统（Adaptive Bollinger Z-Score 配对策略）

目录

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 时平仓
• 使用 1-5 分钟 K 线，最大持仓时间控制在 30 分钟以内

1.2 待解决的问题

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

1.3 四项约束目标

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

2. 算法选择与理由

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

候选算法对比

选择：N 根 K 线净位移（Net Displacement）

核心公式：

N = max_hold_minutes / candle_period_minutes
  = 30 / 1  = 30 根（1分钟K线）
  = 30 / 5  = 6  根（5分钟K线）

net_return = (close[-1] - close[-N]) / close[-N]

约束1 触发：direction == 'short' AND net_return <= -sustained_threshold
约束2 触发：direction == 'long'  AND net_return >= +sustained_threshold

选择原因：

1. 对 1-5min 噪音免疫：1 分钟 K 线经常出现 +1/-1/+1/-1 交替，纯计数无法准确判断趋势。净位移只看"起点到终点"，中间震荡不影响结果。

2. 时间窗口与持仓时间对齐：N = max_hold_minutes / candle_period，过滤的时间范围恰好等于最大持仓周期，语义一致：「过去 30 分钟内」单边幅度是否过大。

3. 无需预热：不依赖历史统计分布，只需 N 根 K 线历史即可工作，系统启动后很快就能生效。

4. 参数直接可解释：sustained_threshold = 0.8% 意味着「过去 30 分钟内净涨/跌幅超过 0.8% 则判定为持续单边」，可以直接对照行情校准。

2.2 约束 3&4：急涨急跌过滤

候选算法对比

选择：短期 ATR 倍数（ATR Multiple）

核心公式：

ATR(period=10) = 近 10 根 K 线的平均真实波幅
True Range(i) = max(high-low, |high-close_prev|, |low-close_prev|)

candle_return = (close[-1] - close[-2]) / close[-2]
spike_threshold_pct = spike_k * ATR(10) / close[-1]

约束3 触发：direction == 'short' AND 近 spike_lookback 根K线内存在 candle_return >= +spike_threshold_pct
约束4 触发：direction == 'long'  AND 近 spike_lookback 根K线内存在 candle_return <= -spike_threshold_pct

选择原因：

1. 自适应波动率：加密货币在不同时段（亚盘/美盘/重大事件）波动率差异 3-5 倍。固定百分比会在低波动期过于宽松、高波动期过于严格。ATR(10) 在 1-5 分钟周期约为 10-50 分钟的滚动波幅，响应足够灵敏。

2. 预热时间短：只需 10 根 K 线即可计算（vs Z-score 需要 50+），系统启动后 10-50 分钟内生效。

3. spike_lookback 设计：检查最近 1-2 根 K 线，确保捕捉"刚刚发生"的急动，而非几分钟前的历史。

4. 有 high/low 时更精确：当系统提供完整 OHLCV 时自动使用 True Range；仅有 close 时降级为 close-to-close 差，保证向下兼容。

2.3 算法组合逻辑

两类过滤器相互独立、串联运行：

净位移过滤（约束1&2）→ ATR急动过滤（约束3&4）→ 均通过则允许入场

两个过滤器覆盖不同时间尺度的风险：

• 净位移：过去 30 分钟的"趋势是否已过度延伸"（中时间尺度）
• ATR急动：最近 1-2 根 K 线是否有"暴力冲击"（极短时间尺度）

3. 算法具体实现

3.1 模块位置

src/trading/
  momentum_filter.py    ← 新建（独立模块，不耦合策略逻辑）
  strategy.py           ← 修改 _check_entry()，注入并调用 filter
  config.py             ← 修改 StrategyParams，新增过滤器参数

3.2 MomentumFilter 完整代码

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

实现四项开单约束：
  约束1：连续下跌不追跌（不做空）  → 净位移 < -threshold
  约束2：连续上涨不追涨（不做多）  → 净位移 > +threshold
  约束3：迅速暴涨不做空            → 单根K线涨幅 > spike_k * ATR
  约束4：迅速暴跌不做多            → 单根K线跌幅 > spike_k * ATR

设计原则：
  - 按 symbol（单腿）维度过滤，与配对逻辑解耦
  - 无外部依赖，仅需 close/high/low 序列
  - 数据不足时默认放行（安全降级）
"""

from collections import deque


class MomentumFilter:

    def __init__(
        self,
        candle_period_minutes: int = 1,
        max_hold_minutes: int = 30,
        sustained_threshold: float = 0.008,
        spike_atr_multiple: float = 2.5,
        spike_atr_period: int = 10,
        spike_lookback: int = 2,
        enabled: bool = True,
    ):
        """
        Args:
            candle_period_minutes:  K线周期（分钟），用于计算净位移回望根数
            max_hold_minutes:       最大持仓时间（分钟），决定净位移窗口大小
            sustained_threshold:    净位移阈值（小数，0.008=0.8%）
            spike_atr_multiple:     急动检测：单根涨幅 > k * ATR(period)
            spike_atr_period:       ATR计算周期（根数）
            spike_lookback:         检查最近几根K线内是否有急动（1-3）
            enabled:                总开关，False时所有check直接放行
        """
        self._enabled = enabled
        self._sustained_n = max(3, max_hold_minutes // max(1, candle_period_minutes))
        self._sustained_thresh = sustained_threshold
        self._spike_k = spike_atr_multiple
        self._atr_period = spike_atr_period
        self._spike_lookback = spike_lookback

        buf_size = self._sustained_n + spike_atr_period + spike_lookback + 5
        # symbol → deque of (close, high, low)
        self._buffers: dict[str, deque] = {}
        self._buf_size = buf_size

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

    def update(self, symbol: str, close: float, high: float = 0.0, low: float = 0.0):
        """
        每根新K线收盘时调用。
        high/low 可选；有则 ATR 更精确，无则降级为 close-to-close。
        """
        if symbol not in self._buffers:
            self._buffers[symbol] = deque(maxlen=self._buf_size)
        self._buffers[symbol].append((close, high, low))

    def check(self, symbol: str, direction: str) -> tuple[bool, str]:
        """
        检查 direction 方向的入场是否被动量过滤器阻止。

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

        Returns:
            (allowed: bool, reason: str)
            allowed=False 表示该方向被阻止，reason 为拒绝原因
        """
        if not self._enabled:
            return True, ""

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

        data = list(buf)
        closes = [d[0] for d in data]
        highs  = [d[1] for d in data]
        lows   = [d[2] for d in data]

        # 约束 1&2：净位移过滤
        ok, reason = self._check_sustained(closes, direction)
        if not ok:
            return False, reason

        # 约束 3&4：急动过滤
        ok, reason = self._check_spike(closes, highs, lows, direction)
        if not ok:
            return False, reason

        return True, ""

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

    # ------------------------------------------------------------------
    # 内部实现
    # ------------------------------------------------------------------

    def _check_sustained(self, closes: list, direction: str) -> tuple[bool, str]:
        """
        约束1&2：过去 sustained_n 根K线的净位移过滤。

        逻辑：
          净位移 <= -threshold → 市场已持续下跌 → 拒绝 short（不追跌）
          净位移 >= +threshold → 市场已持续上涨 → 拒绝 long（不追涨）
        """
        n = self._sustained_n
        if len(closes) < n + 1:
            return True, ""

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

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

        if direction == 'short' and net_return <= -self._sustained_thresh:
            return False, (
                f"约束1-不追跌: 过去{n}根净跌幅={net_return:.2%}"
                f" <= -{self._sustained_thresh:.2%}"
            )
        if direction == 'long' and net_return >= self._sustained_thresh:
            return False, (
                f"约束2-不追涨: 过去{n}根净涨幅={net_return:.2%}"
                f" >= +{self._sustained_thresh:.2%}"
            )
        return True, ""

    def _check_spike(
        self, closes: list, highs: list, lows: list, direction: str
    ) -> tuple[bool, str]:
        """
        约束3&4：最近 spike_lookback 根K线内是否发生急涨/急跌。

        逻辑：
          任意一根 candle_return >= +spike_k*ATR → 急涨 → 拒绝 short
          任意一根 candle_return <= -spike_k*ATR → 急跌 → 拒绝 long
        """
        atr = self._calc_atr(closes, highs, lows, self._atr_period)
        if atr <= 0 or closes[-1] <= 0:
            return True, ""

        spike_pct = self._spike_k * atr / closes[-1]

        for i in range(-1, -1 - self._spike_lookback, -1):
            idx = len(closes) + i
            if idx < 1:
                break
            prev = closes[idx - 1]
            if prev <= 0:
                continue
            candle_ret = (closes[idx] - prev) / prev

            if direction == 'short' and candle_ret >= spike_pct:
                return False, (
                    f"约束3-暴涨不做空: 第{abs(i)}根涨幅={candle_ret:.2%}"
                    f" >= {self._spike_k}×ATR={spike_pct:.2%}"
                )
            if direction == 'long' and candle_ret <= -spike_pct:
                return False, (
                    f"约束4-暴跌不做多: 第{abs(i)}根跌幅={candle_ret:.2%}"
                    f" <= -{self._spike_k}×ATR={spike_pct:.2%}"
                )

        return True, ""

    @staticmethod
    def _calc_atr(closes: list, highs: list, lows: list, period: int) -> float:
        """
        计算 ATR(period)。
        有 high/low 时使用 True Range，否则降级为 close-to-close 差。
        """
        trs = []
        start = max(1, len(closes) - period)
        for i in range(start, len(closes)):
            prev_c = closes[i - 1]
            h = highs[i]
            l = lows[i]
            if h > 0 and l > 0 and prev_c > 0:
                tr = max(h - l, abs(h - prev_c), abs(l - prev_c))
            elif prev_c > 0:
                tr = abs(closes[i] - prev_c)
            else:
                continue
            trs.append(tr)

        return sum(trs) / len(trs) if trs else 0.0

3.3 StrategyParams 新增字段

在 src/trading/config.py 的 StrategyParams 中新增：

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

    # 动量过滤器参数
    momentum_filter_enabled: bool = True
    momentum_candle_period_minutes: int = 1       # K线周期（分钟）
    momentum_max_hold_minutes: int = 30            # 最大持仓时间（分钟），决定净位移窗口
    momentum_sustained_threshold: float = 0.008   # 净位移阈值（0.8%）
    momentum_spike_atr_multiple: float = 2.5       # 急动检测：N倍ATR
    momentum_spike_atr_period: int = 10            # ATR计算周期
    momentum_spike_lookback: int = 2               # 检查最近N根K线有无急动

3.4 strategy.py 集成改动

在 AdaptiveBollingerStrategy.__init__() 中初始化：

from src.trading.momentum_filter import MomentumFilter

class AdaptiveBollingerStrategy:
    def __init__(self, config: TradingConfig):
        # ... 现有初始化 ...

        # 动量过滤器（symbol 维度，非配对维度）
        default_p = self._params_for("")
        self._momentum_filter = MomentumFilter(
            candle_period_minutes=default_p.momentum_candle_period_minutes,
            max_hold_minutes=default_p.momentum_max_hold_minutes,
            sustained_threshold=default_p.momentum_sustained_threshold,
            spike_atr_multiple=default_p.momentum_spike_atr_multiple,
            spike_atr_period=default_p.momentum_spike_atr_period,
            spike_lookback=default_p.momentum_spike_lookback,
            enabled=default_p.momentum_filter_enabled,
        ) if default_p.momentum_filter_enabled else None

在 process_tick() 中新增 open_ / high / low 参数，并调用 update()：

def process_tick(
    self,
    symbol: str,
    base_symbol: str,
    z4h: float,
    timestamp: datetime,
    kline_time: datetime | None = None,
    latest_price: float | None = None,
    kline_open: float | None = None,    # 新增
    kline_high: float | None = None,    # 新增
    kline_low: float | None = None,     # 新增
    kline_close: float | None = None,   # 新增（等同 latest_price，但语义更明确）
) -> tuple[EntrySignal | None, ExitSignal | None]:
    # 新K线时更新动量过滤器
    if is_new_candle and self._momentum_filter and kline_close:
        close = kline_close or latest_price or 0.0
        self._momentum_filter.update(
            symbol=symbol,
            close=close,
            high=kline_high or 0.0,
            low=kline_low or 0.0,
        )

在 _check_entry() 步骤 4（z4h 过滤）之后、步骤 5（方向判断）之前插入：

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

    # 步骤 4.5：动量过滤（四项开单约束）
    if self._momentum_filter and self._momentum_filter.ready(key[0]):
        # 提前确定方向（用于过滤器检查）
        threshold = params.adaptive_threshold
        if adaptive_z < -threshold:
            proposed_direction = 'long'
        elif adaptive_z > threshold:
            proposed_direction = 'short'
        else:
            return None  # 不在阈值范围，不产生信号

        allowed, reason = self._momentum_filter.check(key[0], proposed_direction)
        if not allowed:
            logger.info(f"🚫 动量过滤 | {pair_label} | {reason}")
            return None

    # 步骤 5：方向判断（原有逻辑）
    # ...

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

4.1 数据流全景

WebSocket K线推送
      |
      v
realtime_kline_service（1min/5min 原始 OHLCV）
      |
      v  process_tick(z4h, close, high, low, open)
      |
AdaptiveBollingerStrategy
      |
      +-- [新K线] MomentumFilter.update(symbol, close, high, low)
      |
      +-- _check_exit()  → ExitSignal
      |
      +-- _check_entry()
              |
              步骤1: 冷却期
              步骤2: 突破检测（adaptive_z 首次穿越阈值）
              步骤3: 持仓检查
              步骤4: z4h 绝对值过滤
         [新增]
              步骤4.5: MomentumFilter.check(symbol, direction)
                        ├── 约束1&2: 净位移过滤
                        └── 约束3&4: ATR急动过滤
              步骤5: 产生 EntrySignal
      |
      v
TradingOrchestrator.on_entry_signal()
      |
      v
RiskManager.pre_trade_check()（现有9项风控）
      |
      v
HyperliquidExecutor.open_position()

4.2 改动范围最小化原则

不需要改动：

• orchestrator.py（无感知）
• risk_manager.py（过滤在上游完成）
• executor.py（无感知）
• models.py（无感知）
• 数据库 schema（无需持久化过滤器状态）

4.3 配置层集成

在 StrategyParams 增加带默认值的字段，支持全局默认 + 按 symbol 覆盖 + 按 pair 覆盖，与现有参数覆盖体系完全一致。

环境变量示例：

# 全局动量过滤器配置
TRADING_MOMENTUM_FILTER_ENABLED=true
TRADING_MOMENTUM_CANDLE_PERIOD_MINUTES=1
TRADING_MOMENTUM_MAX_HOLD_MINUTES=30
TRADING_MOMENTUM_SUSTAINED_THRESHOLD=0.008
TRADING_MOMENTUM_SPIKE_ATR_MULTIPLE=2.5
TRADING_MOMENTUM_SPIKE_ATR_PERIOD=10
TRADING_MOMENTUM_SPIKE_LOOKBACK=2

# 针对特定高波动 symbol 放宽阈值（示例）
TRADING_STRATEGY_OVERRIDE_SYMBOLS=BTC,SOL
TRADING_STRATEGY_BTC_MOMENTUM_SUSTAINED_THRESHOLD=0.012
TRADING_STRATEGY_SOL_MOMENTUM_SPIKE_ATR_MULTIPLE=3.0

4.4 线程安全

MomentumFilter 不持有锁，由 AdaptiveBollingerStrategy 的 _lock（已存在）统一保护。update() 和 check() 均在 _lock 的持有范围内调用，无竞态风险。

4.5 日志集成

拒绝入场时使用 INFO 级别（与现有入场过滤日志一致）：

🚫 动量过滤 | BTC|ETH | 约束1-不追跌: 过去30根净跌幅=-1.23% <= -0.80%
🚫 动量过滤 | SOL|BTC | 约束3-暴涨不做空: 第1根涨幅=1.85% >= 2.5×ATR=1.20%

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

5.1 交易风格特征

5.2 契合度分析

高度契合之处：

1. 时间窗口对齐：净位移窗口 N = max_hold_minutes / candle_period，与策略的最大持仓时间语义一致——过滤的是「持仓期间内」可能影响均值回归的单边趋势。

2. 信号层级匹配：过滤器在 _check_entry() 内执行，与现有的 z4h 过滤、冷却期等并列，均属于「信号层」过滤，而非「风控层」拒绝。逻辑层次清晰。

3. 均值回归哲学一致：均值回归策略本身就假设「偏离终将回归」。动量过滤器在此基础上增加一个判断：「确实是偏离，但偏离是否仍在单边趋势的惯性中？」——只有趋势动量已衰减，均值回归才更可靠。

4. 独立模块不影响出场：过滤器只影响入场判断，对 _check_exit() 和止损逻辑完全不干预，已有持仓的管理不受影响。

5. symbol 维度 vs 配对维度：过滤器在 symbol 单腿维度工作，与策略引擎的 PairKey 维度正交，不干扰多配对并发运行。

潜在张力（需注意）：

• 均值回归策略逻辑上倾向于在"趋势最强的时候入场"（z4h 偏离最大），而动量过滤器会在此时压制部分信号。这是有意为之的权衡：宁可少开几单，不要开在动量最猛的那一刻。

5.3 综合评分

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

6.1 逐项验证

约束 1：连续下跌不追跌

触发条件：net_return(过去N根) <= -sustained_threshold
被阻止：direction == 'short'
原因：市场已持续下跌，跌幅充分释放，继续做空空间有限且反弹风险高

示例（1min K线，N=30，threshold=0.8%）：

• 过去 30 分钟从 100 跌到 99（净跌 1%）→ 拒绝做空 ✓
• 过去 30 分钟震荡，净位移 -0.3%（低于阈值）→ 允许做空 ✓

约束 2：连续上涨不追涨

触发条件：net_return(过去N根) >= +sustained_threshold
被阻止：direction == 'long'
原因：市场已持续上涨，涨幅过度延伸，继续做多风险极高

示例（1min K线，N=30，threshold=0.8%）：

• 过去 30 分钟从 100 涨到 101.2（净涨 1.2%）→ 拒绝做多 ✓
• 过去 30 分钟震荡，净位移 +0.5% → 允许做多 ✓

约束 3：迅速暴涨不做空

触发条件：任意近spike_lookback根K线的 candle_return >= spike_k * ATR(10)
被阻止：direction == 'short'
原因：刚刚发生的急涨说明多方动量极强，做空极易被轧空

示例（spike_k=2.5，ATR≈0.5%，spike_threshold≈1.25%）：

• 最近一根K线涨了 1.8%（> 2.5 × 0.5%）→ 拒绝做空 ✓
• 最近一根K线涨了 0.6%（< 2.5 × 0.5%）→ 允许做空 ✓

约束 4：迅速暴跌不做多

触发条件：任意近spike_lookback根K线的 candle_return <= -(spike_k * ATR(10))
被阻止：direction == 'long'
原因：刚刚发生的急跌说明空方动量极强，做多极易被继续砸穿

示例（spike_k=2.5，ATR≈0.5%，spike_threshold≈1.25%）：

• 最近一根K线跌了 2.1%（> 2.5 × 0.5%）→ 拒绝做多 ✓
• 最近一根K线跌了 0.4% → 允许做多 ✓

6.2 约束覆盖矩阵

四项约束完全覆盖，无遗漏，无逻辑矛盾。

7. 参数配置指南

7.1 基础参数推荐值

7.2 参数调优思路

sustained_threshold 调优：

• 过低（< 0.5%）：过于敏感，过多有效信号被过滤
• 过高（> 2%）：过于宽松，无法拦截真正的单边行情
• 建议：用历史 30min 滚动净位移的 60th-70th 百分位作为初始值

spike_atr_multiple 调优：

• 过低（< 2.0）：高波动期频繁误杀
• 过高（> 4.0）：只拦截极端行情，日常急动漏过
• 建议：从 2.5 开始，观察被拦截的信号与实际行情对应关系

回测验证建议：

对比两组回测结果：
  A 组：无动量过滤（当前）
  B 组：开启动量过滤（本方案）

对比指标：
  - Sharpe Ratio（风险调整收益）
  - 最大连续亏损次数
  - 平均单笔盈亏比（W/L Ratio）
  - 信号数量减少比例（过滤率）

8. 风险与局限性

8.1 已知局限

8.2 不适合的场景

• 极低流动性资产：ATR 计算可能因价格跳空而失真，需配合黑名单
• HIP-3 资产：若 K 线数据稀疏，buffer 积累时间更长，预热期更长

8.3 后续迭代方向

1. spread 净位移：除单腿价格外，同时检查 z4h spread 的净位移方向
2. 自适应阈值：sustained_threshold = k * rolling_daily_vol，随波动率自动调整
3. 分时段参数：亚盘 / 欧盘 / 美盘使用不同阈值（美盘波动显著更大）
4. 预填充历史：启动时从 DB 的 klines 表回填近 N 根 K 线，消除预热等待

文档结束