"""KIS Open API 기반 한국 시장 데이터 fetcher.

태버지 노트북의 yfinance/FinanceDataReader/investing.com 비공식 API 를
KIS 공식 API 로 대체하기 위한 헬퍼 모듈.

8 가지 핵심 fetch 함수 + 24h TTL 캐시 + 1.5 초 rate limit.

사용:
    from kis_data import (
        get_ohlcv_kis,
        get_neighbour_high_low_top,
        get_volume_rank,
        get_sector_index_daily,
        get_etf_nav,
        get_vi_status,
        get_market_holidays,
        get_ohlcv_wide,  # yc_close_wide / yc_ohlcv_wide 호환
    )

    df = get_ohlcv_kis("005930", start="2024-01-01", end="2024-12-31")
"""
from __future__ import annotations

import json
import logging
import os
import sys
import time
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from pathlib import Path
from threading import Lock
from typing import Iterable

import pandas as pd
import requests

# ────────────────────────────────────────────────────────────────────
# Reuse KISOrderClient for token/session/credential plumbing
# ────────────────────────────────────────────────────────────────────
_AI_TRADING_SRC = "/Users/loonshots/projects/core/ai_trading_project/src"
if _AI_TRADING_SRC not in sys.path:
    sys.path.insert(0, _AI_TRADING_SRC)

from ai_trading.broker.kis_order import KISOrderClient  # type: ignore  # noqa: E402

logger = logging.getLogger(__name__)

# ────────────────────────────────────────────────────────────────────
# Constants
# ────────────────────────────────────────────────────────────────────
LIVE_BASE_URL = "https://openapi.koreainvestment.com:9443"

CACHE_DIR = Path(__file__).resolve().parent / ".kis_cache"
CACHE_DIR.mkdir(exist_ok=True)
CACHE_TTL_SEC = 24 * 60 * 60  # 24h

RATE_LIMIT_SEC = 1.5  # 호출 사이 최소 간격 (KIS 권장: 초당 20건, 보수적으로 1.5s)
_last_call_time = 0.0
_rate_lock = Lock()


# ────────────────────────────────────────────────────────────────────
# Singleton client (token 공유)
# ────────────────────────────────────────────────────────────────────
_client_singleton: KISOrderClient | None = None


def _client() -> KISOrderClient:
    global _client_singleton
    if _client_singleton is None:
        _client_singleton = KISOrderClient()
    return _client_singleton


def _rate_limit() -> None:
    global _last_call_time
    with _rate_lock:
        now = time.time()
        wait = RATE_LIMIT_SEC - (now - _last_call_time)
        if wait > 0:
            time.sleep(wait)
        _last_call_time = time.time()


def _normalize_ticker(ticker: str) -> str:
    """삼성전자.KS / 005930.KS / 005930 → "005930" (KIS 6자리 단축코드)."""
    if not ticker:
        return ticker
    t = str(ticker).strip().upper()
    # Strip Yahoo suffixes
    for suffix in (".KS", ".KQ", ".KO"):
        if t.endswith(suffix):
            t = t[: -len(suffix)]
            break
    return t.zfill(6) if t.isdigit() else t


# ────────────────────────────────────────────────────────────────────
# Cache helpers (24h TTL, JSON for small / parquet for DataFrames)
# ────────────────────────────────────────────────────────────────────
def _cache_path(key: str, ext: str = "parquet") -> Path:
    safe = key.replace("/", "_").replace(":", "_")
    return CACHE_DIR / f"{safe}.{ext}"


def _cache_load_df(key: str) -> pd.DataFrame | None:
    p = _cache_path(key, "parquet")
    if not p.exists():
        return None
    age = time.time() - p.stat().st_mtime
    if age > CACHE_TTL_SEC:
        return None
    try:
        return pd.read_parquet(p)
    except Exception as exc:  # noqa: BLE001
        logger.debug("cache read fail %s: %s", key, exc)
        return None


def _cache_save_df(key: str, df: pd.DataFrame) -> None:
    p = _cache_path(key, "parquet")
    try:
        df.to_parquet(p)
    except Exception as exc:  # noqa: BLE001
        logger.debug("cache save fail %s: %s", key, exc)


