fix: 箱体/收敛突破标记价反向越界时自动撤销监控

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

crypto_monitor_binance/app.py

@@ -168,6 +168,8 @@ from key_monitor_lib import (
     KEY_MONITOR_RS_TYPES,
     auto_amp_ok,
     auto_confirm_ok,
+    box_breakout_invalidate_by_mark,
+    box_breakout_invalidate_edge_label,
     claim_rs_level_notify,
     detect_rs_box_break,
     format_auto_amp_line,
@@ -6036,6 +6038,21 @@ def check_key_monitors():
         direction = (r["direction"] or "long").lower()
         if direction == KEY_DIRECTION_WATCH:
             continue
+        if typ in KEY_MONITOR_AUTO_TYPES:
+            mark = get_symbol_mark_price(sym)
+            if mark is not None and box_breakout_invalidate_by_mark(direction, mark, up, low):
+                edge = float(low) if direction == "long" else float(up)
+                edge_label = box_breakout_invalidate_edge_label(direction)
+                msg = (
+                    f"# ⚠️ {sym} 关键位监控失效\n"
+                    f"**账户：{_wechat_account_label()}**\n"
+                    f"- 类型：{typ}｜{_wechat_direction_text(direction)}\n"
+                    f"- 标记价 {format_price_for_symbol(sym, mark)} 已突破反向{edge_label} "
+                    f"{format_price_for_symbol(sym, edge)}（设置失效）\n"
+                )
+                send_wechat_msg(msg)
+                _finalize_key_monitor_one_shot(conn, r, msg, "box_opposite_break")
+                continue
         try:
             checks = _key_hard_checks(sym, direction, up, low, typ)
         except Exception:
@@ -7197,6 +7214,7 @@ def api_price_snapshot():
         fib_gate_ok = True
         fb_gate_ok = True
         te_gate_ok = True
+        box_gate_ok = True
         if is_fib:
             direction = (r["direction"] or "long").lower()
             inval = fib_invalidate_by_mark(direction, price, r["upper"], r["lower"])
@@ -7253,16 +7271,22 @@ def api_price_snapshot():
             except Exception:
                 gate_summary = "-"
         elif (r["monitor_type"] or "").strip() in KEY_MONITOR_AUTO_TYPES:
-            try:
-                gate = _key_hard_checks(
-                    r["symbol"],
-                    (r["direction"] or "long").lower(),
-                    r["upper"],
-                    r["lower"],
-                    r["monitor_type"],
-                )
-            except Exception:
-                gate = None
+            direction = (r["direction"] or "long").lower()
+            if box_breakout_invalidate_by_mark(direction, price, r["upper"], r["lower"]):
+                edge_label = box_breakout_invalidate_edge_label(direction)
+                gate_summary = f"反向突破{edge_label}·将撤销"
+                box_gate_ok = False
+            else:
+                try:
+                    gate = _key_hard_checks(
+                        r["symbol"],
+                        direction,
+                        r["upper"],
+                        r["lower"],
+                        r["monitor_type"],
+                    )
+                except Exception:
+                    gate = None
         if gate:
             rank_seg = "ERR" if int(gate.get("rank_total") or 0) <= 0 else f"{gate.get('rank')}/{gate.get('rank_total')}"
             gate_summary = (
@@ -7301,7 +7325,7 @@ def api_price_snapshot():
                 fib_gate_ok if is_fib
                 else fb_gate_ok if is_fb
                 else te_gate_ok if is_te
-                else bool(gate and gate.get("ok"))
+                else box_gate_ok and bool(gate and gate.get("ok"))
             ),
             "gate_metrics": gate_metrics,
         })

crypto_monitor_binance/关键位自动下单说明.md

