From 24bfb07be1460a5a8f6df8846b203f9763671ba3 Mon Sep 17 00:00:00 2001 From: dekun Date: Sat, 23 May 2026 17:25:54 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=B6=8B=E5=8A=BF=E7=AA=81?= =?UTF-8?q?=E7=A0=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- onchain_scout_gate/app/config.py | 1 + onchain_scout_gate/app/key_gate.py | 8 ++++-- onchain_scout_gate/app/key_monitor_service.py | 18 ++++++++++-- onchain_scout_gate/app/key_sl_tp.py | 18 +++++++++++- onchain_scout_gate/app/web.py | 7 +++-- onchain_scout_gate/config.example.yaml | 1 + onchain_scout_gate/tests/test_key_gate.py | 28 +++++++++++++++++++ onchain_scout_gate/tests/test_key_sl_tp.py | 12 +++++++- 8 files changed, 83 insertions(+), 10 deletions(-) diff --git a/onchain_scout_gate/app/config.py b/onchain_scout_gate/app/config.py index 36171a9..91d8a5e 100644 --- a/onchain_scout_gate/app/config.py +++ b/onchain_scout_gate/app/config.py @@ -86,6 +86,7 @@ class KeyMonitorConfig(BaseModel): volume_ratio_min: float = Field(1.3, gt=0.0) breakout_amp_min_pct: float = Field(0.03, ge=0.0) breakout_amp_max_pct: float = Field(0.5, gt=0.0) + trend_breakout_amp_max_pct: float = Field(1.5, gt=0.0) daily_volume_rank_max: int = Field(30, ge=1) # 全市场 5m TRIGGER 是否仍转发执行器（默认关；关键位走 forward_executor） auto_scan_forward_executor: bool = False diff --git a/onchain_scout_gate/app/key_gate.py b/onchain_scout_gate/app/key_gate.py index afefbe0..f683c1b 100644 --- a/onchain_scout_gate/app/key_gate.py +++ b/onchain_scout_gate/app/key_gate.py @@ -114,6 +114,8 @@ def key_hard_checks_from_rows( "breakout_open": open_b, "direction": direction, "swing4h_pct": swing4h_pct, + "amp_min_pct": breakout_amp_min_pct, + "amp_max_pct": breakout_amp_max_pct, } ) return out @@ -122,6 +124,8 @@ def key_hard_checks_from_rows( def key_hard_lines_from_checks(checks: dict, *, volume_ratio_min: float) -> list[str]: breach = float(checks.get("breach_pct") if checks.get("breach_pct") is not None else checks.get("amp_pct") or 0) body = float(checks.get("body_pct") or 0) + amp_min = float(checks.get("amp_min_pct") or 0.03) + amp_max = float(checks.get("amp_max_pct") or 0.5) br_hi = checks.get("breakout_high") br_lo = checks.get("breakout_low") return [ @@ -129,9 +133,9 @@ def key_hard_lines_from_checks(checks: dict, *, volume_ratio_min: float) -> list f"突破价位：{'通过' if checks.get('breakout_ok') else '不通过'}（突破K收盘 {round(float(checks.get('breakout_close') or 0), 8)}，关键位 {checks.get('edge_price')}）", ( f"突破越过关键位：{'通过' if checks.get('amp_ok') else '不通过'}" - f"（越过 {round(breach, 4)}%，K线实体 {round(body, 4)}%，要求越过 0.03%~0.5%）" + f"（越过 {round(breach, 4)}%，K线实体 {round(body, 4)}%，要求越过 {amp_min:g}%~{amp_max:g}%）" ), f"第二根确认：{'通过' if checks.get('confirm_ok') else '不通过'}（确认收盘 {checks.get('confirm_close')}，关键位 {checks.get('edge_price')}）", f"日成交量排名：{'通过' if checks.get('rank_ok') else '不通过'}（{checks.get('rank')}/{checks.get('rank_total')}，要求前30）", - f"突破K极值：高 {br_hi}｜低 {br_lo}（止损据此 ± 外扩%）", + f"突破K极值：高 {br_hi}｜低 {br_lo}（空→高点+外扩%｜多→低点−外扩%）", ] diff --git a/onchain_scout_gate/app/key_monitor_service.py b/onchain_scout_gate/app/key_monitor_service.py index cff68a8..5d28b91 100644 --- a/onchain_scout_gate/app/key_monitor_service.py +++ b/onchain_scout_gate/app/key_monitor_service.py @@ -10,6 +10,7 @@ from .config import Settings from .gate import GateClient from .key_gate import key_hard_checks_from_rows, key_hard_lines_from_checks from .key_sl_tp import ( + breakout_amp_max_for_mode, calc_planned_rr, normalize_sl_tp_mode, plan_key_sl_tp, @@ -54,6 +55,12 @@ class KeyMonitorService: sorted_insts = sorted(vol_map.keys(), key=lambda x: float(vol_map.get(x, 0.0)), reverse=True) rank_map = {i: idx + 1 for idx, i in enumerate(sorted_insts)} rank = rank_map.get(inst) + sl_tp_mode = normalize_sl_tp_mode(row.get("sl_tp_mode")) + amp_max = breakout_amp_max_for_mode( + sl_tp_mode, + standard_max_pct=cfg.breakout_amp_max_pct, + trend_max_pct=cfg.trend_breakout_amp_max_pct, + ) checks = key_hard_checks_from_rows( closed, direction=direction, @@ -62,7 +69,7 @@ class KeyMonitorService: volume_ma_bars=cfg.volume_ma_bars, volume_ratio_min=cfg.volume_ratio_min, breakout_amp_min_pct=cfg.breakout_amp_min_pct, - breakout_amp_max_pct=cfg.breakout_amp_max_pct, + breakout_amp_max_pct=amp_max, volume_rank=rank, volume_rank_total=len(sorted_insts), volume_rank_max=cfg.daily_volume_rank_max, @@ -119,6 +126,12 @@ class KeyMonitorService: closed = raw_rows[:-1] rank = rank_map.get(inst) + sl_tp_mode = normalize_sl_tp_mode(row.get("sl_tp_mode")) + amp_max = breakout_amp_max_for_mode( + sl_tp_mode, + standard_max_pct=cfg.breakout_amp_max_pct, + trend_max_pct=cfg.trend_breakout_amp_max_pct, + ) checks = key_hard_checks_from_rows( closed, direction=direction, @@ -127,7 +140,7 @@ class KeyMonitorService: volume_ma_bars=cfg.volume_ma_bars, volume_ratio_min=cfg.volume_ratio_min, breakout_amp_min_pct=cfg.breakout_amp_min_pct, - breakout_amp_max_pct=cfg.breakout_amp_max_pct, + breakout_amp_max_pct=amp_max, volume_rank=rank, volume_rank_total=rank_total, volume_rank_max=cfg.daily_volume_rank_max, @@ -135,7 +148,6 @@ class KeyMonitorService: if not checks.get("ok"): return - sl_tp_mode = normalize_sl_tp_mode(row.get("sl_tp_mode")) outside = float(row.get("stop_outside_pct") or stop_outside_pct_for_mode(sl_tp_mode)) trend_out = cfg.trend_stop_outside_pct manual_tp = row.get("manual_take_profit") diff --git a/onchain_scout_gate/app/key_sl_tp.py b/onchain_scout_gate/app/key_sl_tp.py index 8a0aade..f10ea47 100644 --- a/onchain_scout_gate/app/key_sl_tp.py +++ b/onchain_scout_gate/app/key_sl_tp.py @@ -36,6 +36,16 @@ def stop_outside_pct_for_mode(sl_tp_mode: str) -> float: return 1.0 if normalize_sl_tp_mode(sl_tp_mode) == "trend_manual" else 0.3 +def breakout_amp_max_for_mode( + sl_tp_mode: str, + *, + standard_max_pct: float, + trend_max_pct: float, +) -> float: + """标准突破越过关键位上限默认 0.5%；趋势突破放宽（默认 1.5%）。""" + return float(trend_max_pct) if normalize_sl_tp_mode(sl_tp_mode) == "trend_manual" else float(standard_max_pct) + + def plan_key_sl_tp( mode: str, direction: str, @@ -127,10 +137,16 @@ def sl_tp_plan_summary_text( trend_outside_pct: float, ) -> str: mode_n = normalize_sl_tp_mode(mode) + dir_n = (direction or "long").strip().lower() if mode_n == "trend_manual": + sl_side = ( + f"空=突破K高点+{trend_outside_pct:g}%" + if dir_n == "short" + else f"多=突破K低点−{trend_outside_pct:g}%" + ) return ( f"方案：{sl_tp_mode_label(mode_n)}｜E={e}｜" - f"SL=突破K极值外{trend_outside_pct:g}%｜TP={tp_raw}（录入）" + f"SL={sl_side}｜TP={tp_raw}（录入）" ) return ( f"方案：{sl_tp_mode_label(mode_n)}｜E={e}｜" diff --git a/onchain_scout_gate/app/web.py b/onchain_scout_gate/app/web.py index a52a2ad..48a0e3c 100644 --- a/onchain_scout_gate/app/web.py +++ b/onchain_scout_gate/app/web.py @@ -530,10 +530,11 @@ def create_app(settings: Settings) -> FastAPI: km = settings.key_monitor return ( f"周期 5m｜突破K/确认K：倒数第2/第1根闭合K｜量能：突破K量 > 前{km.volume_ma_bars}均量×{km.volume_ratio_min}｜" - f"突破越过关键位 {km.breakout_amp_min_pct:g}%~{km.breakout_amp_max_pct:g}%（非K线实体）｜" + f"越过关键位：标准 {km.breakout_amp_min_pct:g}%~{km.breakout_amp_max_pct:g}%｜" + f"趋势 {km.breakout_amp_min_pct:g}%~{km.trend_breakout_amp_max_pct:g}%｜" f"计划RR须 > {km.min_planned_rr:g}｜日成交额排名前{km.daily_volume_rank_max}｜" - f"箱体/收敛方案：标准突破（止损突破K外{km.standard_stop_outside_pct:g}%｜止盈1×H）或 " - f"趋势突破（止损突破K外{km.trend_stop_outside_pct:g}%｜止盈手填）｜" + f"箱体/收敛方案：标准突破（止损突破K低/高外{km.standard_stop_outside_pct:g}%｜止盈1×H）或 " + f"趋势突破（空=突破K高点+{km.trend_stop_outside_pct:g}%｜多=突破K低点−{km.trend_stop_outside_pct:g}%｜止盈手填）｜" f"触发后企微+{'转发执行器' if km.forward_executor else '不转发执行器'}" ) diff --git a/onchain_scout_gate/config.example.yaml b/onchain_scout_gate/config.example.yaml index 8aa04ad..bd46c6f 100644 --- a/onchain_scout_gate/config.example.yaml +++ b/onchain_scout_gate/config.example.yaml @@ -48,6 +48,7 @@ key_monitor: volume_ratio_min: 1.3 breakout_amp_min_pct: 0.03 breakout_amp_max_pct: 0.5 + trend_breakout_amp_max_pct: 1.5 daily_volume_rank_max: 30 # 全市场 5m TRIGGER 是否仍转发执行器（默认 false，仅关键位转发） auto_scan_forward_executor: false diff --git a/onchain_scout_gate/tests/test_key_gate.py b/onchain_scout_gate/tests/test_key_gate.py index 7bf6083..018ffa1 100644 --- a/onchain_scout_gate/tests/test_key_gate.py +++ b/onchain_scout_gate/tests/test_key_gate.py @@ -49,6 +49,34 @@ class TestKeyGate(unittest.TestCase): self.assertTrue(checks["amp_ok"]) self.assertAlmostEqual(checks["breach_pct"], (83.28 - 83.18) / 83.28 * 100, places=4) + def test_trend_breach_wider_max_passes(self): + """趋势突破：越过 1% 在 1.5% 上限内应通过幅度门控。""" + rows = _history(22, vol=40000.0) + # lower 83.28, close 82.45 → 越过约 0.996% + rows.append(_row(1, 83.0, 83.5, 82.3, 82.45, 76791.0)) + rows.append(_row(2, 82.4, 82.5, 82.2, 82.30, 50000.0)) + checks = key_hard_checks_from_rows( + rows, + direction="short", + upper=87.8, + lower=83.28, + breakout_amp_max_pct=1.5, + volume_rank=3, + volume_rank_total=721, + ) + self.assertTrue(checks["amp_ok"]) + self.assertFalse( + key_hard_checks_from_rows( + rows, + direction="short", + upper=87.8, + lower=83.28, + breakout_amp_max_pct=0.5, + volume_rank=3, + volume_rank_total=721, + )["amp_ok"] + ) + def test_trend_short_sl_uses_breakout_high(self): rows = _history(22, vol=40000.0) br_hi = 83.55 diff --git a/onchain_scout_gate/tests/test_key_sl_tp.py b/onchain_scout_gate/tests/test_key_sl_tp.py index f10a503..6f24f6c 100644 --- a/onchain_scout_gate/tests/test_key_sl_tp.py +++ b/onchain_scout_gate/tests/test_key_sl_tp.py @@ -2,7 +2,13 @@ from __future__ import annotations import unittest -from app.key_sl_tp import calc_planned_rr, normalize_sl_tp_mode, plan_key_sl_tp, stop_outside_pct_for_mode +from app.key_sl_tp import ( + breakout_amp_max_for_mode, + calc_planned_rr, + normalize_sl_tp_mode, + plan_key_sl_tp, + stop_outside_pct_for_mode, +) class TestKeySlTp(unittest.TestCase): @@ -62,6 +68,10 @@ class TestKeySlTp(unittest.TestCase): self.assertEqual(stop_outside_pct_for_mode("trend_manual"), 1.0) self.assertEqual(normalize_sl_tp_mode("box_1p5"), "standard") + def test_breakout_amp_max_for_mode(self): + self.assertEqual(breakout_amp_max_for_mode("standard", standard_max_pct=0.5, trend_max_pct=1.5), 0.5) + self.assertEqual(breakout_amp_max_for_mode("trend_manual", standard_max_pct=0.5, trend_max_pct=1.5), 1.5) + def test_rr(self): rr = calc_planned_rr("long", 100.0, 98.0, 104.0) self.assertAlmostEqual(rr, 2.0, places=4)