kis_bot/kis_trader/network/ls_ws_validator.py

"""
kis_trader/network/ls_ws_validator.py — KIS/키움 ↔ LS 시세 갭 검증
================================================================
``LS_WS_VALIDATION_ENABLED=true`` 일 때만 기동.
실매 시세 경로는 건드리지 않음 — 구독 동기화 + DB 비교 INSERT 만.
"""
from __future__ import annotations

import threading
import time
from typing import Optional, Set

from ..utils.env import get_env_float, get_env_int
from ..utils.logger import get_logger

logger = get_logger("kis_trader.ls_ws_validator")


class LSWSPriceValidator:
    """KIS(+키움) 구독 종목을 LS 에도 맞추고 가격 갭을 기록."""

    def __init__(
        self,
        *,
        ls_ws,
        db,
        kis_ws=None,
        kiwoom_ws=None,
    ) -> None:
        self.ls_ws = ls_ws
        self.db = db
        self.kis_ws = kis_ws
        self.kiwoom_ws = kiwoom_ws
        self._thread: Optional[threading.Thread] = None
        self._running = False
        self._last_warn_ts: dict = {}

    def start(self) -> bool:
        if self._thread and self._thread.is_alive():
            return True
        self._running = True
        self._thread = threading.Thread(
            target=self._loop, daemon=True, name="LSWSPriceValidator",
        )
        self._thread.start()
        logger.info(
            "✅ LS WS 갭 검증기 시작 (interval=%ds warn≥%.2f%%)",
            self._interval_sec(),
            self._warn_pct(),
        )
        return True

    def stop(self) -> None:
        self._running = False

    def _interval_sec(self) -> int:
        return max(1, get_env_int("LS_WS_VALIDATION_INTERVAL_SEC", 5))

    def _warn_pct(self) -> float:
        return max(0.0, get_env_float("LS_WS_VALIDATION_DIFF_WARN_PCT", 0.10))

    def _codes_from(self, ws) -> Set[str]:
        if ws is None:
            return set()
        try:
            with ws._sub_lock:  # type: ignore[attr-defined]
                return set(ws._subscribed)  # type: ignore[attr-defined]
        except Exception:
            return set()

    def _sync_ls(self, target: Set[str]) -> None:
        try:
            with self.ls_ws._sub_lock:
                current = set(self.ls_ws._subscribed)
            # KR 6자리만 동기화 (US 는 영구구독에서 별도 붙일 수 있음)
            kr_target = {c for c in target if c.isdigit() and len(c) == 6}
            for code in kr_target - current:
                self.ls_ws.subscribe(code)
            for code in current - kr_target:
                self.ls_ws.unsubscribe(code)
        except Exception as e:
            logger.debug("LS 구독 동기화 실패: %s", e)

    def _parse_price(self, data, key: str = "stck_prpr") -> Optional[float]:
        if not data:
            return None
        raw = data.get(key)
        if raw is None and "_price_f" in data:
            raw = data.get("_price_f")
        try:
            v = float(str(raw).replace(",", ""))
            return v if v > 0 else None
        except (TypeError, ValueError):
            return None

    def _parse_age_ms(self, data) -> Optional[int]:
        if not data:
            return None
        try:
            return int(data.get("_age_ms"))
        except (TypeError, ValueError):
            return None

    def _loop(self) -> None:
        time.sleep(15)
        while self._running:
            try:
                self._tick()
            except Exception as e:
                logger.warning("LS 검증기 tick 예외: %s", e)
            time.sleep(self._interval_sec())

    def _tick(self) -> None:
        kis_codes = self._codes_from(self.kis_ws)
        kw_codes = self._codes_from(self.kiwoom_ws)
        codes = kis_codes | kw_codes
        if not codes:
            return
        self._sync_ls(codes)
        if not self.ls_ws.is_connected():
            return

        warn_pct = self._warn_pct()
        n = 0
        for code in codes:
            if not (code.isdigit() and len(code) == 6):
                continue
            kis_data = (
                self.kis_ws.get_price(code, max_age_sec=10.0)
                if self.kis_ws is not None
                else None
            )
            kw_data = (
                self.kiwoom_ws.get_price(code, max_age_sec=10.0)
                if self.kiwoom_ws is not None and self.kiwoom_ws.is_connected()
                else None
            )
            ls_data = self.ls_ws.get_price(code, max_age_sec=10.0)

            kis_p = self._parse_price(kis_data)
            kw_p = self._parse_price(kw_data)
            ls_p = self._parse_price(ls_data)
            if ls_p is None and kis_p is None and kw_p is None:
                continue

            self.db.insert_ws_price_validation_ls(
                code=code,
                kis_price=kis_p,
                kiwoom_price=kw_p,
                ls_price=ls_p,
                kis_age_ms=self._parse_age_ms(kis_data),
                kiwoom_age_ms=self._parse_age_ms(kw_data),
                ls_age_ms=self._parse_age_ms(ls_data),
            )
            n += 1

            if kis_p not in (None, 0) and ls_p is not None:
                diff = (ls_p - kis_p) / kis_p * 100.0
                if abs(diff) >= warn_pct:
                    now = time.time()
                    if now - self._last_warn_ts.get(code, 0) >= 60:
                        self._last_warn_ts[code] = now
                        logger.warning(
                            "⚠️ [LS 갭] %s KIS↔LS %.3f%%  (KIS=%.0f LS=%.0f)",
                            code, diff, kis_p, ls_p,
                        )
        if n > 0:
            logger.debug("LS 갭 검증 샘플 %d건", n)