crypto_monitor/time_close_lib.py

"""持仓时间平仓：开仓后按 1h/2h/4h 定时市价平仓。"""
from __future__ import annotations

import time
from typing import Any, Optional

ALLOWED_TIME_CLOSE_HOURS = (1, 2, 4)
TIME_CLOSE_RESULT = "时间平仓"


def parse_time_close_enabled_form(form_value: Any) -> int:
    return 1 if str(form_value or "").strip().lower() in ("1", "true", "on", "yes") else 0


def parse_time_close_hours_form(form_value: Any, *, default: int = 4) -> Optional[int]:
    raw = str(form_value or "").strip().lower().rstrip("h")
    if not raw:
        return None
    try:
        h = int(float(raw))
    except (TypeError, ValueError):
        return None
    if h in ALLOWED_TIME_CLOSE_HOURS:
        return h
    return None


def normalize_time_close_hours(value: Any) -> Optional[int]:
    try:
        h = int(value)
    except (TypeError, ValueError):
        return None
    return h if h in ALLOWED_TIME_CLOSE_HOURS else None


def _row_val(row: Any, key: str, default=None):
    if row is None:
        return default
    try:
        if hasattr(row, "keys") and key in row.keys():
            return row[key]
    except Exception:
        pass
    if isinstance(row, dict):
        return row.get(key, default)
    return default


def time_close_settings_from_row(row: Any) -> tuple[int, Optional[int], Optional[int]]:
    """返回 (enabled, hours, close_at_ms)。"""
    enabled = int(_row_val(row, "time_close_enabled", 0) or 0) != 0
    hours = normalize_time_close_hours(_row_val(row, "time_close_hours"))
    close_at = _row_val(row, "time_close_at_ms")
    try:
        close_at_ms = int(close_at) if close_at not in (None, "") else None
    except (TypeError, ValueError):
        close_at_ms = None
    if enabled and hours and not close_at_ms:
        opened_ms = _row_val(row, "opened_at_ms")
        try:
            opened_ms = int(opened_ms) if opened_ms not in (None, "") else None
        except (TypeError, ValueError):
            opened_ms = None
        close_at_ms = compute_close_at_ms(opened_ms, hours)
    return (1 if enabled and hours else 0, hours, close_at_ms)


def compute_close_at_ms(opened_at_ms: Any, hours: Any) -> Optional[int]:
    h = normalize_time_close_hours(hours)
    try:
        opened = int(opened_at_ms)
    except (TypeError, ValueError):
        return None
    if not h or opened <= 0:
        return None
    return opened + h * 3600 * 1000


def should_trigger_time_close(row: Any, *, now_ms: Optional[int] = None) -> bool:
    enabled, hours, close_at_ms = time_close_settings_from_row(row)
    if not enabled or not close_at_ms:
        return False
    now = int(now_ms if now_ms is not None else time.time() * 1000)
    return now >= int(close_at_ms)


def time_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_time_close_countdown(seconds: Any) -> str:
    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 time_close_label(hours: Any) -> str:
    h = normalize_time_close_hours(hours)
    return f"时间平仓 {h}h" if h else "时间平仓"


def apply_time_close_to_payload(payload: dict[str, Any], row: Any, *, now_ms: Optional[int] = None) -> None:
    enabled, hours, close_at_ms = time_close_settings_from_row(row)
    payload["time_close_enabled"] = bool(enabled)
    payload["time_close_hours"] = hours
    payload["time_close_at_ms"] = close_at_ms
    payload["time_close_label"] = time_close_label(hours) if enabled else ""
    if enabled and close_at_ms:
        rem = time_close_remaining_seconds(close_at_ms, now_ms=now_ms)
        payload["time_close_remaining_sec"] = rem
        payload["time_close_countdown"] = format_time_close_countdown(rem)
    else:
        payload["time_close_remaining_sec"] = None
        payload["time_close_countdown"] = ""


def ensure_time_close_schema(cursor) -> None:
    ddl_list = (
        "ALTER TABLE order_monitors ADD COLUMN time_close_enabled INTEGER DEFAULT 0",
        "ALTER TABLE order_monitors ADD COLUMN time_close_hours INTEGER",
        "ALTER TABLE order_monitors ADD COLUMN time_close_at_ms INTEGER",
        "ALTER TABLE key_monitors ADD COLUMN time_close_enabled INTEGER DEFAULT 0",
        "ALTER TABLE key_monitors ADD COLUMN time_close_hours INTEGER",
    )
    for ddl in ddl_list:
        try:
            cursor.execute(ddl)
        except Exception:
            pass


def time_close_insert_values(
    enabled: int,
    hours: Optional[int],
    opened_at_ms: Optional[int],
) -> tuple[int, Optional[int], Optional[int]]:
    en = 1 if int(enabled or 0) != 0 and hours else 0
    h = normalize_time_close_hours(hours) if en else None
    close_at = compute_close_at_ms(opened_at_ms, h) if en else None
    return en, h, close_at