@@ -112,6 +112,7 @@ Binance / OKX 见各自目录下同名文档；共享逻辑在仓库根目录 `k
 
 | `close_reason` | 含义 |
 |----------------|------|
+| `box_opposite_break` | 标记价先突破反向边界（多：≤下沿；空：≥上沿） |
 | `rr_insufficient` | 门控通过但 RR 不达标或 SL/TP 几何无效 |
 | `exchange_failed` | RR 达标但实盘/交易所等原因未开仓 |
 | `auto_opened` | RR 达标且市价开仓成功 |

crypto_monitor_gate/app.py

@@ -167,6 +167,8 @@ from key_monitor_lib import (
     KEY_MONITOR_RS_TYPES,
     auto_amp_ok,
     auto_confirm_ok,
+    box_breakout_invalidate_by_mark,
+    box_breakout_invalidate_edge_label,
     claim_rs_level_notify,
     detect_rs_box_break,
     format_auto_amp_line,
@@ -5780,6 +5782,21 @@ def check_key_monitors():
         direction = (r["direction"] or "long").lower()
         if direction == KEY_DIRECTION_WATCH:
             continue
+        if typ in KEY_MONITOR_AUTO_TYPES:
+            mark = get_symbol_mark_price(sym)
+            if mark is not None and box_breakout_invalidate_by_mark(direction, mark, up, low):
+                edge = float(low) if direction == "long" else float(up)
+                edge_label = box_breakout_invalidate_edge_label(direction)
+                msg = (
+                    f"# ⚠️ {sym} 关键位监控失效\n"
+                    f"**账户：{_wechat_account_label()}**\n"
+                    f"- 类型：{typ}｜{_wechat_direction_text(direction)}\n"
+                    f"- 标记价 {format_price_for_symbol(sym, mark)} 已突破反向{edge_label} "
+                    f"{format_price_for_symbol(sym, edge)}（设置失效）\n"
+                )
+                send_wechat_msg(msg)
+                _finalize_key_monitor_one_shot(conn, r, msg, "box_opposite_break")
+                continue
         try:
             checks = _key_hard_checks(sym, direction, up, low, typ)
         except Exception:
@@ -7106,6 +7123,7 @@ def api_price_snapshot():
         fib_gate_ok = True
         fb_gate_ok = True
         te_gate_ok = True
+        box_gate_ok = True
         if is_fib:
             direction = (r["direction"] or "long").lower()
             inval = fib_invalidate_by_mark(direction, price, r["upper"], r["lower"])
@@ -7162,16 +7180,22 @@ def api_price_snapshot():
             except Exception:
                 gate_summary = "-"
         elif (r["monitor_type"] or "").strip() in KEY_MONITOR_AUTO_TYPES:
-            try:
-                gate = _key_hard_checks(
-                    r["symbol"],
-                    (r["direction"] or "long").lower(),
-                    r["upper"],
-                    r["lower"],
-                    r["monitor_type"],
-                )
-            except Exception:
-                gate = None
+            direction = (r["direction"] or "long").lower()
+            if box_breakout_invalidate_by_mark(direction, price, r["upper"], r["lower"]):
+                edge_label = box_breakout_invalidate_edge_label(direction)
+                gate_summary = f"反向突破{edge_label}·将撤销"
+                box_gate_ok = False
+            else:
+                try:
+                    gate = _key_hard_checks(
+                        r["symbol"],
+                        direction,
+                        r["upper"],
+                        r["lower"],
+                        r["monitor_type"],
+                    )
+                except Exception:
+                    gate = None
         if gate:
             rank_seg = "ERR" if int(gate.get("rank_total") or 0) <= 0 else f"{gate.get('rank')}/{gate.get('rank_total')}"
             gate_summary = (
@@ -7214,7 +7238,7 @@ def api_price_snapshot():
                 fib_gate_ok if is_fib
                 else fb_gate_ok if is_fb
                 else te_gate_ok if is_te
-                else bool(gate and gate.get("ok"))
+                else box_gate_ok and bool(gate and gate.get("ok"))
             ),
             "gate_metrics": gate_metrics,
         })

crypto_monitor_gate/关键位自动下单说明.md

@@ -112,6 +112,7 @@ Binance / OKX 见各自目录下同名文档；共享逻辑在仓库根目录 `k
 
 | `close_reason` | 含义 |
 |----------------|------|
+| `box_opposite_break` | 标记价先突破反向边界（多：≤下沿；空：≥上沿） |
 | `rr_insufficient` | 门控通过但 RR 不达标或 SL/TP 几何无效 |
 | `exchange_failed` | RR 达标但实盘/交易所等原因未开仓 |
 | `auto_opened` | RR 达标且市价开仓成功 |

crypto_monitor_gate_bot/app.py

