"""
kis_trader/ws/kis_ws_overseas.py — KIS 해외주식 실시간(지연)체결가 수신기 (HDFSCNT0)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
역할: 미국 ETF/주식(QQQM·SPYM 등) 영구구독 실시간 체결가를 push 로 받아 캐시/봉집계.
      국내 WS(kis_ws.KISWebSocketPriceCache, H0STCNT0)와 **완전히 분리된 별도 인스턴스**.
      → 국내 라이브 매매 경로(실거래)에 일절 영향 없음 (회귀 위험 0).

왜 분리했나 (구조 결정):
  - 기존 KISWebSocketPriceCache 의 _is_market_hours() 는 국장(08:25~16:05 KST)만 True →
    그 외 시간엔 다음 국장까지 슬립. 그러나 미국장은 한국 '밤'(대략 22:30~06:00, 서머타임
    여부에 따라 ±1h)이라 같은 루프로는 야간에 push 를 못 받음.
  - 따라서 야간(미국장) 가동 전용 루프를 가진 별도 클래스로 구현.

KIS 공식 스펙:
  TR_ID    : HDFSCNT0  (해외주식 실시간지연체결가)
  tr_key   : D + 시세거래소(NAS/NYS/AMS) + 심볼   예) DNASQQQM, DAMSSPYM
  실전 URL : ws://ops.koreainvestment.com:21000
  세션 구독 한도: 최대 41

⚠️ 실시간 vs 지연:
  '해외주식 실시간시세' 신청이 안 돼 있으면 HDFSCNT0 는 **15분 지연**으로 옵니다.
  15분봉 전략에선 신청 필수. 미신청 시에도 동작은 하나 데이터가 지연됩니다.

KIS 안티-어뷰즈 정책 준수: 재연결 횟수/시간 제한 + 지수 백오프 (국내 수신기와 동일 사상).
"""
from __future__ import annotations

import datetime as _dt
import json
import logging
import threading
import time
from typing import Any, Dict, Optional, Set

import requests

logger = logging.getLogger("KISWebSocketOverseas")

try:
    from kis_trader.utils.env import get_env_from_db, get_env_bool, get_env_int
except ImportError:  # 단독 실행 fallback
    def get_env_from_db(key, default=""):  # type: ignore[misc]
        return default

    def get_env_bool(key, default=False):  # type: ignore[misc]
        return default

    def get_env_int(key, default):  # type: ignore[misc]
        return default


