关键位监控

This commit is contained in:
dekun
2026-05-21 06:26:07 +08:00
parent ee38eb1c45
commit 2cddce92a0
12 changed files with 784 additions and 110 deletions
+139
View File
@@ -0,0 +1,139 @@
"""关键位箱体/收敛:止盈止损方案(Binance / Gate / OKX 共用)。"""
KEY_SL_TP_MODES = frozenset({"standard", "box_1p5", "trend_manual"})
KEY_SL_TP_MODE_LABELS = {
"standard": "标准突破",
"box_1p5": "箱体1R·止盈1.5H",
"trend_manual": "趋势单·自填止盈",
}
KEY_MONITOR_AUTO_TYPES_FOR_FORM = frozenset({"箱体突破", "收敛突破"})
def normalize_sl_tp_mode(raw):
m = (raw or "standard").strip().lower()
if m in ("box_1p5", "box15", "box-1.5", "box_1.5"):
return "box_1p5"
if m in ("trend_manual", "trend", "manual"):
return "trend_manual"
if m in KEY_SL_TP_MODES:
return m
return "standard"
def sl_tp_mode_label(mode):
return KEY_SL_TP_MODE_LABELS.get(normalize_sl_tp_mode(mode), normalize_sl_tp_mode(mode))
def sl_tp_mode_from_row(row, default="standard"):
try:
if hasattr(row, "keys") and "sl_tp_mode" in row.keys():
raw = row["sl_tp_mode"]
else:
raw = row.get("sl_tp_mode") if isinstance(row, dict) else None
except Exception:
raw = None
return normalize_sl_tp_mode(raw if raw not in (None, "") else default)
def breakeven_enabled_from_row(row, default=0):
try:
if hasattr(row, "keys") and "breakeven_enabled" in row.keys():
v = row["breakeven_enabled"]
else:
v = row.get("breakeven_enabled") if isinstance(row, dict) else None
except Exception:
v = None
if v is None:
return int(default) != 0
return int(v) != 0
def parse_breakeven_enabled_form(form_value):
return 1 if (form_value or "").strip().lower() in ("1", "true", "on", "yes") else 0
def plan_key_sl_tp(
mode,
direction,
upper,
lower,
checks,
*,
outside_pct,
trend_outside_pct,
manual_take_profit=None,
):
"""
以确认 K 收盘 E 为「当前价」计算计划 SL/TP。
返回 (E, sl_raw, tp_raw, box_h) 或 None(几何无效 / 模式3缺止盈)。
"""
try:
E = float(checks["confirm_close"])
H = abs(float(upper) - float(lower))
except (TypeError, ValueError, KeyError):
return None
if H <= 0:
return None
direction = (direction or "long").strip().lower()
mode = normalize_sl_tp_mode(mode)
if mode == "box_1p5":
if direction == "long":
sl_raw = E - H
tp_raw = E + 1.5 * H
else:
sl_raw = E + H
tp_raw = E - 1.5 * H
return E, sl_raw, tp_raw, H
if mode == "trend_manual":
try:
br_hi = float(checks["breakout_high"])
br_lo = float(checks["breakout_low"])
tp_raw = float(manual_take_profit)
except (TypeError, ValueError, KeyError):
return None
m = float(trend_outside_pct) / 100.0
if direction == "long":
sl_raw = br_lo * (1.0 - m) if br_lo > 0 else 0.0
if tp_raw <= E or sl_raw <= 0:
return None
else:
sl_raw = br_hi * (1.0 + m) if br_hi > 0 else 0.0
if tp_raw >= E or sl_raw <= 0:
return None
return E, sl_raw, tp_raw, H
# standard:突破 K 极值外侧 + 止盈 E±1×H
try:
br_hi = float(checks["breakout_high"])
br_lo = float(checks["breakout_low"])
except (TypeError, ValueError, KeyError):
return None
om = float(outside_pct) / 100.0
if direction == "long":
sl_raw = br_lo * (1.0 - om) if br_lo > 0 else 0.0
tp_raw = E + H
else:
sl_raw = br_hi * (1.0 + om) if br_hi > 0 else 0.0
tp_raw = E - H
return E, sl_raw, tp_raw, H
def sl_tp_plan_summary_text(mode, direction, E, sl_raw, tp_raw, box_h, *, outside_pct, trend_outside_pct):
"""微信/页面用一行计划 SL/TP 说明。"""
mode = normalize_sl_tp_mode(mode)
direction = (direction or "long").strip().lower()
if mode == "box_1p5":
return (
f"方案:{sl_tp_mode_label(mode)}E={E}SL=E∓1×H({box_h})TP=E∓1.5×H"
)
if mode == "trend_manual":
return (
f"方案:{sl_tp_mode_label(mode)}E={E}SL=突破K极值外{trend_outside_pct}%TP={tp_raw}(录入)"
)
return (
f"方案:{sl_tp_mode_label(mode)}E={E}SL=突破K外{outside_pct}%TP=E±1×H({box_h})"
)