@@ -167,6 +167,8 @@ from key_monitor_lib import (
     KEY_MONITOR_RS_TYPES,
     auto_amp_ok,
     auto_confirm_ok,
+    box_breakout_invalidate_by_mark,
+    box_breakout_invalidate_edge_label,
     claim_rs_level_notify,
     detect_rs_box_break,
     format_auto_amp_line,
@@ -5780,6 +5782,21 @@ def check_key_monitors():
         direction = (r["direction"] or "long").lower()
         if direction == KEY_DIRECTION_WATCH:
             continue
+        if typ in KEY_MONITOR_AUTO_TYPES:
+            mark = get_symbol_mark_price(sym)
+            if mark is not None and box_breakout_invalidate_by_mark(direction, mark, up, low):
+                edge = float(low) if direction == "long" else float(up)
+                edge_label = box_breakout_invalidate_edge_label(direction)
+                msg = (
+                    f"# ⚠️ {sym} 关键位监控失效\n"
+                    f"**账户：{_wechat_account_label()}**\n"
+                    f"- 类型：{typ}｜{_wechat_direction_text(direction)}\n"
+                    f"- 标记价 {format_price_for_symbol(sym, mark)} 已突破反向{edge_label} "
+                    f"{format_price_for_symbol(sym, edge)}（设置失效）\n"
+                )
+                send_wechat_msg(msg)
+                _finalize_key_monitor_one_shot(conn, r, msg, "box_opposite_break")
+                continue
         try:
             checks = _key_hard_checks(sym, direction, up, low, typ)
         except Exception:
@@ -7102,6 +7119,7 @@ def api_price_snapshot():
         fib_gate_ok = True
         fb_gate_ok = True
         te_gate_ok = True
+        box_gate_ok = True
         if is_fib:
             direction = (r["direction"] or "long").lower()
             inval = fib_invalidate_by_mark(direction, price, r["upper"], r["lower"])
@@ -7158,16 +7176,22 @@ def api_price_snapshot():
             except Exception:
                 gate_summary = "-"
         elif (r["monitor_type"] or "").strip() in KEY_MONITOR_AUTO_TYPES:
-            try:
-                gate = _key_hard_checks(
-                    r["symbol"],
-                    (r["direction"] or "long").lower(),
-                    r["upper"],
-                    r["lower"],
-                    r["monitor_type"],
-                )
-            except Exception:
-                gate = None
+            direction = (r["direction"] or "long").lower()
+            if box_breakout_invalidate_by_mark(direction, price, r["upper"], r["lower"]):
+                edge_label = box_breakout_invalidate_edge_label(direction)
+                gate_summary = f"反向突破{edge_label}·将撤销"
+                box_gate_ok = False
+            else:
+                try:
+                    gate = _key_hard_checks(
+                        r["symbol"],
+                        direction,
+                        r["upper"],
+                        r["lower"],
+                        r["monitor_type"],
+                    )
+                except Exception:
+                    gate = None
         if gate:
             rank_seg = "ERR" if int(gate.get("rank_total") or 0) <= 0 else f"{gate.get('rank')}/{gate.get('rank_total')}"
             gate_summary = (
@@ -7210,7 +7234,7 @@ def api_price_snapshot():
                 fib_gate_ok if is_fib
                 else fb_gate_ok if is_fb
                 else te_gate_ok if is_te
-                else bool(gate and gate.get("ok"))
+                else box_gate_ok and bool(gate and gate.get("ok"))
             ),
             "gate_metrics": gate_metrics,
         })

crypto_monitor_gate_bot/关键位自动下单说明.md

