Add amplitude filter as first open gate for 永期 hedge.

Reuse OO amplitude settings (default off); Plan shows index/HL when filtering.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/strategy/amplitude_gate.py

@@ -0,0 +1,69 @@
+"""振幅过滤门：回看窗内 range% 须 ≤ 上限（可关）。永期开仓第一关 / 期期共用。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ..exchange.candles import AmplitudeHL
+
+
+def evaluate_amplitude_gate(
+    *,
+    filter_enabled: bool,
+    amp: AmplitudeHL | None,
+    max_pct: float,
+    hours: float,
+) -> dict[str, Any]:
+    """
+    返回:
+      blocked: 过滤开启且未过关（无K线或超限）
+      ok: 展示用是否过关（过滤关视为 True）
+      reason: 拒开文案（未拒则为 None）
+      snapshot: 写入 session._oo_amp 的字典（amp 为 None 时仅含元数据）
+    """
+    max_p = float(max_pct)
+    hrs = float(hours)
+    filt = bool(filter_enabled)
+    if amp is None:
+        snap = {
+            "high": None,
+            "low": None,
+            "mid": None,
+            "range_pct": None,
+            "hours": hrs,
+            "max_pct": max_p,
+            "filter_enabled": filt,
+            "ok": False if filt else True,
+        }
+        if filt:
+            return {
+                "blocked": True,
+                "ok": False,
+                "reason": f"振幅未过关：无法获取近 {hrs:g}h K 线高低",
+                "snapshot": snap,
+            }
+        return {"blocked": False, "ok": True, "reason": None, "snapshot": snap}
+
+    range_pct = float(amp.range_pct)
+    over = range_pct > max_p + 1e-12
+    ok = (not filt) or (not over)
+    snap = {
+        "high": float(amp.high),
+        "low": float(amp.low),
+        "mid": float(amp.mid),
+        "range_pct": range_pct,
+        "hours": hrs,
+        "max_pct": max_p,
+        "filter_enabled": filt,
+        "ok": ok,
+    }
+    if filt and over:
+        return {
+            "blocked": True,
+            "ok": False,
+            "reason": (
+                f"振幅未过关：{hrs:g}h 内 {range_pct:.2f}% > {max_p:g}%"
+            ),
+            "snapshot": snap,
+        }
+    return {"blocked": False, "ok": ok, "reason": None, "snapshot": snap}

backend/app/strategy/engine.py

@@ -1047,6 +1047,14 @@ class StrategyEngine:
                 )
                 return
         if pick is None:
+            amp_fail = None
+            try:
+                amp_fail = get_session().amplitude_gate_fail_reason()
+            except Exception:
+                amp_fail = None
+            if amp_fail:
+                self._set_state(last_error=amp_fail)
+                return
             try:
                 from .auto_usdc import preview_capacity_for_convert
                 from .open_capacity import funds_gate_blocks

backend/app/strategy/session.py

@@ -352,8 +352,7 @@ class StrategySession:
 
     def align_to_held_position(self) -> OptionPair | None:
         """有活跃仓时：监控对锁定为持仓合约的到期/行权价。"""
-        if _hedge_mode() == "option_option":
-            self.refresh_oo_amplitude()
+        self.refresh_oo_amplitude()
         call_id, put_id = _held_option_legs()
         held = call_id or _held_option_inst_id()
         if not held:
@@ -421,6 +420,8 @@ class StrategySession:
             return self.align_to_held_position()
         if _hedge_mode() == "option_option":
             return self.align_oo_instruments()
+        # 永期：刷新振幅供 Plan（过滤关也展示；开仓门禁在 pick）
+        self.refresh_oo_amplitude()
         s = self.settings
         idx = self.ex.fetch_index(s.index_inst_id)
         mark = self.ex.fetch_mark(s.perp_inst_id) or idx
@@ -464,35 +465,67 @@ class StrategySession:
         return self._apply_pair(pair, mark=float(mark), idx=idx)
 
     def refresh_oo_amplitude(self) -> dict[str, Any] | None:
-        """刷新振幅高低（有仓/无仓都要，否则持仓后 UI 指数/振幅会空）。"""
-        if _hedge_mode() != "option_option":
-            return None
+        """刷新振幅高低（永期/期期；有仓/无仓都要，否则 UI 指数/振幅会空）。"""
         try:
             from ..exchange.candles import fetch_amplitude_hl_for_runtime
+            from .amplitude_gate import evaluate_amplitude_gate
 
             amp_pct, amp_hours, _, _, _, amp_filt = _oo_settings()
             amp = fetch_amplitude_hl_for_runtime(amp_hours)
-            if amp is None:
-                return self._oo_amp
-            self._oo_amp = {
-                "high": float(amp.high),
-                "low": float(amp.low),
-                "mid": float(amp.mid),
-                "range_pct": float(amp.range_pct),
-                "hours": float(amp_hours),
-                "max_pct": float(amp_pct),
-                "filter_enabled": bool(amp_filt),
-                "ok": (
-                    True
-                    if not amp_filt
-                    else float(amp.range_pct) <= float(amp_pct) + 1e-12
-                ),
-            }
+            gate = evaluate_amplitude_gate(
+                filter_enabled=bool(amp_filt),
+                amp=amp,
+                max_pct=float(amp_pct),
+                hours=float(amp_hours),
+            )
+            self._oo_amp = dict(gate["snapshot"])
             return self._oo_amp
         except Exception:
             logger.exception("refresh_oo_amplitude failed")
             return self._oo_amp
 
+    def _apply_amplitude_first_gate(self) -> bool:
+        """
+        振幅过滤为开仓第一关。返回 True=可继续选约；False=拒开。
+        无论是否开启过滤都写入 _oo_amp 供 Plan 展示。
+        """
+        from ..exchange.candles import fetch_amplitude_hl_for_runtime
+        from .amplitude_gate import evaluate_amplitude_gate
+
+        amp_pct, amp_hours, _, _, _, amp_filt = _oo_settings()
+        amp = fetch_amplitude_hl_for_runtime(amp_hours)
+        gate = evaluate_amplitude_gate(
+            filter_enabled=bool(amp_filt),
+            amp=amp,
+            max_pct=float(amp_pct),
+            hours=float(amp_hours),
+        )
+        self._oo_amp = dict(gate["snapshot"])
+        if gate["blocked"]:
+            logger.info("amplitude gate blocked: %s", gate.get("reason"))
+            return False
+        return True
+
+    def amplitude_gate_fail_reason(self) -> str | None:
+        """若最近一次振幅快照显示过滤开启且未过关，返回文案。"""
+        amp = self._oo_amp
+        if not amp or not amp.get("filter_enabled"):
+            return None
+        if amp.get("ok") is True:
+            return None
+        hours = amp.get("hours")
+        range_pct = amp.get("range_pct")
+        max_pct = amp.get("max_pct")
+        if range_pct is None:
+            return f"振幅未过关：无法获取近 {hours:g}h K 线高低" if hours is not None else "振幅未过关"
+        try:
+            return (
+                f"振幅未过关：{float(hours):g}h 内 "
+                f"{float(range_pct):.2f}% > {float(max_pct):g}%"
+            )
+        except (TypeError, ValueError):
+            return "振幅未过关"
+
     def align_oo_instruments(self) -> OptionPair | None:
         """期期监控：按振幅高低点选虚值 Call/Put（展示用；振幅超限仍对齐候选）。"""
         from .oo_selection import select_oo_pair
@@ -693,6 +726,9 @@ class StrategySession:
         min_hours, min_lev, max_atm_off, atm_off_on = _strategy_floats()
         fixed_on, fixed_perp = _fixed_direction()
         opt_side_hint = _option_side_for_perp(fixed_perp) if fixed_on else None
+        # 第一关：振幅过滤（默认关；开启则回看窗振幅须 ≤ 最大%）
+        if not self._apply_amplitude_first_gate():
+            return None
         idx = self.ex.fetch_index(s.index_inst_id)
         mark = self.ex.fetch_mark(s.perp_inst_id) or idx
         if mark is None or mark <= 0:
@@ -932,9 +968,8 @@ class StrategySession:
 
     async def ensure_atm_async(self, *, force: bool = False) -> OptionPair | None:
         if _has_open_position():
-            # 持仓期间：钉住持仓行权价（禁止漂到新 ATM/虚值）；期期仍刷新振幅供 UI
-            if _hedge_mode() == "option_option":
-                await asyncio.to_thread(self.refresh_oo_amplitude)
+            # 持仓期间：钉住持仓行权价（禁止漂到新 ATM/虚值）；仍刷新振幅供 UI
+            await asyncio.to_thread(self.refresh_oo_amplitude)
             call_id, put_id = _held_option_legs()
             held = call_id or _held_option_inst_id()
             if held and (

backend/tests/test_amplitude_gate.py

@@ -0,0 +1,60 @@
+"""振幅过滤门：永期开仓第一关 / 与期期共用口径。"""
+
+from __future__ import annotations
+
+from app.exchange.candles import AmplitudeHL
+from app.strategy.amplitude_gate import evaluate_amplitude_gate
+
+
+def test_amplitude_gate_off_always_pass() -> None:
+    amp = AmplitudeHL(high=2060, low=1940, mid=2000, hours=12, bar_count=12)
+    # 6% 超常见上限，但过滤关 → 不拦
+    g = evaluate_amplitude_gate(
+        filter_enabled=False, amp=amp, max_pct=2.0, hours=12
+    )
+    assert g["blocked"] is False
+    assert g["ok"] is True
+    assert g["reason"] is None
+    assert g["snapshot"]["filter_enabled"] is False
+    assert g["snapshot"]["range_pct"] == amp.range_pct
+
+
+def test_amplitude_gate_on_blocks_over_max() -> None:
+    amp = AmplitudeHL(high=2060, low=1940, mid=2000, hours=12, bar_count=12)
+    assert amp.range_pct == 6.0
+    g = evaluate_amplitude_gate(
+        filter_enabled=True, amp=amp, max_pct=2.0, hours=12
+    )
+    assert g["blocked"] is True
+    assert g["ok"] is False
+    assert "振幅未过关" in (g["reason"] or "")
+    assert "6.00%" in (g["reason"] or "")
+    assert "2%" in (g["reason"] or "")
+
+
+def test_amplitude_gate_on_pass_within_max() -> None:
+    amp = AmplitudeHL(high=2010, low=1990, mid=2000, hours=12, bar_count=12)
+    assert amp.range_pct == 1.0
+    g = evaluate_amplitude_gate(
+        filter_enabled=True, amp=amp, max_pct=2.0, hours=12
+    )
+    assert g["blocked"] is False
+    assert g["ok"] is True
+    assert g["reason"] is None
+
+
+def test_amplitude_gate_on_no_candles_fail_closed() -> None:
+    g = evaluate_amplitude_gate(
+        filter_enabled=True, amp=None, max_pct=2.0, hours=12
+    )
+    assert g["blocked"] is True
+    assert g["ok"] is False
+    assert "无法获取" in (g["reason"] or "")
+
+
+def test_amplitude_gate_off_no_candles_still_ok() -> None:
+    g = evaluate_amplitude_gate(
+        filter_enabled=False, amp=None, max_pct=2.0, hours=12
+    )
+    assert g["blocked"] is False
+    assert g["ok"] is True