예제 소스

kis_client.py

# -*- coding: utf-8 -*-
"""KIS Open API REST 클라이언트 (GUI·matplotlib·Qt 불필요)."""

from __future__ import annotations

import json
import os
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

import pandas as pd
import requests
from dotenv import load_dotenv

BASE_REAL = "https://openapi.koreainvestment.com:9443"
BASE_PAPER = "https://openapivts.koreainvestment.com:29443"

# 국내주식 일/주/월/년 기간별 시세
TR_DAILY_ITEMCHART = "FHKST03010100"
API_DAILY_ITEMCHART = "********"

# 국내주식 현재가 호가·예상체결 (output1=호가, output2=예상체결)
TR_ASKING_PRICE_EXP_CCN = "FHKST01010200"
API_ASKING_PRICE_EXP_CCN = "********"


def _load_env() -> None:
    root = Path(__file__).resolve().parent
    for name in (".env.local", ".env"):
        p = root / name
        if p.is_file():
            load_dotenv(p, override=False)
            return
    load_dotenv(override=False)


def _truthy(val: str | None) -> bool:
    if not val:
        return False
    return val.strip().lower() in ("1", "true", "yes", "y", "on")


class KISClient:
    """oauth2/tokenP 발급 + GET 호출의 최소 패턴."""

    def __init__(self, *, paper: bool | None = None) -> None:
        _load_env()
        self.app_key = (os.environ.get("APP_KEY") or os.environ.get("KIS_APP_KEY") or "").strip()
        self.app_secret = (
            os.environ.get("APP_SECRET") or os.environ.get("KIS_APP_SECRET") or ""
        ).strip()
        if not self.app_key or not self.app_secret:
            raise RuntimeError(
                "APP_KEY / APP_SECRET 이 비어 있습니다. .env.local 을 설정해 주세요."
            )

        if paper is None:
            paper = _truthy(os.environ.get("KIS_PAPER"))
        self.paper = paper
        self.base_url = BASE_PAPER if paper else BASE_REAL
        self._token: str | None = None

        self._user_agent = (
            os.environ.get("KIS_USER_AGENT")
            or "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        )

    def _base_headers(self) -> dict[str, str]:
        h: dict[str, str] = {
            "Content-Type": "application/json; charset=utf-8",
            "Accept": "application/json",
            "User-Agent": self._user_agent,
            "appkey": self.app_key,
            "appsecret": self.app_secret,
        }
        if self._token:
            h["authorization"] = f"Bearer {self._token}"
        return h

    def issue_access_token(self) -> str:
        url = f"{self.base_url}/oauth2/tokenP"
        body = {
            "grant_type": "client_credentials",
            "appkey": self.app_key,
            "appsecret": self.app_secret,
        }
        # 토큰 발급 자체는 인증 헤더 없이 요청 (혹시 비유효 토큰이 들어가면 발급 거부될 수 있음)
        headers = {
            "Content-Type": "application/json; charset=utf-8",
            "Accept": "application/json",
            "User-Agent": self._user_agent,
        }
        r = requests.post(url, headers=headers, data=json.dumps(body), timeout=30)
        if r.status_code != 200:
            raise RuntimeError(f"토큰 발급 HTTP {r.status_code}: {r.text}")
        token = r.json().get("access_token")
        if not token:
            raise RuntimeError(f"토큰 응답에 access_token 없음: {r.text}")
        self._token = token
        return token

    @staticmethod
    def _is_token_expired_response(r: "requests.Response") -> bool:
        """KIS 응답이 토큰 만료/비유효를 의미하는지 판정.

        KIS 는 보통 HTTP 200 으로 응답하고 본문의 msg_cd 에 토큰 관련 코드를 담는다.
        가끔 401 도 발생할 수 있다.
        """
        if r.status_code == 401:
            return True
        try:
            body = r.json()
        except Exception:
            return False
        msg_cd = str(body.get("msg_cd") or "").upper()
        msg1 = str(body.get("msg1") or "")
        # 대표적인 토큰 만료/비유효 코드
        if msg_cd in ("EGW00104", "EGW00121", "EGW00123"):
            return True
        if "토큰" in msg1 and ("만료" in msg1 or "유효" in msg1 or "기간" in msg1):
            return True
        return False

    def _tr_id_for_domestic(self, tr_id: str) -> str:
        if self.paper and tr_id and tr_id[0] in ("T", "J", "C"):
            return "V" + tr_id[1:]
        return tr_id

    def get_json(
        self,
        path: str,
        tr_id: str,
        params: dict[str, Any],
        *,
        tr_cont: str = "",
        _retry: bool = True,
    ) -> dict[str, Any]:
        if not self._token:
            self.issue_access_token()
        url = f"{self.base_url}{path}"
        headers = self._base_headers()
        headers["tr_id"] = self._tr_id_for_domestic(tr_id)
        headers["custtype"] = "P"
        headers["tr_cont"] = tr_cont
        r = requests.get(url, headers=headers, params=params, timeout=30)

        # 토큰 만료 감지 → 1회 자동 재발급 후 재시도 (무한 재귀 방지: _retry=False)
        if _retry and self._is_token_expired_response(r):
            print("[KIS] 토큰 만료/비유효 감지 → 자동 재발급 시도", file=sys.stderr)
            self._token = None
            try:
                self.issue_access_token()
            except Exception as exc:
                raise RuntimeError(f"토큰 자동 재발급 실패: {exc}") from exc
            return self.get_json(path, tr_id, params, tr_cont=tr_cont, _retry=False)

        if r.status_code != 200:
            raise RuntimeError(f"API HTTP {r.status_code}: {r.text}")
        return r.json()

    def inquire_daily_itemchart(
        self,
        *,
        fid_input_iscd: str = "005930",
        period: str = "D",  # D=일, W=주, M=월, Y=년
        start_date: str | None = None,  # YYYYMMDD
        end_date: str | None = None,  # YYYYMMDD
        adj_price: bool = True,
    ) -> dict[str, Any]:
        if end_date is None:
            end_date = datetime.now().strftime("%Y%m%d")
        if start_date is None:
            # 기본 ~7개월
            start_date = (datetime.now() - timedelta(days=210)).strftime("%Y%m%d")
        params = {
            "FID_COND_MRKT_DIV_CODE": "J",  # KRX
            "FID_INPUT_ISCD": fid_input_iscd,
            "FID_INPUT_DATE_1": start_date,
            "FID_INPUT_DATE_2": end_date,
            "FID_PERIOD_DIV_CODE": period,
            "FID_ORG_ADJ_PRC": "0" if adj_price else "1",
        }
        return self.get_json(API_DAILY_ITEMCHART, TR_DAILY_ITEMCHART, params)

    def inquire_asking_price_exp_ccn(
        self,
        fid_input_iscd: str,
        *,
        fid_cond_mrkt_div_code: str = "J",
    ) -> dict[str, Any]:
        """주식현재가 호가/예상체결. ``output1`` 매수·매도 1~10호가, ``output2`` 예상체결 등."""
        params = {
            "FID_COND_MRKT_DIV_CODE": fid_cond_mrkt_div_code,
            "FID_INPUT_ISCD": fid_input_iscd,
        }
        return self.get_json(API_ASKING_PRICE_EXP_CCN, TR_ASKING_PRICE_EXP_CCN, params)


def _num_from_output(val: Any) -> float | None:
    if val is None or val == "":
        return None
    try:
        x = float(val)
    except (TypeError, ValueError):
        return None
    if x <= 0:
        return None
    return x


def parse_orderbook_output1(output1: dict[str, Any] | None) -> dict[str, float | None]:
    """``inquire_asking_price_exp_ccn`` 의 ``output1`` 에서 대표 호가·잔량·당일 고저 추출."""
    if not output1 or not isinstance(output1, dict):
        return {
            "bidp1": None,
            "askp1": None,
            "stck_prpr": None,
            "stck_hgpr": None,
            "stck_lwpr": None,
            "bidp_rsqn1": None,
            "askp_rsqn1": None,
            "total_bidp_rsqn": None,
            "total_askp_rsqn": None,
        }
    return {
        "bidp1": _num_from_output(output1.get("bidp1")),
        "askp1": _num_from_output(output1.get("askp1")),
        "stck_prpr": _num_from_output(output1.get("stck_prpr")),
        "stck_hgpr": _num_from_output(output1.get("stck_hgpr")),
        "stck_lwpr": _num_from_output(output1.get("stck_lwpr")),
        "bidp_rsqn1": _num_from_output(output1.get("bidp_rsqn1")),
        "askp_rsqn1": _num_from_output(output1.get("askp_rsqn1")),
        "total_bidp_rsqn": _num_from_output(output1.get("total_bidp_rsqn")),
        "total_askp_rsqn": _num_from_output(output1.get("total_askp_rsqn")),
    }


def candles_to_dataframe(api_body: dict[str, Any]) -> pd.DataFrame:
    """KIS 응답(output2) → OHLCV DataFrame (index=Date)."""
    rows = api_body.get("output2") or []
    if not rows:
        return pd.DataFrame(columns=["Open", "High", "Low", "Close", "Volume"])
    df = pd.DataFrame(rows).rename(
        columns={
            "stck_bsop_date": "Date",
            "stck_oprc": "Open",
            "stck_hgpr": "High",
            "stck_lwpr": "Low",
            "stck_clpr": "Close",
            "acml_vol": "Volume",
        }
    )
    keep = ["Date", "Open", "High", "Low", "Close", "Volume"]
    df = df[[c for c in keep if c in df.columns]].copy()
    df["Date"] = pd.to_datetime(df["Date"], format="%Y%m%d", errors="coerce")
    for col in ("Open", "High", "Low", "Close", "Volume"):
        df[col] = pd.to_numeric(df[col], errors="coerce")
    return df.dropna(subset=["Date"]).set_index("Date").sort_index()