@@ -110,6 +110,7 @@ Binance / OKX 见各自目录下同名文档；共享逻辑在仓库根目录 `k
 
 | `close_reason` | 含义 |
 |----------------|------|
+| `box_opposite_break` | 标记价先突破反向边界（多：≤下沿；空：≥上沿） |
 | `rr_insufficient` | 门控通过但 RR 不达标或 SL/TP 几何无效 |
 | `exchange_failed` | RR 达标但实盘/交易所等原因未开仓 |
 | `auto_opened` | RR 达标且市价开仓成功 |

crypto_monitor_okx/app.py

@@ -166,6 +166,8 @@ from key_monitor_lib import (
     KEY_MONITOR_RS_TYPES,
     auto_amp_ok,
     auto_confirm_ok,
+    box_breakout_invalidate_by_mark,
+    box_breakout_invalidate_edge_label,
     claim_rs_level_notify,
     detect_rs_box_break,
     format_auto_amp_line,
@@ -5539,6 +5541,21 @@ def check_key_monitors():
         direction = (r["direction"] or "long").lower()
         if direction == KEY_DIRECTION_WATCH:
             continue
+        if typ in KEY_MONITOR_AUTO_TYPES:
+            mark = get_symbol_mark_price(sym)
+            if mark is not None and box_breakout_invalidate_by_mark(direction, mark, up, low):
+                edge = float(low) if direction == "long" else float(up)
+                edge_label = box_breakout_invalidate_edge_label(direction)
+                msg = (
+                    f"# ⚠️ {sym} 关键位监控失效\n"
+                    f"**账户：{_wechat_account_label()}**\n"
+                    f"- 类型：{typ}｜{_wechat_direction_text(direction)}\n"
+                    f"- 标记价 {format_price_for_symbol(sym, mark)} 已突破反向{edge_label} "
+                    f"{format_price_for_symbol(sym, edge)}（设置失效）\n"
+                )
+                send_wechat_msg(msg)
+                _finalize_key_monitor_one_shot(conn, r, msg, "box_opposite_break")
+                continue
         try:
             checks = _key_hard_checks(sym, direction, up, low, typ)
         except Exception:
@@ -6651,6 +6668,7 @@ def api_price_snapshot():
         fib_gate_ok = True
         fb_gate_ok = True
         te_gate_ok = True
+        box_gate_ok = True
         if is_fib:
             direction = (r["direction"] or "long").lower()
             inval = fib_invalidate_by_mark(direction, price, r["upper"], r["lower"])
@@ -6707,16 +6725,22 @@ def api_price_snapshot():
             except Exception:
                 gate_summary = "-"
         elif (r["monitor_type"] or "").strip() in KEY_MONITOR_AUTO_TYPES:
-            try:
-                gate = _key_hard_checks(
-                    r["symbol"],
-                    (r["direction"] or "long").lower(),
-                    r["upper"],
-                    r["lower"],
-                    r["monitor_type"],
-                )
-            except Exception:
-                gate = None
+            direction = (r["direction"] or "long").lower()
+            if box_breakout_invalidate_by_mark(direction, price, r["upper"], r["lower"]):
+                edge_label = box_breakout_invalidate_edge_label(direction)
+                gate_summary = f"反向突破{edge_label}·将撤销"
+                box_gate_ok = False
+            else:
+                try:
+                    gate = _key_hard_checks(
+                        r["symbol"],
+                        direction,
+                        r["upper"],
+                        r["lower"],
+                        r["monitor_type"],
+                    )
+                except Exception:
+                    gate = None
         if gate:
             rank_seg = "ERR" if int(gate.get("rank_total") or 0) <= 0 else f"{gate.get('rank')}/{gate.get('rank_total')}"
             gate_summary = (
@@ -6759,7 +6783,7 @@ def api_price_snapshot():
                 fib_gate_ok if is_fib
                 else fb_gate_ok if is_fb
                 else te_gate_ok if is_te
-                else bool(gate and gate.get("ok"))
+                else box_gate_ok and bool(gate and gate.get("ok"))
             ),
             "gate_metrics": gate_metrics,
         })

crypto_monitor_okx/关键位自动下单说明.md

@@ -112,6 +112,7 @@ Binance / OKX 见各自目录下同名文档；共享逻辑在仓库根目录 `k
 
 | `close_reason` | 含义 |
 |----------------|------|
+| `box_opposite_break` | 标记价先突破反向边界（多：≤下沿；空：≥上沿） |
 | `rr_insufficient` | 门控通过但 RR 不达标或 SL/TP 几何无效 |
 | `exchange_failed` | RR 达标但实盘/交易所等原因未开仓 |
 | `auto_opened` | RR 达标且市价开仓成功 |

key_monitor_lib.py