class KISOverseasWebSocket:
    """
    KIS HDFSCNT0 해외 실시간(지연)체결가 수신기.

    사용법:
        ws = KISOverseasWebSocket(app_key, app_secret)
        ws.add_symbol("QQQM", "NASD")     # 거래소(NASD/NYSE/AMEX) → tr_key 자동 변환
        ws.add_symbol("SPYM", "AMEX")
        ws.start()
        data = ws.get_price("QQQM")        # inquire_price 호환 dict (None=미수신/만료)
        ws.stop()
    """

    # HDFSCNT0 데이터 필드 인덱스 ('^' 구분, 0-based) — KIS 해외 실시간지연체결가 전문
    IDX_RSYM  = 0    # 실시간종목코드 (DNASQQQM)
    IDX_SYMB  = 1    # 종목코드 (QQQM)
    IDX_ZDIV  = 2    # 소수점자리수
    IDX_KHMS  = 7    # 한국시간 (HHMMSS)
    IDX_OPEN  = 8    # 시가
    IDX_HIGH  = 9    # 고가
    IDX_LOW   = 10   # 저가
    IDX_LAST  = 11   # 현재가(체결가)
    IDX_SIGN  = 12   # 대비구분
    IDX_DIFF  = 13   # 전일대비
    IDX_RATE  = 14   # 등락율
    IDX_TVOL  = 20   # 거래량(누적)

    MAX_SUBSCRIPTIONS        = 41
    MAX_RECONNECT_ATTEMPTS   = 10
    MAX_RECONNECTS_PER_HOUR  = 6
    RECONNECT_BASE_DELAY_SEC = 5.0
    RECONNECT_MAX_DELAY_SEC  = 300.0
    STABLE_CONN_RESET_SEC    = 300.0
    APPROVAL_KEY_CACHE_SEC   = 82800

    # 시세 거래소 코드 매핑 (주문용 → 시세용)
    _EXCD_MAP = {
        "NASD": "NAS", "NAS": "NAS",
        "NYSE": "NYS", "NYS": "NYS",
        "AMEX": "AMS", "AMS": "AMS",
    }

    def __init__(self, app_key: str, app_secret: str):
        self.app_key    = app_key
        self.app_secret = app_secret

        self._ws_url = get_env_from_db(
            "KIS_WS_URL_REAL", "ws://ops.koreainvestment.com:21000"
        )
        self._base_url = "https://openapi.koreainvestment.com:9443"

        # 가격 캐시 { code: {"data": dict, "ts": float} }
        self._cache: Dict[str, Dict] = {}
        self._cache_lock = threading.Lock()

        # 구독 종목 메타 { code: tr_key }  (예: QQQM → DNASQQQM)
        self._symbols: Dict[str, str] = {}
        self._sym_lock = threading.Lock()
        # tr_key → code 역매핑 (메시지 파싱용)
        self._key2code: Dict[str, str] = {}

        self._ws = None
        self._ws_thread: Optional[threading.Thread] = None
        self._running   = False
        self._connected = False

        self._approval_key: Optional[str] = None
        self._approval_key_ts: float      = 0.0

        self._reconnect_count   = 0
        self._reconnect_times: list = []
        self._reconnect_delay   = self.RECONNECT_BASE_DELAY_SEC
        self._last_connect_time = 0.0

        self._candle_agg: Optional[Any] = None
        self._tick_recorder: Optional[Any] = None

        try:
            import websocket as _ws_lib
            self._ws_lib    = _ws_lib
            self._available = True
        except ImportError:
            self._ws_lib    = None
            self._available = False
            logger.warning("⚠️ websocket-client 미설치 → 해외 WS 비활성 (pip install websocket-client)")

    # ==================================================================
    # Public API
    # ==================================================================
    def attach_candle_aggregator(self, agg: Any) -> None:
        self._candle_agg = agg
        logger.info("✅ 해외 CandleAggregator 연결 완료 (봉 집계 활성화)")

    def attach_tick_recorder(self, recorder: Any) -> None:
        self._tick_recorder = recorder
        logger.info("✅ 해외 TickRecorder 연결 완료 (HDFSCNT0)")

    def _make_tr_key(self, exchange: str, symbol: str) -> str:
        ex   = str(exchange or "NASD").strip().upper()
        excd = self._EXCD_MAP.get(ex, "NAS")
        sym  = str(symbol or "").strip().upper()
        return f"D{excd}{sym}"

    def add_symbol(self, code: str, exchange: str = "NASD", symbol: Optional[str] = None) -> None:
        """구독할 해외 종목 등록 (start 전/후 모두 가능)."""
        code = str(code or "").strip().upper()
        if not code:
            return
        sym = str(symbol or code).strip().upper()
        tr_key = self._make_tr_key(exchange, sym)
        with self._sym_lock:
            if code in self._symbols:
                return
            if len(self._symbols) >= self.MAX_SUBSCRIPTIONS:
                logger.warning("⚠️ 해외 WS 구독 한도 초과(%d) → %s 거부", self.MAX_SUBSCRIPTIONS, code)
                return
            self._symbols[code] = tr_key
            self._key2code[tr_key] = code
        if self._connected and self._ws:
            self._send_sub_msg(tr_key, subscribe=True)
        logger.info("📡 해외 WS 구독 추가: %s (tr_key=%s)", code, tr_key)

    def remove_symbol(self, code: str) -> None:
        code = str(code or "").strip().upper()
        with self._sym_lock:
            tr_key = self._symbols.pop(code, None)
            if tr_key:
                self._key2code.pop(tr_key, None)
        with self._cache_lock:
            self._cache.pop(code, None)
        if tr_key and self._connected and self._ws:
            self._send_sub_msg(tr_key, subscribe=False)

    def get_price(self, code: str, max_age_sec: float = 1200.0) -> Optional[Dict]:
        """
        해외 캐시에서 가격 dict 반환 (inquire_price 호환 키).
        ※ 기본 max_age 1200초(20분): 미국 실시간 미신청 시 15분 지연 + 거래 공백 고려.
        """
        if not self._available or not self._connected:
            return None
        code = str(code or "").strip().upper()
        with self._cache_lock:
            entry = self._cache.get(code)
        if not entry:
            return None
        if (time.time() - entry.get("ts", 0)) > max_age_sec:
            return None
        return entry.get("data")

    @property
    def is_active(self) -> bool:
        return bool(self._available and self._connected and self._running)

    def start(self, force_cleanup: bool = True) -> bool:
        if not self._available:
            return False
        if self._running:
            return True
        if force_cleanup:
            self._approval_key = None
            self._approval_key_ts = 0.0
            with self._cache_lock:
                self._cache.clear()
        if not self._get_approval_key():
            logger.warning("⚠️ 해외 WS approval_key 발급 실패 → 비활성")
            return False
        self._running = True
        self._ws_thread = threading.Thread(
            target=self._run_ws_loop, daemon=True, name="KIS-WS-HDFSCNT0"
        )
        self._ws_thread.start()
        logger.info("✅ KIS 해외 WebSocket 수신 스레드 시작 (HDFSCNT0 | url=%s)", self._ws_url)
        return True

    def stop(self) -> None:
        self._running   = False
        self._connected = False
        if self._ws:
            try:
                self._ws.close()
            except Exception:
                pass
        if self._ws_thread and self._ws_thread.is_alive():
            self._ws_thread.join(timeout=5)
        logger.info("🛑 KIS 해외 WebSocket 종료")

    # ==================================================================
    # 내부: approval_key / 구독 메시지 / 파싱
    # ==================================================================
    def _get_approval_key(self) -> Optional[str]:
        now = time.time()
        if self._approval_key and (now - self._approval_key_ts) < self.APPROVAL_KEY_CACHE_SEC:
            return self._approval_key
        try:
            url  = f"{self._base_url}/oauth2/Approval"
            body = {"grant_type": "client_credentials", "appkey": self.app_key, "secretkey": self.app_secret}
            r    = requests.post(url, json=body, timeout=10)
            key  = r.json().get("approval_key")
            if key:
                self._approval_key    = key
                self._approval_key_ts = now
                logger.info("✅ 해외 WS approval_key 발급 완료 (앞8자: %s…)", key[:8])
                return key
            logger.error("❌ 해외 WS approval_key 발급 실패: %s", r.text[:200])
        except Exception as e:
            logger.error("❌ 해외 WS approval_key 요청 예외: %s", e)
        return None

    def _build_sub_payload(self, tr_key: str, subscribe: bool) -> str:
        return json.dumps({
            "header": {
                "approval_key": self._approval_key or "",
                "custtype":     "P",
                "tr_type":      "1" if subscribe else "2",
                "content-type": "utf-8",
            },
            "body": {"input": {"tr_id": "HDFSCNT0", "tr_key": tr_key}},
        })

    def _send_sub_msg(self, tr_key: str, subscribe: bool = True) -> None:
        if not self._ws:
            return
        try:
            self._ws.send(self._build_sub_payload(tr_key, subscribe))
        except Exception as e:
            logger.debug("해외 구독 메시지 전송 실패(%s): %s", tr_key, e)

    def _parse_realtime_msg(self, raw: str) -> None:
        """HDFSCNT0 실시간(지연)체결가 파싱 → 캐시 갱신."""
        if not raw:
            return
        # KIS Application-Level PINGPONG
        if raw.strip() == "PINGPONG":
            if self._ws:
                try:
                    self._ws.send("PINGPONG")
                except Exception:
                    pass
            return
        # 구독 응답(JSON) 무시
        if raw.startswith("{"):
            try:
                j = json.loads(raw)
                if j.get("header", {}).get("tr_id") == "HDFSCNT0":
                    b = j.get("body", {})
                    logger.debug("HDFSCNT0 구독 응답: rt_cd=%s msg=%s", b.get("rt_cd"), b.get("msg1"))
            except Exception:
                pass
            return

        parts = raw.split("|")
        if len(parts) < 4 or parts[1] != "HDFSCNT0":
            return
        if parts[0] == "1":  # 암호화 미지원
            logger.debug("HDFSCNT0 암호화 데이터 수신 (스킵)")
            return

        fields = parts[3].split("^")
        if len(fields) <= self.IDX_LAST:
            return
        try:
            tr_key = fields[self.IDX_RSYM].strip()
            symb   = fields[self.IDX_SYMB].strip().upper()
            code   = self._key2code.get(tr_key) or symb
            price  = float(fields[self.IDX_LAST])
            if not code or price <= 0:
                return

            def _g(i, d="0"):
                return fields[i] if len(fields) > i else d

            data_compat = {
                "stck_prpr": str(price),          # 현재가 (해외는 소수점 → 문자열 그대로)
                "prdy_vrss": _g(self.IDX_DIFF),
                "prdy_ctrt": _g(self.IDX_RATE, "0.00"),
                "stck_oprc": _g(self.IDX_OPEN),
                "stck_hgpr": _g(self.IDX_HIGH),
                "stck_lwpr": _g(self.IDX_LOW),
                "last":      str(price),          # 해외 표기 호환
            }
            with self._cache_lock:
                self._cache[code] = {"data": data_compat, "ts": time.time()}

            if self._candle_agg is not None or self._tick_recorder is not None:
                khms = _g(self.IDX_KHMS, "").strip()
                try:
                    tick_vol = int(float(_g(self.IDX_TVOL, "0")))
                except ValueError:
                    tick_vol = 0
                if self._candle_agg is not None:
                    self._candle_agg.on_tick(code, price, tick_vol, khms)
                if self._tick_recorder is not None:
                    try:
                        self._tick_recorder.on_tick(code, price, tick_vol, khms, source="kis_us")
                    except Exception as ex:
                        logger.debug("HDFSCNT0→TickRecorder 실패 %s: %s", code, ex)

            logger.debug("HDFSCNT0 수신: %s → %s", code, price)
        except (ValueError, IndexError) as e:
            logger.debug("HDFSCNT0 파싱 오류: %s | raw=%s", e, raw[:80])

    # ==================================================================
    # 미국장 시간대 게이팅
    # ==================================================================
    def _is_us_market_window(self) -> bool:
        """
        미국 정규장(+프리/애프터 여유)을 한국시간 기준으로 넉넉히 커버.
        서머타임(EDT)·표준시(EST) 모두 포함하도록 22:00~06:30(KST) 윈도우 사용.
        - 22:00~23:59: 월~금(미국장 당일 개장)
        - 00:00~06:30: 화~토(미국장 익일 새벽 마감)
        """
        now = _dt.datetime.now()
        wd  = now.weekday()  # 월=0 … 일=6
        t   = now.time()
        if _dt.time(22, 0) <= t <= _dt.time(23, 59, 59):
            return 0 <= wd <= 4          # 월~금 밤
        if _dt.time(0, 0) <= t <= _dt.time(6, 30):
            return 1 <= wd <= 5          # 화~토 새벽
        return False

    def _seconds_until_us_open(self) -> float:
        """다음 미국장 윈도우(22:00 KST)까지 남은 초 (최소 60)."""
        now = _dt.datetime.now()
        target = now.replace(hour=22, minute=0, second=0, microsecond=0)
        if now.time() > _dt.time(6, 30):
            if now.time() >= _dt.time(22, 0):
                target += _dt.timedelta(days=1)
        else:
            target = now.replace(hour=22, minute=0, second=0, microsecond=0)
        # 금요일 밤~토요일 새벽 이후 주말은 다음 월요일로
        while target.weekday() >= 5:  # 토·일 시작은 건너뜀
            target += _dt.timedelta(days=1)
        return max(60.0, (target - now).total_seconds())

    # ==================================================================
    # WebSocket 메인 루프
    # ==================================================================
    def _run_ws_loop(self) -> None:
        INSTANT_DROP_SEC = 3.0
        INSTANT_DROP_MAX = 3
        _instant_streak  = 0

        while self._running:
            now = time.time()

            if not self._is_us_market_window():
                wait_sec = self._seconds_until_us_open()
                logger.info("🌙 미국장 외 — 해외 WS 대기 (다음 윈도우까지 %.0f분)", wait_sec / 60)
                self._reconnect_count = 0
                self._reconnect_times = []
                self._reconnect_delay = self.RECONNECT_BASE_DELAY_SEC
                _instant_streak = 0
                for _ in range(int(wait_sec // 60)):
                    if not self._running:
                        return
                    time.sleep(60)
                time.sleep(wait_sec % 60)
                continue

            if _instant_streak >= INSTANT_DROP_MAX:
                wait_sec = self._seconds_until_us_open()
                logger.warning("⚠️ 해외 WS 연속 즉시 끊김 %d회 → %.0f분 대기", _instant_streak, wait_sec / 60)
                self._reconnect_count = 0
                self._reconnect_times = []
                _instant_streak = 0
                time.sleep(min(300, wait_sec))
                continue

            if (self._last_connect_time > 0
                    and (now - self._last_connect_time) > self.STABLE_CONN_RESET_SEC
                    and self._reconnect_count > 0):
                self._reconnect_count = 0
                self._reconnect_times = []
                self._reconnect_delay = self.RECONNECT_BASE_DELAY_SEC

            self._reconnect_times = [t for t in self._reconnect_times if now - t < 3600]
            if len(self._reconnect_times) >= self.MAX_RECONNECTS_PER_HOUR:
                wait = max(10, 3600 - (now - self._reconnect_times[0]) + 10)
                logger.warning("⛔ 해외 WS 1시간 재연결 한도 초과 → %.0f분 대기", wait / 60)
                time.sleep(wait)
                continue
            if self._reconnect_count >= self.MAX_RECONNECT_ATTEMPTS:
                logger.error("❌ 해외 WS 최대 재연결 초과 → 종료")
                self._running = False
                break

            if not self._get_approval_key():
                time.sleep(self._reconnect_delay)
                self._reconnect_delay = min(self._reconnect_delay * 2, self.RECONNECT_MAX_DELAY_SEC)
                continue

            self._reconnect_count += 1
            if self._reconnect_count > 1:
                self._reconnect_times.append(time.time())
                time.sleep(self._reconnect_delay)
                self._reconnect_delay = min(self._reconnect_delay * 2, self.RECONNECT_MAX_DELAY_SEC)

            _conn_start = time.time()
            try:
                ws_app = self._ws_lib.WebSocketApp(
                    self._ws_url,
                    on_open    = self._on_open,
                    on_message = self._on_message,
                    on_error   = self._on_error,
                    on_close   = self._on_close,
                )
                self._ws = ws_app
                ws_app.run_forever(ping_interval=20, ping_timeout=10)
            except Exception as e:
                logger.error("해외 WS run_forever 예외: %s", e)

            if not self._running:
                break
            if (time.time() - _conn_start) < INSTANT_DROP_SEC:
                _instant_streak += 1
            else:
                _instant_streak = 0

        logger.info("KIS 해외 WebSocket 루프 종료")

    def _on_open(self, ws) -> None:
        self._connected         = True
        self._reconnect_delay   = self.RECONNECT_BASE_DELAY_SEC
        self._last_connect_time = time.time()
        logger.info("✅ KIS 해외 WebSocket 연결 성공 (HDFSCNT0)")
        with self._sym_lock:
            keys = list(self._symbols.values())
        for tr_key in keys:
            self._send_sub_msg(tr_key, subscribe=True)
        if keys:
            logger.info("📡 해외 WS 구독 일괄 등록: %s", ", ".join(keys))

    def _on_message(self, ws, message: str) -> None:
        self._parse_realtime_msg(message)

    def _on_error(self, ws, error) -> None:
        self._connected = False
        logger.warning("⚠️ KIS 해외 WebSocket 오류: %s", error)

    def _on_close(self, ws, close_status_code, close_msg) -> None:
        self._connected = False
        logger.info("🔌 KIS 해외 WebSocket 종료 (code=%s msg=%s)", close_status_code, close_msg or "")