def _cache_load_json(key: str):
    p = _cache_path(key, "json")
    if not p.exists():
        return None
    age = time.time() - p.stat().st_mtime
    if age > CACHE_TTL_SEC:
        return None
    try:
        return json.loads(p.read_text())
    except Exception:  # noqa: BLE001
        return None


def _cache_save_json(key: str, data) -> None:
    p = _cache_path(key, "json")
    try:
        p.write_text(json.dumps(data, ensure_ascii=False, default=str))
    except Exception as exc:  # noqa: BLE001
        logger.debug("cache save fail %s: %s", key, exc)


# ────────────────────────────────────────────────────────────────────
# Generic GET wrapper
# ────────────────────────────────────────────────────────────────────
def _kis_get(path: str, tr_id: str, params: dict, timeout: int = 15) -> dict:
    """KIS REST GET. Token + headers from KISOrderClient. rate-limited."""
    cli = _client()
    url = f"{cli.base_url}{path}"
    _rate_limit()
    resp = cli._session.get(  # noqa: SLF001
        url, headers=cli._headers(tr_id), params=params, timeout=timeout  # noqa: SLF001
    )
    if resp.status_code >= 400:
        logger.error("KIS GET %s -> %s %s", path, resp.status_code, resp.text[:200])
        resp.raise_for_status()
    data = resp.json()
    if data.get("rt_cd") not in (None, "0"):
        # Some endpoints (e.g. holidays) don't return rt_cd; tolerate.
        msg = data.get("msg1") or data.get("msg") or "?"
        raise RuntimeError(f"KIS API error {tr_id}: {msg}")
    return data


# ────────────────────────────────────────────────────────────────────
# 1) 일별 OHLCV — 국내주식 기간별시세 (FHKST03010100)
#    /uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice
#    period_code: D=일, W=주, M=월, Y=년
#    한 번 호출당 최대 100영업일.
# ────────────────────────────────────────────────────────────────────
TR_DAILY_CHART = "FHKST03010100"
PATH_DAILY_CHART = "/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice"


def _fetch_daily_chart_chunk(
    ticker: str,
    start: str,
    end: str,
    period_code: str = "D",
    adjusted: bool = False,
) -> pd.DataFrame:
    """단일 chunk (≤100 영업일) fetch. start/end 는 'YYYYMMDD'."""
    params = {
        "FID_COND_MRKT_DIV_CODE": "J",  # KRX 정규
        "FID_INPUT_ISCD": ticker,
        "FID_INPUT_DATE_1": start,
        "FID_INPUT_DATE_2": end,
        "FID_PERIOD_DIV_CODE": period_code,
        "FID_ORG_ADJ_PRC": "0" if adjusted else "1",  # 0=수정주가, 1=원주가
    }
    data = _kis_get(PATH_DAILY_CHART, TR_DAILY_CHART, params)
    rows = data.get("output2") or []
    records = []
    for row in rows:
        d = row.get("stck_bsop_date")
        if not d:
            continue
        try:
            records.append(
                {
                    "Date": pd.to_datetime(d, format="%Y%m%d"),
                    "Open": float(row.get("stck_oprc") or 0),
                    "High": float(row.get("stck_hgpr") or 0),
                    "Low": float(row.get("stck_lwpr") or 0),
                    "Close": float(row.get("stck_clpr") or 0),
                    "Volume": float(row.get("acml_vol") or 0),
                }
            )
        except Exception as exc:  # noqa: BLE001
            logger.debug("row parse fail %s: %s", d, exc)
    if not records:
        return pd.DataFrame(columns=["Open", "High", "Low", "Close", "Volume"])
    df = pd.DataFrame.from_records(records).set_index("Date").sort_index()
    return df