@@ -66,6 +66,30 @@ def auto_confirm_ok(direction: str, cfm_close: float, upper: float, lower: float
     return c < float(lower)
 
 
+BOX_BREAKOUT_CLOSE_OPPOSITE = "box_opposite_break"
+
+
+def box_breakout_invalidate_by_mark(
+    direction: str, mark_price: float, upper: float, lower: float
+) -> bool:
+    """箱体/收敛：标记价先突破反向边界则失效。多：mark<=L；空：mark>=H。"""
+    try:
+        m = float(mark_price)
+        h = float(upper)
+        lo = float(lower)
+    except (TypeError, ValueError):
+        return False
+    direction = (direction or "long").strip().lower()
+    if direction == "short":
+        return m >= h
+    return m <= lo
+
+
+def box_breakout_invalidate_edge_label(direction: str) -> str:
+    direction = (direction or "long").strip().lower()
+    return "下沿" if direction == "long" else "上沿"
+
+
 def detect_rs_box_break(close: float, upper: float, lower: float) -> Optional[dict[str, Any]]:
     """
     阻力/支撑人工盯盘：最近 5m 收盘突破上沿或下沿（严格 > / <）。

strategy_templates/key_monitor_panel.html

@@ -114,6 +114,7 @@
 {%- elif r == 'auto_opened' -%}自动开仓
 {%- elif r == 'manual' -%}手动删除
 {%- elif r == 'fib_invalidate' -%}斐波失效
+{%- elif r == 'box_opposite_break' -%}反向突破失效
 {%- elif r == 'trigger_tp_invalidate' -%}触价止盈失效
 {%- elif r == 'trigger_sl_invalidate' -%}触价止损失效
 {%- elif r == 'trigger_entry_expired' -%}触价过期

strategy_templates/key_monitor_rule_tips.html

@@ -14,7 +14,7 @@
 <tr>
 <td class="key-rule-type">箱体突破<br><span class="key-rule-sub">收敛突破</span></td>
 <td class="key-rule-cell">方向必选；填 H/L<br>方案：标准 / 1R·1.5H / 趋势<br>可勾移动保本</td>
-<td class="key-rule-cell">{{ r.tf }} 两根闭合 K（{{ r.breakout_bar }}/{{ r.confirm_bar }}）<br>突破 &gt;{{ r.amp_min_pct }}%；确认在箱外<br>量 &gt;前{{ r.vol_ma_bars }}均×{{ r.vol_ratio_min }}<br>成交 Top{{ r.vol_rank_max }}；RR &gt;{{ r.min_rr }}</td>
+<td class="key-rule-cell">{{ r.tf }} 两根闭合 K（{{ r.breakout_bar }}/{{ r.confirm_bar }}）<br>突破 &gt;{{ r.amp_min_pct }}%；确认在箱外<br>量 &gt;前{{ r.vol_ma_bars }}均×{{ r.vol_ratio_min }}<br>成交 Top{{ r.vol_rank_max }}；RR &gt;{{ r.min_rr }}<br>标记价先破反向边界→失效</td>
 <td class="key-rule-cell">标准：SL 极值外{{ r.stop_outside_pct }}%，TP=E±H<br>1R：SL=E∓H，TP=E∓1.5H<br>趋势：SL 极值外{{ r.trend_stop_outside_pct }}%，TP 自填</td>
 <td class="key-rule-cell">门控过→市价开仓→下单监控<br>满仓不可再加</td>
 </tr>

tests/test_key_monitor_box_invalidate.py

@@ -0,0 +1,34 @@
+import unittest
+
+from key_monitor_lib import (
+    BOX_BREAKOUT_CLOSE_OPPOSITE,
+    box_breakout_invalidate_by_mark,
+    box_breakout_invalidate_edge_label,
+)
+
+
+class BoxBreakoutInvalidateTests(unittest.TestCase):
+    def test_short_invalidates_above_upper(self):
+        self.assertTrue(box_breakout_invalidate_by_mark("short", 62.511, 61.746, 60.569))
+
+    def test_short_stays_valid_inside_or_below(self):
+        self.assertFalse(box_breakout_invalidate_by_mark("short", 61.0, 61.746, 60.569))
+        self.assertFalse(box_breakout_invalidate_by_mark("short", 60.0, 61.746, 60.569))
+
+    def test_long_invalidates_below_lower(self):
+        self.assertTrue(box_breakout_invalidate_by_mark("long", 94.0, 100.0, 95.0))
+
+    def test_long_stays_valid_inside_or_above(self):
+        self.assertFalse(box_breakout_invalidate_by_mark("long", 98.0, 100.0, 95.0))
+        self.assertFalse(box_breakout_invalidate_by_mark("long", 101.0, 100.0, 95.0))
+
+    def test_edge_label(self):
+        self.assertEqual(box_breakout_invalidate_edge_label("long"), "下沿")
+        self.assertEqual(box_breakout_invalidate_edge_label("short"), "上沿")
+
+    def test_close_reason_constant(self):
+        self.assertEqual(BOX_BREAKOUT_CLOSE_OPPOSITE, "box_opposite_break")
+
+
+if __name__ == "__main__":
+    unittest.main()