QuantEngineByItz/src/quant_engine/kis_api_client_v1.py

"""한국투자증권(KIS) Open API 클라이언트 — 조회(read-only) 전용.

근거: https://apiportal.koreainvestment.com/apiservice-summary ,
      https://github.com/koreainvestment/open-trading-api (2026-06-21 실측 확인된
      api_url/tr_id만 사용 — 추정 금지).

══════════════════════════════════════════════════════════════════════════════
[CRITICAL] governance/rules/06_no_direct_api_trading.yaml — 절대 규칙
이 모듈은 매수/매도 주문을 어떤 경로로도 제출하지 않는다. 주문 제출/정정/취소
함수는 이 파일에 일체 작성하지 않으며, 공유 요청 함수(_send_request)는 주문
관련 경로("/trading/")나 TR_ID(TTTC08*/VTTC08* 등)를 만나면 즉시 RuntimeError로
요청을 차단한다(2차 방어). 이 원칙을 어기면 엔진 전체가 '제안 시스템'에서
'자동매매 시스템'으로 변질되어 프로젝트 핵심 전제가 깨진다(사용자 직접 지시).
══════════════════════════════════════════════════════════════════════════════

인증 정보는 Windows 환경변수에서 읽는다(실제계좌: KIS_APP_Key/KIS_APP_Secret,
모의계좌: KIS_APP_Key_TEST/KIS_APP_Secret_TEST). 방금 setx로 설정된 값은 현재
프로세스의 os.environ에 아직 반영되지 않을 수 있어, HKCU\\Environment 레지스트리
폴백을 둔다(읽기만 함, 값을 로그에 남기지 않음).
"""
from __future__ import annotations

import datetime as dt
import json
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

REAL_DOMAIN = "https://openapi.koreainvestment.com:9443"
MOCK_DOMAIN = "https://openapivts.koreainvestment.com:29443"
TOKEN_CACHE_DIR = ROOT / "Temp"
TOKEN_CACHE_DB_NAME = "kis_tokens.db"
TOKEN_REFRESH_SKEW_MINUTES = 10


def _requests():
    try:
        import requests  # type: ignore
    except Exception as exc:  # pragma: no cover - surfaced as runtime validation error
        raise RuntimeError(f"import_error: {exc}") from exc
    return requests

# ── [CRITICAL] 주문 차단 목록 — 절대 수정/완화 금지 (governance/rules/06_no_direct_api_trading.yaml) ──
# "/trading/" 하위 경로는 주문(order)뿐 아니라 계좌잔고조회(inquire-balance)도 포함한다.
# 계좌 보유종목/잔고는 governance/rules/07_no_kis_account_balance_query.yaml에 의해
# 별도로도 금지된다 — HTS 캡처가 유일한 출처(사용자 직접 지시).
FORBIDDEN_PATH_SUBSTRINGS: tuple[str, ...] = ("/trading/",)
FORBIDDEN_TR_ID_PREFIXES: tuple[str, ...] = (
    "TTTC08", "VTTC08", "TTTC01", "VTTC01",  # 현금/신용 매수·매도·정정·취소
    "TTTC8434R", "VTTC8434R",  # 주식잔고조회 — 계좌 보유종목 조회 금지(07번 규칙)
)


class OrderEndpointBlockedError(RuntimeError):
    """주문 제출/정정/취소 경로 호출 시도 — 절대 차단."""


def _assert_read_only(path: str, tr_id: str) -> None:
    for forbidden in FORBIDDEN_PATH_SUBSTRINGS:
        if forbidden in path:
            raise OrderEndpointBlockedError(
                f"BLOCKED: 주문 관련 경로 호출 시도 차단 — path={path!r}. "
                "이 엔진은 매수/매도를 API로 직접 실행하지 않는다(governance/rules/06_no_direct_api_trading.yaml)."
            )
    for prefix in FORBIDDEN_TR_ID_PREFIXES:
        if tr_id.upper().startswith(prefix):
            raise OrderEndpointBlockedError(
                f"BLOCKED: 주문 관련 TR_ID 호출 시도 차단 — tr_id={tr_id!r}. "
                "이 엔진은 매수/매도를 API로 직접 실행하지 않는다(governance/rules/06_no_direct_api_trading.yaml)."
            )