def get_ohlcv_kis(
    ticker: str,
    start: str | date | datetime,
    end: str | date | datetime | None = None,
    period_code: str = "D",
    adjusted: bool = False,
    use_cache: bool = True,
) -> pd.DataFrame:
    """일별 OHLCV. start/end inclusive. KIS 100일 chunk 자동 paging.

    Returns: DataFrame[index=Date(tz-naive), cols=Open,High,Low,Close,Volume]
    """
    code = _normalize_ticker(ticker)
    end_dt = pd.Timestamp(end) if end is not None else pd.Timestamp.today().normalize()
    start_dt = pd.Timestamp(start)
    cache_key = (
        f"ohlcv_{code}_{period_code}_{int(adjusted)}_"
        f"{start_dt.strftime('%Y%m%d')}_{end_dt.strftime('%Y%m%d')}"
    )
    if use_cache:
        cached = _cache_load_df(cache_key)
        if cached is not None:
            return cached

    chunks: list[pd.DataFrame] = []
    cursor_end = end_dt
    # KIS chart returns ≤100 most-recent rows up to FID_INPUT_DATE_2,
    # so we walk backwards in 100-day windows.
    while cursor_end >= start_dt:
        cursor_start = max(start_dt, cursor_end - pd.Timedelta(days=140))
        chunk = _fetch_daily_chart_chunk(
            code,
            cursor_start.strftime("%Y%m%d"),
            cursor_end.strftime("%Y%m%d"),
            period_code=period_code,
            adjusted=adjusted,
        )
        if chunk.empty:
            break
        chunks.append(chunk)
        oldest = chunk.index.min()
        if oldest <= start_dt:
            break
        cursor_end = oldest - pd.Timedelta(days=1)

    if not chunks:
        df = pd.DataFrame(columns=["Open", "High", "Low", "Close", "Volume"])
    else:
        df = pd.concat(chunks).sort_index()
        df = df[~df.index.duplicated(keep="first")]
        df = df.loc[(df.index >= start_dt) & (df.index <= end_dt)]
    if use_cache and not df.empty:
        _cache_save_df(cache_key, df)
    return df


def get_ohlcv_wide(
    tickers: Iterable[str],
    start: str | date | datetime,
    end: str | date | datetime | None = None,
    adjusted: bool = False,
    use_cache: bool = True,
) -> pd.DataFrame:
    """yc_ohlcv_wide 호환: MultiIndex columns (OHLCV, Ticker)."""
    frames: dict[str, pd.DataFrame] = {}
    for tk in tickers:
        try:
            df = get_ohlcv_kis(tk, start, end, adjusted=adjusted, use_cache=use_cache)
            if not df.empty:
                frames[_normalize_ticker(tk)] = df
        except Exception as exc:  # noqa: BLE001
            logger.warning("get_ohlcv_kis fail %s: %s", tk, exc)
    if not frames:
        return pd.DataFrame()
    combined = pd.concat(frames, axis=1, names=["Ticker", "OHLCV"])
    combined = combined.swaplevel(0, 1, axis=1).sort_index(axis=1)
    return combined


def get_close_wide(
    tickers: Iterable[str],
    start: str | date | datetime,
    end: str | date | datetime | None = None,
    adjusted: bool = False,
    use_cache: bool = True,
    min_coverage: float = 0.0,
    ffill: bool = True,
) -> pd.DataFrame:
    """yc_close_wide 호환: Close 만 wide DataFrame (cols=ticker)."""
    df = get_ohlcv_wide(tickers, start, end, adjusted=adjusted, use_cache=use_cache)
    if df.empty:
        return pd.DataFrame()
    close = df["Close"]
    if min_coverage > 0:
        coverage = close.notna().mean(axis=0)
        keep = coverage[coverage >= min_coverage].index
        close = close[keep]
    if ffill:
        close = close.ffill().dropna(how="all")
    return close


# ────────────────────────────────────────────────────────────────────
# 2) 신고/신저 근접 종목 상위 — FHPST01700000
#    /uapi/domestic-stock/v1/ranking/near-new-highlow
# ────────────────────────────────────────────────────────────────────
TR_NEAR_HIGH = "FHPST01700000"
PATH_NEAR_HIGH = "/uapi/domestic-stock/v1/ranking/near-new-highlow"


