dekun/gate_scout_order
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复趋势突破

Browse Source
This commit is contained in:
dekun
2026-05-23 17:25:54 +08:00
parent ffba2e60e6
commit 24bfb07be1
8 changed files with 83 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
onchain_scout_gate/app/config.py
+1
View File
@@ -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
onchain_scout_gate/app/key_gate.py
+6 -2
View File
@@ -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}（空→高点+外扩%｜多→低点−外扩%）",
]
onchain_scout_gate/app/key_monitor_service.py
+15 -3
View File
@@ -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")
onchain_scout_gate/app/key_sl_tp.py
+17 -1
View File
@@ -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}｜"
onchain_scout_gate/app/web.py
+4 -3
View File
@@ -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 '不转发执行器'}"
)
onchain_scout_gate/config.example.yaml
+1
View File
@@ -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
onchain_scout_gate/tests/test_key_gate.py
+28
View File
@@ -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
onchain_scout_gate/tests/test_key_sl_tp.py
+11 -1
View File
@@ -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)