"""
kis_trader/execution/kis_client.py — 한국투자증권 REST 클라이언트 (통합)
==========================================================================
기존 kis_scalping_ver2.KISClient / kis_short_ver3.KISClient 의 공통 기능을
SafeRequest 기반으로 통합. 두 봇이 독립 토큰을 발급받아 충돌나던 문제를
`kis_token_manager` 위임으로 해결.

제공 기능 (매매 봇이 쓰는 최소 세트):
  - inquire_price              : 현재가
  - get_account_balance        : 계좌 잔고 (output1=종목별, output2=예수금)
  - get_broker_holdings_map    : 잔고 → {code: {qty, avg_price, name}} 맵
  - get_minute_chart           : 분봉 (갭보정용)
  - get_daily_chart            : 일봉 (대/중/소형주 판정용)
  - get_orderbook              : 호가 잔량 (익절 지정가용)
  - buy_order / sell_order     : 매수·매도 주문 (ODNO 반환)
  - get_execution_by_odno      : 주문번호로 체결 확인
"""
from __future__ import annotations

import datetime
import logging
from datetime import datetime as dt
from pathlib import Path
from typing import Dict, List, Optional

import pandas as pd

from ..utils.env import (
    get_env_bool,
    get_env_float,
    get_env_from_db,
    get_env_int,
)
from ..utils.logger import get_logger
from ..utils.request_handler import SafeRequest

logger = get_logger("kis_trader.kis_client")

# 토큰 캐시 경로 (프로젝트 루트와 동일 위치 공유)
_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent


class KISClient(SafeRequest):
    """한국투자증권 REST API 통합 클라이언트. SafeRequest 상속."""

    REAL_BASE = "https://openapi.koreainvestment.com:9443"
    MOCK_BASE = "https://openapivts.koreainvestment.com:29443"

    def __init__(
        self,
        *,
        mock: Optional[bool] = None,
        app_key: Optional[str] = None,
        app_secret: Optional[str] = None,
        account_no: Optional[str] = None,
        account_code: Optional[str] = None,
    ):
        # ── 도메인별 REST 최소 호출 간격 ──────────────────────────────
        # 한투 유량(계좌·앱키 단위): 실전 18건/초, 모의 1건/초.
        #   · 모의 도메인(거래) → KIS_MIN_INTERVAL_SEC_MOCK (폴백 KIS_MIN_INTERVAL_SEC=0.22)
        #     ※ 모의 거래 REST는 429 로그 근거가 없어 0.22 유지. 한투 공지 1건/초 엄수가 필요하면 1.0 으로.
        #   · 실전 도메인(시세·운영 통합) → KIS_MIN_INTERVAL_SEC_REAL (기본 0.12초 ≈ 8건/초, 18 한도 여유)
        # 매 호출 간격은 _current_min_interval() 가 env 핫리로드(1초 TTL)로 재해석 → 재시작 없이 반영.
        mock_flag = bool(get_env_bool("KIS_MOCK", True)) if mock is None else bool(mock)
        _legacy_interval = get_env_float("KIS_MIN_INTERVAL_SEC", 0.22)
        if mock_flag:
            _min_interval = get_env_float("KIS_MIN_INTERVAL_SEC_MOCK", _legacy_interval)
        else:
            _min_interval = get_env_float("KIS_MIN_INTERVAL_SEC_REAL", min(_legacy_interval, 0.12))
        super().__init__(
            min_interval_sec=_min_interval,
            max_retries=get_env_int("KIS_REST_MAX_RETRIES", 5),
            backoff_base=1.0,
            backoff_cap=get_env_float("KIS_REST_BACKOFF_CAP_SEC", 8.0),
            timeout_sec=get_env_float("KIS_REST_TIMEOUT_SEC", 10.0),
        )

        self.mock = mock_flag
        # 핫리로드 캐시: env 매 호출 조회 비용 줄이려 1초 TTL (재시작 없이 반영)
        self._interval_cache: float = _min_interval
        self._interval_cache_ts: float = 0.0
        if self.mock:
            self.app_key = app_key or get_env_from_db("KIS_APP_KEY_MOCK", "")
            self.app_secret = app_secret or get_env_from_db("KIS_APP_SECRET_MOCK", "")
            self.account_no = account_no or get_env_from_db("KIS_ACCOUNT_NO_MOCK", "")
            self.account_code = account_code or get_env_from_db(
                "KIS_ACCOUNT_CODE_MOCK", "01"
            )
        else:
            self.app_key = app_key or get_env_from_db("KIS_APP_KEY_REAL", "")
            self.app_secret = app_secret or get_env_from_db("KIS_APP_SECRET_REAL", "")
            self.account_no = account_no or get_env_from_db("KIS_ACCOUNT_NO_REAL", "")
            self.account_code = account_code or get_env_from_db(
                "KIS_ACCOUNT_CODE_REAL", "01"
            )

        self.base_url = self.MOCK_BASE if self.mock else self.REAL_BASE
        self._token: Optional[str] = None
        # 마지막 주문 실패 원인 (매매불가, 잔고없음 등 분기)
        self._last_order_msg_cd: Optional[str] = None
        self._last_order_msg1: Optional[str] = None
        self._last_sell_msg_cd: Optional[str] = None
        self._last_sell_msg1: Optional[str] = None

        self._init_token()

    # ------------------------------------------------------------------
    # 토큰
    # ------------------------------------------------------------------
    def _current_min_interval(self) -> float:
        """REST 최소 호출 간격 — 도메인(mock/real)별 env 핫리로드 (1초 TTL 캐시).

        운영설정에서 KIS_MIN_INTERVAL_SEC_MOCK/REAL 변경 시 재시작 없이 반영.
        """
        import time as _t
        now = _t.time()
        if now - self._interval_cache_ts < 1.0:
            return self._interval_cache
        legacy = get_env_float("KIS_MIN_INTERVAL_SEC", 0.22)
        if self.mock:
            val = get_env_float("KIS_MIN_INTERVAL_SEC_MOCK", legacy)
        else:
            val = get_env_float("KIS_MIN_INTERVAL_SEC_REAL", min(legacy, 0.12))
        val = max(0.0, float(val))
        self._interval_cache = val
        self._interval_cache_ts = now
        self.min_interval_sec = val  # get_throttle_stats 표시 일관성
        return val

    def _init_token(self) -> None:
        """kis_token_manager 경로로만 발급 (23시간 캐시, 파일 잠금 준수)."""
        try:
            from kis_token_manager import KisTokenManager, ensure_token

            # 캐시가 유효하면 즉시 사용, 만료면 재발급
            ensure_token(self.mock)
            self._token = KisTokenManager.instance(is_mock=self.mock).get_token()
        except Exception as e:
            logger.warning("kis_token_manager 연동 실패: %s", e)
            self._token = None

    def _refresh_token_if_needed(self) -> None:
        """호출 직전 토큰 만료 임박 시 선제 갱신."""
        try:
            from kis_token_manager import KisTokenManager

            fresh = KisTokenManager.instance(is_mock=self.mock).get_token()
            if fresh:
                self._token = fresh
        except Exception:
            pass

    # ------------------------------------------------------------------
    # 저수준 GET/POST
    # ------------------------------------------------------------------
    def _headers(self, tr_id: str, is_post: bool = False, tr_cont: str = "") -> dict:
        h = {
            "authorization": f"Bearer {self._token}",
            "appkey": self.app_key,
            "appsecret": self.app_secret,
            "tr_id": tr_id,
            "custtype": "P",
        }
        # 연속조회(페이징) 시에만 tr_cont 헤더 추가 — 빈값이면 미추가(기존 동작 100% 보존)
        if tr_cont:
            h["tr_cont"] = tr_cont
        if is_post:
            h["content-type"] = "application/json; charset=utf-8"
        return h

    def _get(self, path: str, tr_id: str, params: dict, tr_cont: str = ""):
        self._refresh_token_if_needed()
        return self.get(
            self.base_url + path,
            headers=self._headers(tr_id, tr_cont=tr_cont),
            params=params,
        )

    def _post(self, path: str, tr_id: str, body: dict):
        self._refresh_token_if_needed()
        return self.post(
            self.base_url + path,
            headers=self._headers(tr_id, is_post=True),
            json_body=body,
        )

    # ------------------------------------------------------------------
    # 시세
    # ------------------------------------------------------------------
    def get_orderbook(self, code: str) -> Optional[dict]:
        """호가 잔량 [v1_국내주식-009] — 익절 시 매수호가 지정가 판단용."""
        try:
            r = self._get(
                "/uapi/domestic-stock/v1/quotations/inquire-asking-price-exp-ccn",
                "FHKST01010200",
                {
                    "FID_COND_MRKT_DIV_CODE": "J",
                    "FID_INPUT_ISCD": code,
                },
            )
            if r.status_code != 200:
                return None
            j = r.json()
            if j.get("rt_cd") != "0":
                return None
            return j.get("output")
        except Exception as e:
            logger.debug("호가 조회 실패(%s): %s", code, e)
            return None

    def inquire_price(self, code: str) -> Optional[dict]:
        """현재가 [v1_국내주식-007]"""
        r = self._get(
            "/uapi/domestic-stock/v1/quotations/inquire-price",
            "FHKST01010100",
            {"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": code},
        )
        if r.status_code != 200:
            return None
        j = r.json()
        return j.get("output") if j.get("rt_cd") == "0" else None

    def inquire_multi_price(self, codes: List[str]) -> Dict[str, dict]:
        """
        관심종목(멀티종목) 시세조회 [국내주식-205] — FHKST11300006.

        한 번 호출에 **최대 30종목**의 현재가 스냅샷을 받는다(봉 시계열 아님).
        UPDOWN(박스권) REST 피드의 '현재가/하단근접/청산감시' 트리거 전용 —
        종목당 inquire_price 30콜 → 1콜로 대폭 절약(REST 부하·429 방지).

        ※ HTS 관심그룹 등록과 무관하게 임의 코드 리스트를 직접 넣어 조회 가능.
        ※ 30종목 초과 입력 시 앞 30개만 사용(상위 호출자가 배치로 끊어 호출).

        Returns: {code: output_row, ...}
            output_row 주요 키(원본 KIS 필드명 유지):
              inter_shrn_iscd(종목코드), inter2_prpr(현재가),
              inter2_oprc/hgpr/lwpr(당일 시/고/저), inter2_prdy_clpr(전일종가),
              prdy_ctrt(전일대비율), acml_vol(누적거래량)
        """
        result: Dict[str, dict] = {}
        clean = [str(c).strip() for c in (codes or []) if str(c or "").strip()]
        if not clean:
            return result
        # 30종목 초과는 앞 30개만 (상위에서 배치 분할 권장)
        clean = clean[:30]

        params: Dict[str, str] = {}
        for i, code in enumerate(clean, start=1):
            params[f"FID_COND_MRKT_DIV_CODE_{i}"] = "J"  # J=KRX
            params[f"FID_INPUT_ISCD_{i}"] = code
        try:
            r = self._get(
                "/uapi/domestic-stock/v1/quotations/intstock-multprice",
                "FHKST11300006",
                params,
            )
            if r.status_code != 200:
                logger.debug("멀티시세 HTTP %s", r.status_code)
                return result
            j = r.json()
            if j.get("rt_cd") != "0":
                logger.debug("멀티시세 rt_cd=%s msg=%s", j.get("rt_cd"), j.get("msg1"))
                return result
            for row in (j.get("output") or []):
                rc = str(row.get("inter_shrn_iscd", "") or "").strip()
                if rc:
                    result[rc] = row
        except Exception as e:
            logger.debug("멀티시세 조회 실패: %s", e)
        return result

    def inquire_index_price(self, index_code: str = "0001") -> Optional[dict]:
        """
        국내 업종지수 현재지수 [v1_국내주식-066] — MarketGuard 전용.

        index_code:
            - "0001": KOSPI 종합
            - "1001": KOSDAQ 종합
            - "2001": KOSPI200

        반환 dict 주요 키:
            - bstp_nmix_prpr      : 현재 지수
            - bstp_nmix_prdy_clpr : 전일 종가
            - bstp_nmix_prdy_vrss : 전일 대비
            - bstp_nmix_prdy_ctrt : 전일 대비 등락률(%)

        FID_COND_MRKT_DIV_CODE = 'U' (업종/지수). 종목조회의 'J' 와 다름.
        """
        r = self._get(
            "/uapi/domestic-stock/v1/quotations/inquire-index-price",
            "FHPUP02100000",
            {"FID_COND_MRKT_DIV_CODE": "U", "FID_INPUT_ISCD": index_code},
        )
        if r.status_code != 200:
            return None
        j = r.json()
        return j.get("output") if j.get("rt_cd") == "0" else None

    def get_minute_chart(
        self, code: str, period: str = "1", limit: int = 100
    ) -> pd.DataFrame:
        """
        분봉 [v1_국내주식-017] — 갭 보정용.

        ⚠️ 한투 FHKST03010200 스펙:
          * 1분봉만 지원 (period 인자는 상위 호출자 호환용으로만 유지)
          * 1회 호출 ≤ 30봉, 응답은 최신→과거 역순
          * FID_INPUT_HOUR_1 = HHMMSS 형식의 "조회 커서 시각"
          * 커서를 뒤로 밀며 페이지네이션해 limit 개수만큼 수집
        (holding_bot.fetch_and_store_min_candles 과 동일 패턴 — 검증된 파라미터)
        """
        path = "/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"
        tr_id = "FHKST03010200"
        try:
            # 페이지네이션 커서 — 장중이면 현재 시각, 장 마감 후엔 15:30:00
            now = dt.now()
            if now.hour < 9:
                cursor_dt = now.replace(hour=15, minute=30, second=0, microsecond=0) \
                             - datetime.timedelta(days=1)
            elif now.hour > 15 or (now.hour == 15 and now.minute >= 30):
                cursor_dt = now.replace(hour=15, minute=30, second=0, microsecond=0)
            else:
                cursor_dt = now

            rows: list = []
            seen: set = set()  # 중복 제거용 (time 키)
            # 최대 페이지 수 — 한 페이지 ≈ 30봉 기준 여유 있게 계산
            max_pages = max(1, (int(limit) // 25) + 2)
            for _ in range(max_pages):
                params = {
                    "FID_ETC_CLS_CODE": "",
                    "FID_COND_MRKT_DIV_CODE": "J",
                    "FID_INPUT_ISCD": code,
                    # HHMMSS 커서 (역순으로 이동)
                    "FID_INPUT_HOUR_1": cursor_dt.strftime("%H%M%S"),
                    "FID_PW_DATA_INCU_YN": "Y",  # 과거 날짜 포함
                }
                r = self._get(path, tr_id, params)
                if r.status_code != 200:
                    break
                j = r.json()
                if j.get("rt_cd") != "0":
                    break
                out = j.get("output2", [])
                if not out:
                    break

                page_last_dt = None
                for it in out:
                    try:
                        d = str(it.get("stck_bsop_date", "") or "")
                        t = str(it.get("stck_cntg_hour", "") or "000000")
                        if len(d) < 8 or len(t) < 6:
                            continue
                        # YYYYMMDDHHMM (12자리) — CandleAggregator.fill_gap_from_rest 호환
                        tkey = d + t[:4]
                        if tkey in seen:
                            continue
                        seen.add(tkey)
                        rows.append({
                            "time": tkey,
                            "open": abs(float(it.get("stck_oprc", 0))),
                            "high": abs(float(it.get("stck_hgpr", 0))),
                            "low": abs(float(it.get("stck_lwpr", 0))),
                            "close": abs(float(it.get("stck_clpr", 0))),
                            "volume": int(it.get("acml_vol", 0) or 0),
                        })
                        # 절대 시각 파싱 → 다음 커서 계산용
                        try:
                            page_last_dt = dt.strptime(d + t, "%Y%m%d%H%M%S")
                        except Exception:
                            pass
                    except Exception:
                        continue

                if len(rows) >= int(limit) or page_last_dt is None:
                    break
                # 다음 페이지: 마지막(가장 오래된) 레코드 시각 - 1분
                cursor_dt = page_last_dt - datetime.timedelta(minutes=1)
                # 장외시간으로 넘어가면 중단 (더 과거는 한투 FHKST03010200이 안 줌)
                if cursor_dt.hour < 9 or (cursor_dt.hour == 15 and cursor_dt.minute >= 30) \
                        or cursor_dt.hour > 15:
                    break

            if not rows:
                return pd.DataFrame()
            df = pd.DataFrame(rows).sort_values("time").reset_index(drop=True)
            return df.tail(int(limit))
        except Exception as e:
            logger.debug("분봉 조회 실패(%s): %s", code, e)
            return pd.DataFrame()

    def get_daily_chart(self, code: str, limit: int = 10) -> pd.DataFrame:
        """일봉 [v1_국내주식-017] — 거래대금(대/중/소형) 판정용."""
        path = "/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice"
        tr_id = "FHKST03010100"
        try:
            end_dt = dt.now()
            start_dt = end_dt - datetime.timedelta(days=limit + 30)
            r = self._get(
                path, tr_id,
                {
                    "FID_COND_MRKT_DIV_CODE": "J",
                    "FID_INPUT_ISCD": code,
                    "FID_INPUT_DATE_1": start_dt.strftime("%Y%m%d"),
                    "FID_INPUT_DATE_2": end_dt.strftime("%Y%m%d"),
                    "FID_PERIOD_DIV_CODE": "D",
                    "FID_ORG_ADJ_PRC": "1",
                },
            )
            if r.status_code != 200:
                return pd.DataFrame()
            j = r.json()
            if j.get("rt_cd") != "0":
                return pd.DataFrame()
            out = j.get("output2", [])
            if not out:
                return pd.DataFrame()
            rows = []
            for it in out[:limit]:
                try:
                    rows.append({
                        "date": str(it.get("stck_bsop_date", "") or ""),
                        "open": abs(float(it.get("stck_oprc", 0))),
                        "high": abs(float(it.get("stck_hgpr", 0))),
                        "low": abs(float(it.get("stck_lwpr", 0))),
                        "close": abs(float(it.get("stck_clpr", 0))),
                        "volume": int(it.get("acml_vol", 0)),
                    })
                except Exception:
                    continue
            if not rows:
                return pd.DataFrame()
            return pd.DataFrame(rows).sort_values("date").reset_index(drop=True)
        except Exception as e:
            logger.debug("일봉 조회 실패(%s): %s", code, e)
            return pd.DataFrame()

    # ------------------------------------------------------------------
    # 랭킹 (거래량/거래대금/체결강도/등락률 상위 — FHPST01710000)
    # ------------------------------------------------------------------
    # volume-rank API 는 tr_cont 를 받지 않는다. 1회 호출 ≤ 30~50건 반환.
    #   FID_BLNG_CLS_CODE:
    #     0=평균거래량, 1=거래증가율, 2=평균거래회전율,
    #     3=거래금액순, 4=등락률(상승), 5=등락률(하락), 6=체결강도순
    @staticmethod
    def _is_valid_stock_for_rank(name: str, code: str) -> bool:
        """스팩/ETN/레버리지/인버스/우선주 등 비본주 제외 (기존 kiwoom 스캐너 규칙과 동일)."""
        if not code or not name:
            return False
        code = code.strip()
        name = name.strip()
        if len(code) != 6:
            return False
        nm = name.upper()
        # ETN / ETF / 레버리지 / 인버스
        bad_kw = ("ETN", "ETF", "레버리지", "인버스", "2X", "3X", "선물", "KODEX", "TIGER",
                 "KBSTAR", "ARIRANG", "HANARO", "SOL ", "KOSEF", "ACE ", "KINDEX",
                 "파워", "히어로", "SMART", "TREX", "WON", "KOACT", "TIMEFOLIO")
        if any(k in nm for k in bad_kw):
            return False
        # 스팩 (기업인수목적회사)
        if "스팩" in name or "SPAC" in nm:
            return False
        # 우선주 (뒤에 우/우B 등)
        if name.endswith("우") or name.endswith("우B") or name.endswith("(전환)"):
            return False
        return True

    def _filter_rank_rows(self, rows: list) -> list:
        if not rows:
            return []
        out = []
        for it in rows:
            code = (
                it.get("mksc_shrn_iscd") or it.get("stk_cd")
                or it.get("code") or ""
            ).strip()
            name = (
                it.get("hts_kor_isnm") or it.get("stk_nm")
                or it.get("prst_name") or ""
            ).strip()
            if self._is_valid_stock_for_rank(name, code):
                out.append(it)
        return out

    def _fetch_volume_rank(
        self,
        *,
        market: str = "J",
        blng_cls_code: str = "0",
        limit: int = 100,
        exclude_non_stock: bool = True,
    ) -> List[dict]:
        """FHPST01710000 1회 호출 → rows."""
        path = "/uapi/domestic-stock/v1/quotations/volume-rank"
        tr_id = "FHPST01710000"
        params = {
            "FID_COND_MRKT_DIV_CODE": market,
            "FID_COND_SCR_DIV_CODE": "20171",
            "FID_INPUT_ISCD": "0000",
            "FID_DIV_CLS_CODE": "0",
            "FID_BLNG_CLS_CODE": blng_cls_code,
            "FID_TRGT_CLS_CODE": "111111111",
            "FID_TRGT_EXLS_CLS_CODE": "0000000000",
            "FID_INPUT_PRICE_1": "0",
            "FID_INPUT_PRICE_2": "0",
            "FID_VOL_CNT": "0",
            "FID_INPUT_DATE_1": "",
        }
        try:
            r = self._get(path, tr_id, params)
            if r.status_code != 200:
                logger.debug("랭킹 HTTP %s (blng=%s)", r.status_code, blng_cls_code)
                return []
            j = r.json()
            if j.get("rt_cd") != "0":
                logger.debug(
                    "랭킹 실패 rt_cd=%s msg=%s",
                    j.get("rt_cd"), j.get("msg1"),
                )
                return []
            rows = j.get("output") or []
            if isinstance(rows, dict):
                rows = [rows]
            if exclude_non_stock:
                rows = self._filter_rank_rows(rows)
            return rows[:limit]
        except Exception as e:
            logger.debug("랭킹 조회 예외 (blng=%s): %s", blng_cls_code, e)
            return []

    def get_volume_rank(
        self, *, market: str = "J", limit: int = 100, exclude_non_stock: bool = True,
    ) -> List[dict]:
        """거래량 상위."""
        return self._fetch_volume_rank(
            market=market, blng_cls_code="0",
            limit=limit, exclude_non_stock=exclude_non_stock,
        )

    def get_trading_value_rank(
        self, *, market: str = "J", limit: int = 100, exclude_non_stock: bool = True,
    ) -> List[dict]:
        """거래대금 상위."""
        return self._fetch_volume_rank(
            market=market, blng_cls_code="3",
            limit=limit, exclude_non_stock=exclude_non_stock,
        )

    def get_execution_strength_rank(
        self, *, market: str = "J", limit: int = 100, exclude_non_stock: bool = True,
    ) -> List[dict]:
        """체결강도 상위 (매수세 강한 종목)."""
        return self._fetch_volume_rank(
            market=market, blng_cls_code="6",
            limit=limit, exclude_non_stock=exclude_non_stock,
        )

    def get_price_change_rank(
        self,
        *,
        market: str = "J",
        sort_type: str = "up",
        limit: int = 100,
        exclude_non_stock: bool = True,
    ) -> List[dict]:
        """등락률 상위. sort_type='up' 상승, 'down' 하락(낙폭)."""
        blng = "5" if str(sort_type).lower() in ("down", "decline", "2") else "4"
        rows = self._fetch_volume_rank(
            market=market, blng_cls_code=blng,
            limit=limit, exclude_non_stock=exclude_non_stock,
        )
        if rows or blng == "5":
            return rows
        # 일부 계정에서 4/5 미지원 → 거래량 fallback (상승만)
        return self.get_volume_rank(
            market=market, limit=limit, exclude_non_stock=exclude_non_stock,
        )

    def get_execution_strength_map(
        self, *, market: str = "J", limit: int = 200,
    ) -> Dict[str, float]:
        """
        체결강도 상위 조회 → {code: cntr_str(체결강도값)} 맵.

        ⚠️ 검증된 제약 (실호출 확인, 2025-04 기준):
          * KIS volume-rank(FHPST01710000) 응답에 **체결강도 값 필드가 없음**.
            응답은 거래량/거래대금/회전율만 포함. 따라서 본 맵은 0.0 으로 채워짐.
          * 모의투자 서버에서는 ``blng=6`` 정렬이 체결강도 순이 아닌 종목코드 순
            으로 반환됨(실계좌는 정렬이 정상일 가능성 높음).
          * ✅ **정확한 실시간 체결강도**는 WebSocket ``H0STCNT0`` 의 ``cttr``
            필드를 사용할 것. ``WSManager.price_cache`` 에 이미 들어 있음.

        실사용 가이드:
          - 유니버스 시드만 필요하면 ``get_execution_strength_rank()`` 를 직접 호출
            (정렬 순서만 쓰고 값 파싱은 생략).
          - 체결강도 값 기반 필터(≥ 120 등)는 전략 코드에서 WS tick 으로 해결.
        """
        rows = self.get_execution_strength_rank(market=market, limit=limit)
        out: Dict[str, float] = {}
        for it in rows or []:
            code = (
                it.get("mksc_shrn_iscd") or it.get("stk_cd")
                or it.get("code") or ""
            ).strip()
            if not code or len(code) != 6:
                continue
            raw = (
                it.get("cntr_str") or it.get("exec_str")
                or it.get("strg_rt") or ""
            )
            try:
                out[code] = float(str(raw).replace(",", "").strip()) if raw else 0.0
            except (ValueError, TypeError):
                out[code] = 0.0
        return out

    # ------------------------------------------------------------------
    # 계좌/잔고
    # ------------------------------------------------------------------
    def get_account_balance(self) -> Optional[dict]:
        """계좌 잔고 [국내주식-006] (inquire-balance).

        ※ 연속조회(페이징): 한 번의 호출은 실전 50건 / 모의 20건까지만 내려온다.
           보유 종목이 한도를 넘으면 응답 헤더 tr_cont 가 'M'(또는 'F') 로 오며,
           이때 직전 응답의 ctx_area_fk100/nk100 을 다음 요청에 실어 반복 조회한다.
           output1(보유 종목 배열)을 누적 병합해 반환하므로 51번째 이후 종목도 보존된다.
           (이 맵이 모든 매도의 보유수량 검증 기준 → 누락 시 매도 차단 위험이라 페이징 필수)
        """
        tr_id = "VTTC8434R" if self.mock else "TTTC8434R"
        path = "/uapi/domestic-stock/v1/trading/inquire-balance"
        # 무한루프 방지 — 1페이지=실전50/모의20종목, 기본 20p(최대 약 1000종목)면 충분
        max_pages = get_env_int("BALANCE_MAX_PAGES", 20)
        base_params = {
            "CANO": self.account_no, "ACNT_PRDT_CD": self.account_code,
            "AFHR_FLPR_YN": "N", "OFL_YN": "N", "INQR_DVSN": "01",
            "UNPR_DVSN": "01", "FUND_STTL_ICLD_YN": "N",
            "FNCG_AMT_AUTO_RDPT_YN": "N", "PRCS_DVSN": "00",
            "CTX_AREA_FK100": "", "CTX_AREA_NK100": "",
        }
        merged: Optional[dict] = None     # 첫 페이지 원본(output2 요약 등 보존)
        all_output1: List[dict] = []      # 보유 종목 누적 병합
        fk100, nk100, tr_cont = "", "", ""
        try:
            for page in range(1, max_pages + 1):
                params = dict(base_params)
                params["CTX_AREA_FK100"] = fk100
                params["CTX_AREA_NK100"] = nk100
                r = self._get(path, tr_id, params, tr_cont=tr_cont)
                if r.status_code != 200:
                    if page == 1:
                        return None  # 첫 페이지 실패 → None (유령잔고 오판 방지)
                    logger.warning(
                        "잔고 연속조회 %d페이지 HTTP %s → 지금까지 %d종목 부분 반환",
                        page, r.status_code, len(all_output1),
                    )
                    break
                j = r.json()
                if j.get("rt_cd") != "0":
                    if page == 1:
                        return None
                    logger.warning(
                        "잔고 연속조회 %d페이지 rt_cd=%s msg=%s → 부분 반환",
                        page, j.get("rt_cd"), j.get("msg1"),
                    )
                    break
                if merged is None:
                    merged = j
                page_out1 = j.get("output1") or []
                if isinstance(page_out1, dict):
                    page_out1 = [page_out1]
                all_output1.extend(page_out1)
                # 다음 페이지 판단: 응답 헤더 tr_cont (F/M = 다음 있음, D/E/공백 = 끝)
                next_cont = ""
                try:
                    next_cont = str(r.headers.get("tr_cont") or "").strip()
                except Exception:
                    next_cont = ""
                fk100 = str(j.get("ctx_area_fk100") or "").strip()
                nk100 = str(j.get("ctx_area_nk100") or "").strip()
                if next_cont in ("F", "M") and (fk100 or nk100):
                    tr_cont = "N"  # 다음 페이지 요청
                    continue
                break  # 마지막 페이지
            else:
                # for-else: max_pages 소진했는데도 루프가 break 안됨 = 더 남았을 수 있음
                logger.warning(
                    "잔고 연속조회 최대 %d페이지 도달 → %d종목까지만 반환(추가 종목 누락 가능)",
                    max_pages, len(all_output1),
                )
            if merged is None:
                return None
            merged["output1"] = all_output1  # 누적 병합본으로 교체 (단일 페이지면 동일)
            return merged
        except Exception as e:
            logger.error("계좌 잔고 조회 실패: %s", e)
            return None

    def get_broker_holdings_map(self) -> Optional[Dict[str, Dict]]:
        """
        잔고 API output1 → {code: {qty, avg_price, name, evlu_amt}} 맵 변환.
        ** 모든 매도 주문 직전 이 맵으로 실제 보유 수량을 검증해야 한다. **
        API 실패 시 None (빈 dict 와 구분 — 유령잔고 오판 방지).
        """
        balance = self.get_account_balance()
        if not balance:
            return None
        output1 = balance.get("output1") or []
        if isinstance(output1, dict):
            output1 = [output1]
        result: Dict[str, Dict] = {}
        for it in output1:
            code = (it.get("pdno") or it.get("PDNO") or "").strip()
            if not code:
                continue
            qty_raw = (
                it.get("hldg_qty") or it.get("HLDG_QTY")
                or it.get("ord_psbl_qty") or it.get("ORD_PSBL_QTY") or 0
            )
            try:
                qty = int(float(str(qty_raw).replace(",", "")))
            except Exception:
                qty = 0
            if qty <= 0:
                continue
            avg_raw = it.get("pchs_avg_pric") or it.get("PCHS_AVG_PRIC") or 0
            try:
                avg_price = abs(float(str(avg_raw).replace(",", "")))
            except Exception:
                avg_price = 0.0
            name = (it.get("prdt_name") or it.get("PRDT_NAME") or code).strip()
            evlu_raw = it.get("evlu_amt") or it.get("EVLU_AMT") or 0
            try:
                evlu_amt = float(str(evlu_raw).replace(",", ""))
            except Exception:
                evlu_amt = 0.0
            prpr_raw = it.get("prpr") or it.get("PRPR") or 0
            try:
                current_price = abs(float(str(prpr_raw).replace(",", "")))
            except Exception:
                current_price = 0.0
            result[code] = {
                "qty": qty,
                "avg_price": avg_price,
                "name": name,
                "evlu_amt": evlu_amt,
                "current_price": current_price,
            }
        return result

    # ------------------------------------------------------------------
    # 주문
    # ------------------------------------------------------------------
    def _order(
        self,
        *,
        code: str,
        qty: int,
        price: int,
        order_type: str,
        side: str,
    ) -> Optional[str]:
        """
        주문 공통 호출. 성공 시 ODNO(str) 반환, 실패 시 None.
        side: 'BUY' | 'SELL'
        """
        side = side.upper()
        if side == "BUY":
            tr_id = "VTTC0802U" if self.mock else "TTTC0802U"
        else:
            tr_id = "VTTC0801U" if self.mock else "TTTC0801U"
        path = "/uapi/domestic-stock/v1/trading/order-cash"
        body = {
            "CANO": self.account_no, "ACNT_PRDT_CD": self.account_code,
            "PDNO": code, "ORD_DVSN": order_type,
            "ORD_QTY": str(qty), "ORD_UNPR": str(price),
        }
        try:
            r = self._post(path, tr_id, body)
            if r.status_code != 200:
                logger.error("주문 HTTP 에러 side=%s code=%s status=%s",
                             side, code, r.status_code)
                return None
            j = r.json()
            if j.get("rt_cd") == "0":
                if side == "BUY":
                    self._last_order_msg_cd = None
                    self._last_order_msg1 = None
                else:
                    self._last_sell_msg_cd = None
                    self._last_sell_msg1 = None
                ord_no = str((j.get("output") or {}).get("ODNO", "") or "").strip()
                return ord_no or None
            # 실패: 원인 저장 (매매불가/영업일 아님 등 분기용)
            if side == "BUY":
                self._last_order_msg_cd = j.get("msg_cd", "")
                self._last_order_msg1 = str(j.get("msg1", "") or "")
                logger.error(
                    "[매수주문실패] code=%s rt_cd=%s msg_cd=%s msg1=%s",
                    code, j.get("rt_cd"),
                    self._last_order_msg_cd, self._last_order_msg1,
                )
            else:
                self._last_sell_msg_cd = j.get("msg_cd", "")
                self._last_sell_msg1 = str(j.get("msg1", "") or "")
                logger.error(
                    "[매도주문실패] code=%s rt_cd=%s msg_cd=%s msg1=%s",
                    code, j.get("rt_cd"),
                    self._last_sell_msg_cd, self._last_sell_msg1,
                )
            return None
        except Exception as e:
            logger.error("주문 예외 side=%s code=%s: %s", side, code, e)
            return None

    def buy_market_order(self, code: str, qty: int) -> Optional[str]:
        """시장가 매수.  실전은 USE_MARKET_IOC 설정 시 IOC(13), 아니면 일반 시장가(01)."""
        if self.mock:
            order_type = "01"
        else:
            order_type = "13" if get_env_bool("USE_MARKET_IOC", True) else "01"
        return self._order(
            code=code, qty=qty, price=0, order_type=order_type, side="BUY"
        )

    def buy_limit_order(self, code: str, qty: int, price: int) -> Optional[str]:
        """지정가 매수 (ORD_DVSN=00). 돌파/꼬리잡기 전략에서 사용."""
        return self._order(
            code=code, qty=qty, price=int(price), order_type="00", side="BUY"
        )

    def cancel_order(
        self,
        org_odno: str,
        *,
        org_branch: str = "",
        qty: int = 0,
        order_dvsn: str = "00",
    ) -> bool:
        """미체결 주문 전량 취소 (RVSE_CNCL_DVSN_CD=02)."""
        odno = str(org_odno or "").strip()
        if not odno:
            return False
        tr_id = "VTTC0803U" if self.mock else "TTTC0803U"
        path = "/uapi/domestic-stock/v1/trading/order-rvsecncl"
        body = {
            "CANO": self.account_no,
            "ACNT_PRDT_CD": self.account_code,
            "KRX_FWDG_ORD_ORGNO": str(org_branch or ""),
            "ORGN_ODNO": odno,
            "ORD_DVSN": order_dvsn,
            "RVSE_CNCL_DVSN_CD": "02",
            "ORD_QTY": str(int(qty)),
            "ORD_UNPR": "0",
            "QTY_ALL_ORD_YN": "Y",
        }
        try:
            r = self._post(path, tr_id, body)
            if r.status_code != 200:
                logger.error("주문취소 HTTP 에러 odno=%s status=%s", odno, r.status_code)
                return False
            j = r.json()
            if j.get("rt_cd") == "0":
                return True
            logger.warning(
                "주문취소 실패 odno=%s rt_cd=%s msg=%s",
                odno, j.get("rt_cd"), j.get("msg1"),
            )
            return False
        except Exception as e:
            logger.error("주문취소 예외 odno=%s: %s", odno, e)
            return False

    def sell_limit_order(self, code: str, qty: int, price: int) -> Optional[str]:
        """지정가 매도 (ORD_DVSN=00). 익절 시 매수 1호가 등."""
        return self._order(
            code=code, qty=qty, price=int(price), order_type="00", side="SELL"
        )

    def sell_market_order(self, code: str, qty: int) -> Optional[str]:
        """시장가 매도.  IOC 에러가 너무 빈번한 경우를 위해 기본 '01' 일반 시장가."""
        return self._order(
            code=code, qty=qty, price=0, order_type="01", side="SELL"
        )

    # ------------------------------------------------------------------
    # 조건검색 (REST — 웹소켓은 공식 지원 안 함. 30초 폴링이 정석)
    # ------------------------------------------------------------------
    def get_condition_list(self, user_id: str) -> List[Dict]:
        """
        HTS/MTS 에 서버 저장된 조건식 목록 조회.
        [국내주식] 시세분석 - 종목조건검색 목록조회 (psearch-title)

        Returns: [{"seq":"0","condition_name":"우상향돌파",...}, ...]
        """
        try:
            r = self._get(
                "/uapi/domestic-stock/v1/quotations/psearch-title",
                "HHKST03900300",
                {"user_id": user_id},
            )
            if r.status_code != 200:
                return []
            j = r.json()
            if j.get("rt_cd") != "0":
                return []
            out = j.get("output2") or []
            if isinstance(out, dict):
                out = [out]
            return out
        except Exception as e:
            logger.debug("조건식 목록 조회 실패: %s", e)
            return []

    def get_condition_result(self, user_id: str, seq: str) -> List[Dict]:
        """
        조건식에 걸려 있는 종목 목록 조회 (psearch-result).
        Returns: [{"code":"005930","name":"삼성전자",...}, ...]
        """
        try:
            r = self._get(
                "/uapi/domestic-stock/v1/quotations/psearch-result",
                "HHKST03900400",
                {"user_id": user_id, "seq": str(seq)},
            )
            if r.status_code != 200:
                return []
            j = r.json()
            if j.get("rt_cd") != "0":
                return []
            out = j.get("output2") or []
            if isinstance(out, dict):
                out = [out]
            parsed: List[Dict] = []
            for it in out:
                code = (
                    it.get("code") or it.get("stck_shrn_iscd")
                    or it.get("mksc_shrn_iscd") or ""
                ).strip()
                if not code:
                    continue
                name = (
                    it.get("name") or it.get("hts_kor_isnm")
                    or it.get("stck_prpr") or code
                )
                parsed.append({"code": code, "name": str(name).strip() or code})
            return parsed
        except Exception as e:
            logger.debug("조건검색 결과 조회 실패 (seq=%s): %s", seq, e)
            return []

    def get_order_history_today(self, odno: str = "") -> Optional[dict]:
        """당일 주문 내역 조회 [국내주식-005 inquire-daily-ccld].

        ``odno`` 지정 시 해당 주문번호만 조회 → 응답 경량화로 장중 혼잡 시
        모의서버 HTTP 500 회피율을 높인다. (모의가 ODNO 필터를 무시하면
        기존처럼 전체를 받아 호출부에서 필터하므로 동작은 동일)
        """
        tr_id = "VTTC8001R" if self.mock else "TTTC8001R"
        today = dt.now().strftime("%Y%m%d")
        try:
            r = self._get(
                "/uapi/domestic-stock/v1/trading/inquire-daily-ccld",
                tr_id,
                {
                    "CANO": self.account_no, "ACNT_PRDT_CD": self.account_code,
                    "INQR_STRT_DT": today, "INQR_END_DT": today,
                    "SLL_BUY_DVSN_CD": "00", "INQR_DVSN": "00", "PDNO": "",
                    "CCLD_DVSN": "00", "ORD_GNO_BRNO": "", "ODNO": str(odno or ""),
                    "INQR_DVSN_3": "00", "INQR_DVSN_1": "",
                    "CTX_AREA_FK100": "", "CTX_AREA_NK100": "",
                },
            )
            if r.status_code != 200:
                return None
            j = r.json()
            return j if j.get("rt_cd") == "0" else None
        except Exception as e:
            logger.debug("주문 내역 조회 실패: %s", e)
            return None

    @staticmethod
    def _execution_map_from_history(j: Optional[dict]) -> Dict[str, Dict]:
        """inquire-daily-ccld 응답 → {odno: {filled_qty, avg_price, code}}."""
        out: Dict[str, Dict] = {}
        if not j:
            return out
        out1 = j.get("output1") or []
        if isinstance(out1, dict):
            out1 = [out1]
        for row in out1:
            odno = str(row.get("odno") or row.get("ODNO") or "").strip()
            if not odno:
                continue
            filled = row.get("tot_ccld_qty") or row.get("TOT_CCLD_QTY") or 0
            avg = (
                row.get("avg_prvs") or row.get("AVG_PRVS")
                or row.get("ord_unpr") or row.get("ORD_UNPR") or 0
            )
            try:
                q = int(float(str(filled).replace(",", "")))
                p = float(str(avg).replace(",", ""))
            except Exception:
                continue
            if q <= 0 or p <= 0:
                continue
            pdno = str(row.get("pdno") or row.get("PDNO") or "").strip()
            out[odno] = {"filled_qty": q, "avg_price": p, "code": pdno}
        return out

    def get_today_execution_map(self, wait_sec: float = 0.0) -> Optional[Dict[str, Dict]]:
        """당일 체결 일괄 조회 — poll_pending_fills N건→1 REST 용.

        반환 None: API 실패 시 호출부가 건별 get_execution_by_odno 로 폴백.
        """
        import time as _t
        _t.sleep(max(0.0, wait_sec))
        j = self.get_order_history_today(odno="")
        if j is None:
            return None
        return self._execution_map_from_history(j)

    def get_execution_by_odno(
        self, ord_no: str, code: Optional[str] = None, wait_sec: float = 2.0
    ) -> Optional[Dict]:
        """
        ODNO 로 당일 체결 조회.
        반환: {"filled_qty": int, "avg_price": float} 또는 None
        """
        import time as _t
        if not ord_no:
            return None
        _t.sleep(max(0.0, wait_sec))
        # ODNO 단건 조회 (응답 경량 → 모의서버 500 회피). 필터 무시되면 전체를 받아 아래에서 필터.
        j = self.get_order_history_today(odno=str(ord_no).strip())
        fill_map = self._execution_map_from_history(j)
        hit = fill_map.get(str(ord_no).strip())
        if not hit:
            return None
        if code and hit.get("code") and hit["code"] != str(code).strip():
            return None
        return {"filled_qty": hit["filled_qty"], "avg_price": hit["avg_price"]}