def get_neighbour_high_low_top(
    market: str = "0000",  # 0000=전체, 0001=KOSPI, 1001=KOSDAQ
    direction: str = "HIGH",  # HIGH=신고가 근접, LOW=신저가 근접
    near_pct: float = 0.5,  # 0.5% 이내
    period: str = "60",  # 일수 — 60 = 60일 신고가
    use_cache: bool = True,
) -> pd.DataFrame:
    """신고/신저 근접 종목 상위. 60일 신고가 1초 조회.

    Returns: DataFrame[ticker, name, close, prdy_ctrt, near_rate, ...]
    """
    cache_key = f"near_{direction}_{market}_{period}_{int(near_pct*100)}"
    if use_cache:
        cached = _cache_load_df(cache_key)
        if cached is not None:
            return cached

    params = {
        "FID_COND_MRKT_DIV_CODE": "J",
        "FID_COND_SCR_DIV_CODE": "20170",
        "FID_INPUT_ISCD": market,
        "FID_DIV_CLS_CODE": "0",  # 0=전체
        "FID_INPUT_CNT_1": period,  # 일수
        "FID_PRC_CLS_CODE": "1" if direction.upper() == "HIGH" else "0",
        "FID_INPUT_PRICE_1": "",
        "FID_INPUT_PRICE_2": "",
        "FID_VOL_CNT": "",
        "FID_TRGT_EXLS_CLS_CODE": "0",
        "FID_TRGT_CLS_CODE": "0",
        "FID_RANK_SORT_CLS_CODE": "0",  # 0=상승률순
        "FID_RSFL_RATE1": "",  # 등락률 하한 (공란=무제한)
        "FID_RSFL_RATE2": "",  # 등락률 상한
    }
    try:
        data = _kis_get(PATH_NEAR_HIGH, TR_NEAR_HIGH, params)
    except Exception as exc:  # noqa: BLE001
        logger.warning("near_high fetch failed: %s — falling back", exc)
        return pd.DataFrame()
    rows = data.get("output") or []
    records = []
    for row in rows:
        records.append(
            {
                "ticker": (row.get("mksc_shrn_iscd") or row.get("stck_shrn_iscd") or "").zfill(6),
                "name": row.get("hts_kor_isnm") or "",
                "close": float(row.get("stck_prpr") or 0),
                "prdy_ctrt": float(row.get("prdy_ctrt") or 0),
                "near_rate": float(row.get("stck_hgpr_irds") or row.get("hgpr_hour") or 0),
                "volume": float(row.get("acml_vol") or 0),
            }
        )
    df = pd.DataFrame.from_records(records)
    if use_cache and not df.empty:
        _cache_save_df(cache_key, df)
    return df


# ────────────────────────────────────────────────────────────────────
# 3) 거래량/거래대금 순위 — FHPST01710000
#    /uapi/domestic-stock/v1/quotations/volume-rank
# ────────────────────────────────────────────────────────────────────
TR_VOLUME_RANK = "FHPST01710000"
PATH_VOLUME_RANK = "/uapi/domestic-stock/v1/quotations/volume-rank"


def get_volume_rank(
    market: str = "0000",
    blng_cls_code: str = "1",  # 1=거래대금, 0=거래량, 2=거래대금증감
    use_cache: bool = True,
) -> pd.DataFrame:
    """거래량/거래대금 순위 상위 30."""
    cache_key = f"volrank_{market}_{blng_cls_code}"
    if use_cache:
        cached = _cache_load_df(cache_key)
        if cached is not None:
            return cached
    params = {
        "FID_COND_MRKT_DIV_CODE": "J",
        "FID_COND_SCR_DIV_CODE": "20171",
        "FID_INPUT_ISCD": market,
        "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": "",
        "FID_INPUT_PRICE_2": "",
        "FID_VOL_CNT": "",
        "FID_INPUT_DATE_1": "",
    }
    try:
        data = _kis_get(PATH_VOLUME_RANK, TR_VOLUME_RANK, params)
    except Exception as exc:  # noqa: BLE001
        logger.warning("volume_rank failed: %s", exc)
        return pd.DataFrame()
    rows = data.get("output") or []
    records = []
    for row in rows:
        records.append(
            {
                "ticker": (row.get("mksc_shrn_iscd") or "").zfill(6),
                "name": row.get("hts_kor_isnm") or "",
                "rank": int(row.get("data_rank") or 0),
                "close": float(row.get("stck_prpr") or 0),
                "prdy_ctrt": float(row.get("prdy_ctrt") or 0),
                "volume": float(row.get("acml_vol") or 0),
                "trd_value": float(row.get("acml_tr_pbmn") or 0),
            }
        )
    df = pd.DataFrame.from_records(records)
    if use_cache and not df.empty:
        _cache_save_df(cache_key, df)
    return df


