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

fix: Gate force-close badge and classify midnight closes correctly

Browse Source 
Narrow executing window to 15 minutes after the hour, show countdown when flat, run force_close before reconcile, and map 00:00 synced closes to 强制清仓 instead of 手动平仓.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-07 00:23:40 +08:00
parent d520ba6904
commit 99eb16dbd8
5 changed files with 250 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/trade/force_close_lib.py
+149 -8
View File
@@ -8,6 +8,7 @@ from typing import Any, Optional
from zoneinfo import ZoneInfo
FORCE_CLOSE_RESULT = "强制清仓"
FORCE_CLOSE_GRACE_MINUTES = 15
def app_timezone_name() -> str:
@@ -42,10 +43,138 @@ def is_force_close_active_hour(
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
) -> bool:
"""当前是否处于整点强制清仓执行窗口（整点起 grace 分钟内）。"""
return is_force_close_executing(
bj_hour,
now_ms=now_ms,
tz_name=tz_name,
grace_minutes=grace_minutes,
)
def is_force_close_executing(
bj_hour: Any,
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
) -> bool:
"""当前是否处于整点强制清仓执行窗口（该北京时间整点小时内）。"""
hour = normalize_force_close_bj_hour(bj_hour)
return _now_dt(now_ms=now_ms, tz_name=tz_name).hour == hour
now = _now_dt(now_ms=now_ms, tz_name=tz_name)
target = now.replace(hour=hour, minute=0, second=0, microsecond=0)
if now < target:
return False
end = target + timedelta(minutes=max(1, int(grace_minutes)))
return now < end
def parse_closed_at_dt(
closed_at: Any,
*,
tz_name: Optional[str] = None,
) -> Optional[datetime]:
if closed_at is None:
return None
text = str(closed_at).strip()
if not text:
return None
tz = ZoneInfo(tz_name or app_timezone_name())
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M"):
try:
return datetime.strptime(text, fmt).replace(tzinfo=tz)
except ValueError:
continue
return None
def is_close_at_force_close_window(
closed_at: Any,
bj_hour: Any,
*,
grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
tz_name: Optional[str] = None,
) -> bool:
"""平仓时刻是否落在北京时间整点强制清仓窗口内。"""
dt = parse_closed_at_dt(closed_at, tz_name=tz_name)
if dt is None:
return False
hour = normalize_force_close_bj_hour(bj_hour)
if dt.hour != hour:
return False
return dt.minute < max(1, int(grace_minutes))
def infer_force_close_result(
closed_at: Any,
*,
enabled: bool,
bj_hour: Any,
grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
tz_name: Optional[str] = None,
) -> Optional[str]:
if not enabled:
return None
if is_close_at_force_close_window(
closed_at, bj_hour, grace_minutes=grace_minutes, tz_name=tz_name
):
return FORCE_CLOSE_RESULT
return None
def coerce_force_close_result(
result: Optional[str],
closed_at: Any,
*,
enabled: bool,
bj_hour: Any,
miss_reason: Optional[str] = None,
grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
tz_name: Optional[str] = None,
) -> tuple[str, str]:
"""同步平仓归类：整点窗口内优先记为强制清仓。"""
res = (result or "").strip()
note = (miss_reason or "").strip()
if res == FORCE_CLOSE_RESULT:
return res, note
fc = infer_force_close_result(
closed_at,
enabled=enabled,
bj_hour=bj_hour,
grace_minutes=grace_minutes,
tz_name=tz_name,
)
if not fc:
return res, note
if not note:
note = f"北京时间 {force_close_hour_label(bj_hour)} 整点风控清仓"
return fc, note
def apply_force_close_display_result(
result: Optional[str],
closed_at: Any,
*,
enabled: bool,
bj_hour: Any,
grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
tz_name: Optional[str] = None,
) -> str:
"""展示层：外部平仓/手动平仓若落在整点窗口，显示为强制清仓。"""
res = (result or "").strip()
if res == FORCE_CLOSE_RESULT:
return res
fc = infer_force_close_result(
closed_at,
enabled=enabled,
bj_hour=bj_hour,
grace_minutes=grace_minutes,
tz_name=tz_name,
)
if fc and (res in ("", "外部平仓", "手动平仓") or res.startswith("外部平仓")):
return fc
return res
def compute_next_force_close_at_ms(
@@ -58,9 +187,8 @@ def compute_next_force_close_at_ms(
hour = normalize_force_close_bj_hour(bj_hour)
now = _now_dt(now_ms=now_ms, tz_name=tz_name)
target = now.replace(hour=hour, minute=0, second=0, microsecond=0)
if now.hour > hour or now.hour == hour:
if now.hour > hour:
target += timedelta(days=1)
if now >= target:
target += timedelta(days=1)
return int(target.timestamp() * 1000)
@@ -96,6 +224,7 @@ def build_force_close_state(
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
has_active_positions: Optional[bool] = None,
) -> dict[str, Any]:
"""实例级强制清仓状态（模板 / API 共用）。"""
if not enabled:
@@ -110,7 +239,8 @@ def build_force_close_state(
"active": False,
}
hour = normalize_force_close_bj_hour(bj_hour)
active = is_force_close_active_hour(hour, now_ms=now_ms, tz_name=tz_name)
executing = is_force_close_executing(hour, now_ms=now_ms, tz_name=tz_name)
active = executing and (has_active_positions is not False)
next_at_ms = compute_next_force_close_at_ms(bj_hour=hour, now_ms=now_ms, tz_name=tz_name)
rem = force_close_remaining_seconds(next_at_ms, now_ms=now_ms) if next_at_ms else None
return {
@@ -131,10 +261,15 @@ def force_close_template_context(
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
has_active_positions: Optional[bool] = None,
) -> dict[str, dict[str, Any]]:
return {
"force_close": build_force_close_state(
enabled, bj_hour, now_ms=now_ms, tz_name=tz_name
enabled,
bj_hour,
now_ms=now_ms,
tz_name=tz_name,
has_active_positions=has_active_positions,
)
}
@@ -148,7 +283,13 @@ def apply_force_close_to_payload(
tz_name: Optional[str] = None,
) -> None:
"""为 active 持仓 JSON 附加整点强制清仓倒计时。"""
state = build_force_close_state(enabled, bj_hour, now_ms=now_ms, tz_name=tz_name)
state = build_force_close_state(
enabled,
bj_hour,
now_ms=now_ms,
tz_name=tz_name,
has_active_positions=True,
)
payload["force_close_enabled"] = bool(state["enabled"])
payload["force_close_bj_hour"] = state["bj_hour"]
payload["force_close_at_ms"] = state["next_at_ms"]