feat: show force-close badge and countdown when FORCE_CLOSE is enabled

Add shared lib/UI for midnight force-close indicator on instance and hub pages, and document Gate intraday 0-point exit alignment.

Co-authored-by: Cursor <cursoragent@cursor.com>

lib/trade/force_close_lib.py

@@ -0,0 +1,179 @@
+"""整点强制清仓（FORCE_CLOSE_*）：UI 标识与持仓倒计时。"""
+from __future__ import annotations
+
+import os
+import time
+from datetime import datetime, timedelta
+from typing import Any, Optional
+from zoneinfo import ZoneInfo
+
+FORCE_CLOSE_RESULT = "强制清仓"
+
+
+def app_timezone_name() -> str:
+    return (os.getenv("APP_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai").strip() or "Asia/Shanghai"
+
+
+def normalize_force_close_bj_hour(value: Any) -> int:
+    try:
+        h = int(value)
+    except (TypeError, ValueError):
+        return 0
+    return max(0, min(23, h))
+
+
+def _now_dt(*, now_ms: Optional[int] = None, tz_name: Optional[str] = None) -> datetime:
+    tz = ZoneInfo(tz_name or app_timezone_name())
+    if now_ms is None:
+        return datetime.now(tz)
+    return datetime.fromtimestamp(int(now_ms) / 1000, tz=tz)
+
+
+def force_close_hour_label(bj_hour: Any) -> str:
+    return f"{normalize_force_close_bj_hour(bj_hour):02d}:00"
+
+
+def force_close_label(bj_hour: Any) -> str:
+    return f"强制清仓 {force_close_hour_label(bj_hour)}"
+
+
+def is_force_close_active_hour(
+    bj_hour: Any,
+    *,
+    now_ms: Optional[int] = None,
+    tz_name: Optional[str] = None,
+) -> bool:
+    """当前是否处于整点强制清仓执行窗口（该北京时间整点小时内）。"""
+    hour = normalize_force_close_bj_hour(bj_hour)
+    return _now_dt(now_ms=now_ms, tz_name=tz_name).hour == hour
+
+
+def compute_next_force_close_at_ms(
+    *,
+    bj_hour: Any,
+    now_ms: Optional[int] = None,
+    tz_name: Optional[str] = None,
+) -> Optional[int]:
+    """下一次强制清仓时刻（北京时间整点）的 epoch 毫秒。"""
+    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)
+    return int(target.timestamp() * 1000)
+
+
+def force_close_remaining_seconds(
+    close_at_ms: Any,
+    *,
+    now_ms: Optional[int] = None,
+) -> Optional[int]:
+    try:
+        close_at = int(close_at_ms)
+    except (TypeError, ValueError):
+        return None
+    now = int(now_ms if now_ms is not None else time.time() * 1000)
+    return max(0, int((close_at - now) / 1000))
+
+
+def format_force_close_countdown(seconds: Any, *, active: bool = False) -> str:
+    if active:
+        return "执行中"
+    try:
+        sec = max(0, int(seconds))
+    except (TypeError, ValueError):
+        return "--:--:--"
+    h = sec // 3600
+    m = (sec % 3600) // 60
+    s = sec % 60
+    return f"{h:02d}:{m:02d}:{s:02d}"
+
+
+def build_force_close_state(
+    enabled: bool,
+    bj_hour: Any,
+    *,
+    now_ms: Optional[int] = None,
+    tz_name: Optional[str] = None,
+) -> dict[str, Any]:
+    """实例级强制清仓状态（模板 / API 共用）。"""
+    if not enabled:
+        return {
+            "enabled": False,
+            "bj_hour": normalize_force_close_bj_hour(bj_hour),
+            "hour_label": force_close_hour_label(bj_hour),
+            "label": force_close_label(bj_hour),
+            "next_at_ms": None,
+            "remaining_sec": None,
+            "countdown": "",
+            "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)
+    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 {
+        "enabled": True,
+        "bj_hour": hour,
+        "hour_label": force_close_hour_label(hour),
+        "label": force_close_label(hour),
+        "next_at_ms": next_at_ms,
+        "remaining_sec": rem,
+        "countdown": format_force_close_countdown(rem, active=active),
+        "active": active,
+    }
+
+
+def force_close_template_context(
+    enabled: bool,
+    bj_hour: Any,
+    *,
+    now_ms: Optional[int] = None,
+    tz_name: Optional[str] = None,
+) -> dict[str, dict[str, Any]]:
+    return {
+        "force_close": build_force_close_state(
+            enabled, bj_hour, now_ms=now_ms, tz_name=tz_name
+        )
+    }
+
+
+def apply_force_close_to_payload(
+    payload: dict[str, Any],
+    *,
+    enabled: bool,
+    bj_hour: Any,
+    now_ms: Optional[int] = None,
+    tz_name: Optional[str] = None,
+) -> None:
+    """为 active 持仓 JSON 附加整点强制清仓倒计时。"""
+    state = build_force_close_state(enabled, bj_hour, now_ms=now_ms, tz_name=tz_name)
+    payload["force_close_enabled"] = bool(state["enabled"])
+    payload["force_close_bj_hour"] = state["bj_hour"]
+    payload["force_close_at_ms"] = state["next_at_ms"]
+    payload["force_close_label"] = state["label"] if state["enabled"] else ""
+    payload["force_close_remaining_sec"] = state["remaining_sec"]
+    payload["force_close_countdown"] = state["countdown"]
+    payload["force_close_active"] = bool(state["active"])
+
+
+def enrich_orders_force_close(
+    orders: list[dict[str, Any]],
+    enabled: bool,
+    bj_hour: Any,
+    *,
+    now_ms: Optional[int] = None,
+    tz_name: Optional[str] = None,
+) -> None:
+    if not enabled or not orders:
+        return
+    for item in orders:
+        if isinstance(item, dict):
+            apply_force_close_to_payload(
+                item,
+                enabled=enabled,
+                bj_hour=bj_hour,
+                now_ms=now_ms,
+                tz_name=tz_name,
+            )