# ────────────────────────────────────────────────────────────────────
# 4) 업종 일자별 지수 — FHPUP02120000
#    /uapi/domestic-stock/v1/quotations/inquire-index-daily-price
# ────────────────────────────────────────────────────────────────────
TR_INDEX_DAILY = "FHPUP02120000"
PATH_INDEX_DAILY = "/uapi/domestic-stock/v1/quotations/inquire-index-daily-price"

# 자주 쓰는 업종/지수 코드 (KIS 한국거래소 기준)
SECTOR_CODES = {
    "KOSPI": "0001",
    "KOSDAQ": "1001",
    "KOSPI200": "2001",
    "KRX300": "5300",
}


def get_sector_index_daily(
    sector: str,
    start: str | date | datetime | None = None,
    end: str | date | datetime | None = None,
    use_cache: bool = True,
) -> pd.DataFrame:
    """업종/지수 일자별 시세. sector: KOSPI / KOSDAQ / KOSPI200 / KRX300 / 4자리코드."""
    code = SECTOR_CODES.get(sector.upper(), sector)
    end_dt = pd.Timestamp(end) if end else pd.Timestamp.today().normalize()
    start_dt = pd.Timestamp(start) if start else end_dt - pd.Timedelta(days=400)
    cache_key = (
        f"sector_{code}_{start_dt.strftime('%Y%m%d')}_{end_dt.strftime('%Y%m%d')}"
    )
    if use_cache:
        cached = _cache_load_df(cache_key)
        if cached is not None:
            return cached

    params = {
        "FID_COND_MRKT_DIV_CODE": "U",
        "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",
    }
    try:
        data = _kis_get(PATH_INDEX_DAILY, TR_INDEX_DAILY, params)
    except Exception as exc:  # noqa: BLE001
        logger.warning("sector index failed: %s", exc)
        return pd.DataFrame()
    rows = data.get("output2") or data.get("output") or []
    records = []
    for row in rows:
        d = row.get("stck_bsop_date") or row.get("bsop_date")
        if not d:
            continue
        records.append(
            {
                "Date": pd.to_datetime(d, format="%Y%m%d"),
                "Open": float(row.get("bstp_nmix_oprc") or 0),
                "High": float(row.get("bstp_nmix_hgpr") or 0),
                "Low": float(row.get("bstp_nmix_lwpr") or 0),
                "Close": float(row.get("bstp_nmix_prpr") or 0),
                "Volume": float(row.get("acml_vol") or 0),
            }
        )
    if not records:
        return pd.DataFrame(columns=["Open", "High", "Low", "Close", "Volume"])
    df = pd.DataFrame.from_records(records).set_index("Date").sort_index()
    if use_cache and not df.empty:
        _cache_save_df(cache_key, df)
    return df


# ────────────────────────────────────────────────────────────────────
# 5) ETF NAV 비교추이 — FHPST02440000
#    /uapi/etfetn/v1/quotations/nav-comparison-daily-trend
# ────────────────────────────────────────────────────────────────────
TR_ETF_NAV = "FHPST02440000"
PATH_ETF_NAV = "/uapi/etfetn/v1/quotations/nav-comparison-daily-trend"


def get_etf_nav(
    ticker: str,
    days: int = 100,
    use_cache: bool = True,
) -> pd.DataFrame:
    """ETF NAV 비교 일자별. 최근 N일."""
    code = _normalize_ticker(ticker)
    cache_key = f"etfnav_{code}_{days}"
    if use_cache:
        cached = _cache_load_df(cache_key)
        if cached is not None:
            return cached
    params = {
        "FID_COND_MRKT_DIV_CODE": "J",
        "FID_INPUT_ISCD": code,
    }
    try:
        data = _kis_get(PATH_ETF_NAV, TR_ETF_NAV, params)
    except Exception as exc:  # noqa: BLE001
        logger.warning("etf_nav failed: %s", exc)
        return pd.DataFrame()
    raw = data.get("output2") or data.get("output") or []
    rows = raw if isinstance(raw, list) else [raw]
    records = []
    for row in rows[:days]:
        d = row.get("stck_bsop_date")
        if not d:
            continue
        records.append(
            {
                "Date": pd.to_datetime(d, format="%Y%m%d"),
                "Close": float(row.get("stck_clpr") or 0),
                "NAV": float(row.get("nav") or 0),
                "Premium": float(row.get("prdy_ctrt") or 0),
            }
        )
    if not records:
        return pd.DataFrame(columns=["Close", "NAV", "Premium"])
    df = pd.DataFrame.from_records(records).set_index("Date").sort_index()
    if use_cache and not df.empty:
        _cache_save_df(cache_key, df)
    return df


