crypto_monitor/lib/trade/force_close_lib.py

"""整点强制清仓（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,
            )