dekun
126 lines
4.0 KiB
Python
126 lines
4.0 KiB
Python
"""
|
||
关键位监控：阻力/支撑双向提醒与箱体/收敛自动门控的共享逻辑。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from typing import Any, Optional
|
||
|
||
KEY_MONITOR_AUTO_TYPES = frozenset({"箱体突破", "收敛突破"})
|
||
KEY_MONITOR_RS_TYPES = frozenset({"关键阻力位", "关键支撑位"})
|
||
KEY_MONITOR_ALERT_ONLY_TYPES = KEY_MONITOR_RS_TYPES
|
||
KEY_DIRECTION_WATCH = "watch"
|
||
|
||
|
||
def calc_breakout_breach_pct(direction: str, close: float, upper: float, lower: float) -> float:
|
||
"""突破 K 收盘相对关键位的越过幅度（%）。未越过对应边界时返回 0。"""
|
||
direction = (direction or "long").strip().lower()
|
||
c = float(close)
|
||
if direction == "long":
|
||
u = float(upper)
|
||
if u <= 0 or c <= u:
|
||
return 0.0
|
||
return (c - u) / u * 100.0
|
||
lo = float(lower)
|
||
if lo <= 0 or c >= lo:
|
||
return 0.0
|
||
return (lo - c) / lo * 100.0
|
||
|
||
|
||
def auto_amp_ok(
|
||
direction: str,
|
||
close_b: float,
|
||
upper: float,
|
||
lower: float,
|
||
min_pct: float,
|
||
) -> tuple[bool, float]:
|
||
breach = calc_breakout_breach_pct(direction, close_b, upper, lower)
|
||
return breach > float(min_pct), breach
|
||
|
||
|
||
def auto_confirm_ok(direction: str, cfm_close: float, upper: float, lower: float) -> bool:
|
||
"""确认 K 收盘须在箱体外（不得回到 [lower, upper] 内）。"""
|
||
direction = (direction or "long").strip().lower()
|
||
c = float(cfm_close)
|
||
if direction == "long":
|
||
return c > float(upper)
|
||
return c < float(lower)
|
||
|
||
|
||
def detect_rs_box_break(close: float, upper: float, lower: float) -> Optional[dict[str, Any]]:
|
||
"""
|
||
阻力/支撑人工盯盘：最近 5m 收盘突破上沿或下沿（严格 > / <）。
|
||
上沿优先：同一根 K 不可能同时满足两者。
|
||
"""
|
||
u, lo, c = float(upper), float(lower), float(close)
|
||
if c > u:
|
||
return {
|
||
"break_side": "upper",
|
||
"direction": "long",
|
||
"edge_price": u,
|
||
"key_price": u,
|
||
"break_label": "向上突破上沿",
|
||
}
|
||
if c < lo:
|
||
return {
|
||
"break_side": "lower",
|
||
"direction": "short",
|
||
"edge_price": lo,
|
||
"key_price": lo,
|
||
"break_label": "向下突破下沿",
|
||
}
|
||
return None
|
||
|
||
|
||
def rs_break_from_direction(direction: str, upper: float, lower: float) -> Optional[dict[str, Any]]:
|
||
"""已触发后根据入库方向还原突破边（long=上沿，short=下沿）。"""
|
||
d = (direction or "").strip().lower()
|
||
if d == "long":
|
||
return {
|
||
"break_side": "upper",
|
||
"direction": "long",
|
||
"edge_price": float(upper),
|
||
"key_price": float(upper),
|
||
"break_label": "向上突破上沿",
|
||
}
|
||
if d == "short":
|
||
return {
|
||
"break_side": "lower",
|
||
"direction": "short",
|
||
"edge_price": float(lower),
|
||
"key_price": float(lower),
|
||
"break_label": "向下突破下沿",
|
||
}
|
||
return None
|
||
|
||
|
||
def notify_interval_elapsed(
|
||
last_notified_at: Optional[str],
|
||
interval_min: int,
|
||
now_dt: datetime,
|
||
) -> bool:
|
||
if not last_notified_at:
|
||
return True
|
||
try:
|
||
last_dt = datetime.fromisoformat(str(last_notified_at).replace("Z", "+00:00"))
|
||
if last_dt.tzinfo is not None:
|
||
last_dt = last_dt.replace(tzinfo=None)
|
||
except Exception:
|
||
return True
|
||
return (now_dt - last_dt).total_seconds() >= max(1, int(interval_min)) * 60
|
||
|
||
|
||
def format_auto_amp_line(amp_ok: bool, amp_pct: float, min_pct: float) -> str:
|
||
return (
|
||
f"突破越过幅度：{'通过' if amp_ok else '不通过'}"
|
||
f"（{round(float(amp_pct), 4)}%，要求 > {min_pct}%）"
|
||
)
|
||
|
||
|
||
def format_auto_confirm_line(confirm_ok: bool, cfm_close, edge_price, direction: str) -> str:
|
||
side = "箱外上方" if (direction or "").lower() == "long" else "箱外下方"
|
||
return (
|
||
f"第二根确认：{'通过' if confirm_ok else '不通过'}"
|
||
f"（确认收盘 {cfm_close}，须收于{side}，关键位 {edge_price}）"
|
||
)
|