# ────────────────────────────────────────────────────────────────────
# 6) VI 발동 현황 — FHPST01390000
#    /uapi/domestic-stock/v1/quotations/inquire-vi-status
# ────────────────────────────────────────────────────────────────────
TR_VI = "FHPST01390000"
PATH_VI = "/uapi/domestic-stock/v1/quotations/inquire-vi-status"


def get_vi_status(market: str = "0000", use_cache: bool = False) -> pd.DataFrame:
    """변동성완화장치(VI) 발동 종목 현황. 실시간성 → cache 기본 OFF."""
    params = {
        "FID_COND_MRKT_DIV_CODE": "J",
        "FID_COND_SCR_DIV_CODE": "20139",
        "FID_INPUT_ISCD": market,
        "FID_DIV_CLS_CODE": "0",
        "FID_RANK_SORT_CLS_CODE": "0",
        "FID_INPUT_DATE_1": "",
        "FID_TRGT_CLS_CODE": "",
        "FID_TRGT_EXLS_CLS_CODE": "",
        "FID_MRKT_CLS_CODE": "",
        "FID_INPUT_PRICE_1": "",
        "FID_INPUT_PRICE_2": "",
        "FID_VOL_CNT": "",
    }
    try:
        data = _kis_get(PATH_VI, TR_VI, params)
    except Exception as exc:  # noqa: BLE001
        logger.warning("vi_status failed: %s", exc)
        return pd.DataFrame()
    rows = data.get("output") or []
    records = []
    for row in rows:
        records.append(
            {
                "ticker": (row.get("mksc_shrn_iscd") or "").zfill(6),
                "name": row.get("hts_kor_isnm") or "",
                "vi_type": row.get("vi_cls_code") or "",
                "vi_time": row.get("vi_kind_code") or "",
                "trgt_price": float(row.get("vi_prc") or 0),
                "release_time": row.get("vi_release_dt") or "",
            }
        )
    return pd.DataFrame.from_records(records)


# ────────────────────────────────────────────────────────────────────
# 7) 휴장일 — CTCA0903R
#    /uapi/domestic-stock/v1/quotations/chk-holiday
# ────────────────────────────────────────────────────────────────────
TR_HOLIDAY = "CTCA0903R"
PATH_HOLIDAY = "/uapi/domestic-stock/v1/quotations/chk-holiday"


def get_market_holidays(
    start: str | date | datetime | None = None,
    end: str | date | datetime | None = None,
    use_cache: bool = True,
) -> list[date]:
    """국내 휴장일 list (date). start~end 사이 휴장일만 반환."""
    end_dt = pd.Timestamp(end) if end else pd.Timestamp.today().normalize() + pd.Timedelta(days=365)
    start_dt = pd.Timestamp(start) if start else pd.Timestamp.today().normalize()
    cache_key = f"holidays_{start_dt.strftime('%Y%m%d')}_{end_dt.strftime('%Y%m%d')}"
    if use_cache:
        cached = _cache_load_json(cache_key)
        if cached is not None:
            return [pd.Timestamp(d).date() for d in cached]

    holidays: list[date] = []
    cursor = start_dt
    while cursor <= end_dt:
        params = {"BASS_DT": cursor.strftime("%Y%m%d"), "CTX_AREA_NK": "", "CTX_AREA_FK": ""}
        try:
            data = _kis_get(PATH_HOLIDAY, TR_HOLIDAY, params)
        except Exception as exc:  # noqa: BLE001
            logger.warning("holiday fetch failed at %s: %s", cursor, exc)
            break
        rows = data.get("output") or []
        for row in rows:
            d = row.get("bass_dt")
            if not d:
                continue
            opnd = row.get("opnd_yn") or "Y"
            if opnd.upper() != "Y":
                holidays.append(pd.to_datetime(d, format="%Y%m%d").date())
        # advance ~30 days to avoid hitting page limits
        cursor = cursor + pd.Timedelta(days=30)
    if use_cache:
        _cache_save_json(cache_key, [d.isoformat() for d in holidays])
    return holidays


