kis_bot/kis_trader/utils/ls_ws_session_windows.py

"""
LS WebSocket 세션 시간 분할 (국내 ↔ 해외) — 벽시계 기준
======================================================
LS 는 KIS approval 과 다름:
  - 접근토큰 1개 + WS URL 1개로 국내(US3) · 해외(GSC) 동시 구독 가능
  - 세션 전환 시 **토큰 재발급 금지** (재발급하면 조건검색 REST·WS 공용 토큰이 무효화됨)
  - 이 모듈은 **소켓 hold / 대기 / 워치독 게이트** 만 담당 (oauth 호출 없음)

운용 (KIS ``kis_ws_session_windows`` 와 동일 시각 철학):
  - 국내 hold: 기본 07:00~20:00 (장 전후 여유) — 소켓 유지·국장 워치독은 별도 정규창
  - 해외 hold: 기본 21:00~06:00 — 해외 구독이 있을 때만 소켓 유지 사유
  - 갭(20:00~21:00, 06:00~07:00): 소켓 close 후 대기 (**재발급 없음**)

env (DB 등록):
  LS_WS_KR_HOLD_START_HM / LS_WS_KR_HOLD_END_HM   (기본 700 / 2000)
  LS_WS_US_HOLD_START_HM / LS_WS_US_HOLD_END_HM   (기본 2100 / 600)
  LS_WS_SESSION_GUARD_SEC
"""
from __future__ import annotations

import datetime as _dt
from typing import Optional, Tuple

from .env import get_env_float, get_env_int
from .session_hm import hm_in_trading_window


def kr_ws_hold_bounds() -> Tuple[int, int]:
    """국내 LS WS 세션 점유 HHMM (당일 구간). 기본 0700~2000."""
    start = int(get_env_int("LS_WS_KR_HOLD_START_HM", 700) or 700)
    end = int(get_env_int("LS_WS_KR_HOLD_END_HM", 2000) or 2000)
    if start <= 0:
        start = 700
    if end <= 0:
        end = 2000
    return start, end


def us_ws_hold_bounds() -> Tuple[int, int]:
    """해외 LS WS 세션 점유 HHMM (자정 넘김). 기본 2100~0600."""
    start = int(get_env_int("LS_WS_US_HOLD_START_HM", 2100) or 2100)
    end = int(get_env_int("LS_WS_US_HOLD_END_HM", 600) or 600)
    if start <= 0:
        start = 2100
    if end <= 0:
        end = 600
    return start, end


def session_guard_interval_sec() -> float:
    """연결 유지 중 hold 창 이탈 감시 주기(초)."""
    return max(5.0, float(get_env_float("LS_WS_SESSION_GUARD_SEC", 15.0) or 15.0))


def _hm_now(now: Optional[_dt.datetime] = None) -> int:
    n = now or _dt.datetime.now()
    return int(n.hour * 100 + n.minute)


def in_kr_ws_hold_window(now: Optional[_dt.datetime] = None) -> bool:
    """True = 국내 LS WS 소켓을 유지해도 되는 시간 (월~금)."""
    n = now or _dt.datetime.now()
    if n.weekday() >= 5:
        return False
    start, end = kr_ws_hold_bounds()
    return hm_in_trading_window(_hm_now(n), start, end, wrap_midnight=False)


def in_us_ws_hold_window(now: Optional[_dt.datetime] = None) -> bool:
    """
    True = 해외 LS WS 소켓 유지 사유가 되는 시간.
    - start~23:59: 월~금
    - 00:00~end: 화~토 (미국장 새벽 마감)
    """
    n = now or _dt.datetime.now()
    wd = n.weekday()
    hm = _hm_now(n)
    start, end = us_ws_hold_bounds()
    if start > end:
        if hm >= start:
            return 0 <= wd <= 4
        if hm < end:
            return 1 <= wd <= 5
        return False
    if n.weekday() >= 5:
        return False
    return hm_in_trading_window(hm, start, end, wrap_midnight=False)


def should_hold_ls_socket(
    *,
    n_us_subscribed: int = 0,
    now: Optional[_dt.datetime] = None,
) -> bool:
    """소켓을 열어 둘지. 해외 hold 는 해외 구독이 있을 때만."""
    n = now or _dt.datetime.now()
    if in_kr_ws_hold_window(n):
        return True
    if int(n_us_subscribed or 0) > 0 and in_us_ws_hold_window(n):
        return True
    return False


def seconds_until_kr_ws_open(now: Optional[_dt.datetime] = None) -> float:
    """다음 국내 hold 시작까지 초 (최소 60). hold 중이면 짧은 쿨다운."""
    n = now or _dt.datetime.now()
    if in_kr_ws_hold_window(n):
        return max(30.0, float(get_env_float("LS_WS_HOLD_INNER_COOLDOWN_SEC", 90.0) or 90.0))
    start, _end = kr_ws_hold_bounds()
    sh, sm = divmod(int(start), 100)
    hm = _hm_now(n)
    target = n.replace(hour=sh, minute=sm, second=0, microsecond=0)
    if n.weekday() >= 5:
        days = (0 - n.weekday()) % 7
        if days == 0:
            days = 7
        target += _dt.timedelta(days=days)
    elif hm >= start:
        target += _dt.timedelta(days=1)
        while target.weekday() >= 5:
            target += _dt.timedelta(days=1)
    return max(60.0, (target - n).total_seconds())


def seconds_until_us_ws_open(now: Optional[_dt.datetime] = None) -> float:
    """다음 해외 hold 시작까지 초 (최소 60). hold 중이면 짧은 쿨다운."""
    n = now or _dt.datetime.now()
    if in_us_ws_hold_window(n):
        return max(30.0, float(get_env_float("LS_WS_HOLD_INNER_COOLDOWN_SEC", 90.0) or 90.0))
    start, end = us_ws_hold_bounds()
    sh, sm = divmod(int(start), 100)
    hm = _hm_now(n)
    target = n.replace(hour=sh, minute=sm, second=0, microsecond=0)
    if start > end:
        if end <= hm < start:
            if n.weekday() >= 5:
                days = (0 - n.weekday()) % 7
                if days == 0:
                    days = 7
                target += _dt.timedelta(days=days)
        elif hm >= start:
            target += _dt.timedelta(days=1)
            while target.weekday() >= 5:
                target += _dt.timedelta(days=1)
        else:
            if n.weekday() == 6:
                target += _dt.timedelta(days=1)
            while target.weekday() >= 5:
                target += _dt.timedelta(days=1)
    else:
        if hm >= start or n.weekday() >= 5:
            target += _dt.timedelta(days=1)
        while target.weekday() >= 5:
            target += _dt.timedelta(days=1)
    return max(60.0, (target - n).total_seconds())


def seconds_until_ls_socket_open(
    *,
    n_us_subscribed: int = 0,
    now: Optional[_dt.datetime] = None,
) -> float:
    """다음 소켓 hold 시작까지 초. 해외 구독 없으면 국내만."""
    n = now or _dt.datetime.now()
    if should_hold_ls_socket(n_us_subscribed=n_us_subscribed, now=n):
        return max(30.0, float(get_env_float("LS_WS_HOLD_INNER_COOLDOWN_SEC", 90.0) or 90.0))
    wait = seconds_until_kr_ws_open(n)
    if int(n_us_subscribed or 0) > 0:
        wait = min(wait, seconds_until_us_ws_open(n))
    return float(wait)