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 = "强制清仓"
FORCE_CLOSE_GRACE_MINUTES = 15


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,
    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)
    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(
    *,
    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 >= target:
        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,
    has_active_positions: Optional[bool] = 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)
    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 {
        "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,
    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,
            has_active_positions=has_active_positions,
        )
    }


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,
        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"]
    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,
            )