# ────────────────────────────────────────────────────────────────────
# yfinance compatibility shim
# ────────────────────────────────────────────────────────────────────
def yf_download_compat(
    tickers,
    period: str | None = None,
    start: str | date | datetime | None = None,
    end: str | date | datetime | None = None,
    auto_adjust: bool = False,
    progress: bool = False,  # noqa: ARG001
    **_unused,
) -> pd.DataFrame:
    """yf.download(tickers, period=...) 와 비슷한 형태 반환 (KIS 단일/멀티 자동)."""
    if isinstance(tickers, str):
        ticker_list = [tickers]
        single = True
    else:
        ticker_list = list(tickers)
        single = len(ticker_list) == 1

    end_dt = pd.Timestamp(end) if end else pd.Timestamp.today().normalize()
    if start is not None:
        start_dt = pd.Timestamp(start)
    elif period:
        # parse "1y", "3y", "60d", "100d", "1mo"
        pmap = {"y": 365, "mo": 30, "d": 1, "wk": 7}
        n_str = "".join(c for c in period if c.isdigit()) or "1"
        unit = "".join(c for c in period if not c.isdigit()) or "y"
        days = int(n_str) * pmap.get(unit, 365)
        start_dt = end_dt - pd.Timedelta(days=days)
    else:
        start_dt = end_dt - pd.Timedelta(days=365)

    if single:
        return get_ohlcv_kis(ticker_list[0], start_dt, end_dt, adjusted=auto_adjust)
    return get_ohlcv_wide(ticker_list, start_dt, end_dt, adjusted=auto_adjust)


# ────────────────────────────────────────────────────────────────────
# Quick sanity self-test (CLI)
# ────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    import argparse

    p = argparse.ArgumentParser(description="KIS data smoke test")
    p.add_argument("--all", action="store_true", help="run all 7 fetches")
    p.add_argument("--ticker", default="005930", help="test ticker")
    args = p.parse_args()

    t0 = time.time()
    print(f"\n[1] OHLCV {args.ticker} (60d)")
    df = get_ohlcv_kis(args.ticker, pd.Timestamp.today() - pd.Timedelta(days=80), use_cache=False)
    print(f"  rows={len(df)}, last={df.tail(2).to_dict('records') if not df.empty else 'EMPTY'}")
    print(f"  elapsed={time.time()-t0:.2f}s")

    if args.all:
        t1 = time.time()
        print(f"\n[2] near-high (60d, KOSPI+KOSDAQ)")
        df = get_neighbour_high_low_top(period="60", use_cache=False)
        print(f"  rows={len(df)}, top5={df.head().to_dict('records')[:5] if not df.empty else 'EMPTY'}")
        print(f"  elapsed={time.time()-t1:.2f}s")

        t1 = time.time()
        print(f"\n[3] volume_rank")
        df = get_volume_rank(use_cache=False)
        print(f"  rows={len(df)}, top3={df.head(3).to_dict('records') if not df.empty else 'EMPTY'}")
        print(f"  elapsed={time.time()-t1:.2f}s")

        t1 = time.time()
        print(f"\n[4] sector KOSPI daily")
        df = get_sector_index_daily("KOSPI", use_cache=False)
        print(f"  rows={len(df)}, last={df.tail(2).to_dict('records') if not df.empty else 'EMPTY'}")
        print(f"  elapsed={time.time()-t1:.2f}s")

        t1 = time.time()
        print(f"\n[5] etf_nav 091170 (KODEX 은행)")
        df = get_etf_nav("091170", use_cache=False)
        print(f"  rows={len(df)}")
        print(f"  elapsed={time.time()-t1:.2f}s")

        t1 = time.time()
        print(f"\n[6] vi_status")
        df = get_vi_status()
        print(f"  rows={len(df)}")
        print(f"  elapsed={time.time()-t1:.2f}s")

        t1 = time.time()
        print(f"\n[7] holidays (next 90d)")
        hols = get_market_holidays(use_cache=False)
        print(f"  count={len(hols)}, first5={hols[:5]}")
        print(f"  elapsed={time.time()-t1:.2f}s")

    print(f"\nTOTAL elapsed={time.time()-t0:.2f}s")