def _read_env_var(name: str) -> str | None:
    import os

    value = os.environ.get(name)
    if value:
        return value
    if sys.platform != "win32":
        return None
    try:
        import winreg

        with winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Environment") as key:
            value, _ = winreg.QueryValueEx(key, name)
            return value or None
    except OSError:
        return None


class KisCredentials:
    def __init__(self, app_key: str, app_secret: str, account: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.account = account  # "real" | "mock"
        self.domain = REAL_DOMAIN if account == "real" else MOCK_DOMAIN

    @classmethod
    def load(cls, account: str = "mock") -> "KisCredentials":
        if account == "real":
            key_name, secret_name = "KIS_APP_Key", "KIS_APP_Secret"
        elif account == "mock":
            key_name, secret_name = "KIS_APP_Key_TEST", "KIS_APP_Secret_TEST"
        else:
            raise ValueError("account must be 'real' or 'mock'")
        app_key = _read_env_var(key_name)
        app_secret = _read_env_var(secret_name)
        if not app_key or not app_secret:
            raise RuntimeError(
                f"{key_name}/{secret_name} 환경변수를 찾을 수 없음 — Windows 환경변수 설정 후 "
                "새 셸에서 재시도하거나 HKCU\\Environment 레지스트리 반영을 확인하세요."
            )
        return cls(app_key=app_key, app_secret=app_secret, account=account)


import sqlite3

def _token_db_path() -> Path:
    import os

    override = os.environ.get("KIS_TOKEN_DB_PATH", "").strip()
    if override:
        return Path(override)
    TOKEN_CACHE_DIR.mkdir(parents=True, exist_ok=True)
    return TOKEN_CACHE_DIR / TOKEN_CACHE_DB_NAME


def _init_token_db(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS kis_tokens (
            account TEXT PRIMARY KEY,
            access_token TEXT NOT NULL,
            expires_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.commit()


def _issue_or_reuse_token(creds: KisCredentials) -> str:
    """KIS는 토큰 발급 빈도를 제한한다 — 만료 전까지 DB 캐시 재사용 필수."""
    db_path = _token_db_path()
    db_path.parent.mkdir(parents=True, exist_ok=True)

    with sqlite3.connect(db_path, timeout=30) as conn:
        _init_token_db(conn)
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT access_token, expires_at FROM kis_tokens WHERE account = ?",
            (creds.account,),
        ).fetchone()
        if row:
            token, expires_at_str = row
            try:
                expires_at = dt.datetime.fromisoformat(expires_at_str)
                # 만료 시간 10분 전까지 재사용 가능 여부 검사
                if dt.datetime.now(dt.timezone.utc) < expires_at - dt.timedelta(minutes=TOKEN_REFRESH_SKEW_MINUTES):
                    conn.commit()
                    return token
            except ValueError:
                pass

        # 2. 토큰이 만료되었거나 없을 시 KIS API로 새로 발급 요청
        requests = _requests()
        try:
            resp = requests.post(
                f"{creds.domain}/oauth2/tokenP",
                json={"grant_type": "client_credentials", "appkey": creds.app_key, "appsecret": creds.app_secret},
                timeout=15,
            )
            resp.raise_for_status()
            body = resp.json()
        except Exception as exc:
            raise RuntimeError("KIS token refresh failed; check credentials and API availability.") from exc
        access_token = body["access_token"]
        expires_in_sec = int(body.get("expires_in", 86400))
        expires_at = dt.datetime.now(dt.timezone.utc) + dt.timedelta(seconds=expires_in_sec)

        # 3. 새로운 토큰 정보를 DB에 안전하게 업서트
        conn.execute(
            """
            INSERT OR REPLACE INTO kis_tokens (account, access_token, expires_at, updated_at)
            VALUES (?, ?, ?, ?)
            """,
            (
                creds.account,
                access_token,
                expires_at.isoformat(),
                dt.datetime.now(dt.timezone.utc).isoformat(),
            ),
        )
        conn.commit()
        return access_token


def _send_request(creds: KisCredentials, path: str, tr_id: str, params: dict[str, Any]) -> dict[str, Any]:
    """모든 KIS REST 호출의 단일 진입점 — 여기서만 가드가 작동하면 충분하다."""
    _assert_read_only(path, tr_id)  # [CRITICAL] 절대 제거 금지
    access_token = _issue_or_reuse_token(creds)
    headers = {
        "content-type": "application/json; charset=utf-8",
        "authorization": f"Bearer {access_token}",
        "appkey": creds.app_key,
        "appsecret": creds.app_secret,
        "tr_id": tr_id,
        "custtype": "P",
    }
    requests = _requests()
    try:
        resp = requests.get(f"{creds.domain}{path}", headers=headers, params=params, timeout=15)
        resp.raise_for_status()
        return resp.json()
    except Exception as exc:
        raise RuntimeError(f"KIS read-only request failed for {path} / {tr_id}.") from exc


# ── 조회(read-only) 함수 — 전부 GET, 전부 quotations/ranking 카테고리 (실측 확인) ──────────

def get_current_price(creds: KisCredentials, code: str) -> dict[str, Any]:
    """주식현재가 시세. api_url=/uapi/domestic-stock/v1/quotations/inquire-price, tr_id=FHKST01010100."""
    return _send_request(
        creds, "/uapi/domestic-stock/v1/quotations/inquire-price", "FHKST01010100",
        {"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": code},
    )


def get_asking_price_10_level(creds: KisCredentials, code: str) -> dict[str, Any]:
    """주식현재가 호가/예상체결 — 10단계 매수/매도 호가.
    api_url=/uapi/domestic-stock/v1/quotations/inquire-asking-price-exp-ccn, tr_id=FHKST01010200.
    """
    return _send_request(
        creds, "/uapi/domestic-stock/v1/quotations/inquire-asking-price-exp-ccn", "FHKST01010200",
        {"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": code},
    )


def get_daily_short_sale(creds: KisCredentials, code: str, start_date: str, end_date: str) -> dict[str, Any]:
    """국내주식 공매도 일별추이. api_url=/uapi/domestic-stock/v1/quotations/daily-short-sale,
    tr_id=FHPST04830000. start_date/end_date: YYYYMMDD."""
    return _send_request(
        creds, "/uapi/domestic-stock/v1/quotations/daily-short-sale", "FHPST04830000",
        {"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": code,
         "FID_INPUT_DATE_1": start_date, "FID_INPUT_DATE_2": end_date},
    )


def get_daily_item_chart_price(
    creds: KisCredentials, code: str, start_date: str, end_date: str, period: str = "D",
) -> dict[str, Any]:
    """주식현재가 일자별. api_url=/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice,
    tr_id=FHKST03010100."""
    return _send_request(
        creds, "/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice", "FHKST03010100",
        {"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": code,
         "FID_INPUT_DATE_1": start_date, "FID_INPUT_DATE_2": end_date,
         "FID_PERIOD_DIV_CODE": period, "FID_ORG_ADJ_PRC": "0"},
    )


def get_investor_trend(creds: KisCredentials, code: str) -> dict[str, Any]:
    """주식현재가 투자자(개인/외국인/기관) 매매동향.
    api_url=/uapi/domestic-stock/v1/quotations/inquire-investor, tr_id=FHKST01010900."""
    return _send_request(
        creds, "/uapi/domestic-stock/v1/quotations/inquire-investor", "FHKST01010900",
        {"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": code},
    )