예제 소스

main.py

# -*- coding: utf-8 -*-
"""
한국투자증권 Open API(KIS Developers) 연동 예시.
베이스: https://github.com/koreainvestment/open-trading-api 의 examples_user 흐름
(oauth2/tokenP 발급 → domestic-stock inquire-price GET).

환경 변수(.env.local 권장):
  APP_KEY, APP_SECRET  — KIS 앱키/앱시크릿
  KIS_PAPER=1          — 모의투자(선택, 기본은 실전 도메인)
"""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path
from typing import Any

import requests
from dotenv import load_dotenv

# 저장소 샘플과 동일한 REST 베이스 URL
BASE_REAL = "https://openapi.koreainvestment.com:9443"
BASE_PAPER = "https://openapivts.koreainvestment.com:29443"

# 국내주식 현재가 시세 (examples_user/domestic_stock_functions.inquire_price 와 동일 TR)
TR_INQUIRE_PRICE = "FHKST01010100"
API_INQUIRE_PRICE = "********"


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:
    """open-trading-api 의 kis_auth.auth + _url_fetch(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:
            print(
                "APP_KEY / APP_SECRET 이 비어 있습니다. .env.local 에 설정해 주세요.",
                file=sys.stderr,
            )
            sys.exit(1)

        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,
        }
        if self._token:
            h["authorization"] = f"Bearer {self._token}"
        h["appkey"] = self.app_key
        h["appsecret"] = self.app_secret
        return h

    def issue_access_token(self) -> str:
        """POST /oauth2/tokenP — 샘플 kis_auth.auth 와 동일."""
        url = f"{self.base_url}/oauth2/tokenP"
        body = {
            "grant_type": "client_credentials",
            "appkey": self.app_key,
            "appsecret": self.app_secret,
        }
        r = requests.post(url, headers=self._base_headers(), data=json.dumps(body), timeout=30)
        if r.status_code != 200:
            raise RuntimeError(f"토큰 발급 HTTP {r.status_code}: {r.text}")
        data = r.json()
        token = data.get("access_token")
        if not token:
            raise RuntimeError(f"토큰 응답에 access_token 없음: {data}")
        self._token = token
        return token

    def _tr_id_for_domestic(self, tr_id: str) -> str:
        """모의투자 시 T/J/C 로 시작하는 TR 은 V 접두로 바꿈 (kis_auth._url_fetch)."""
        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 = "",
    ) -> 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)
        if r.status_code != 200:
            raise RuntimeError(f"API HTTP {r.status_code}: {r.text}")
        return r.json()

    def inquire_domestic_price(
        self,
        *,
        env_dv: str = "real",
        fid_cond_mrkt_div_code: str = "J",
        fid_input_iscd: str = "005930",
    ) -> dict[str, Any]:
        """
        주식 현재가 시세 (국내).
        env_dv: 'real' 실전 / 'demo' 모의 (샘플 domestic_stock_functions 와 동일).
        fid_cond_mrkt_div_code: J=KRX 등.
        """
        if env_dv not in ("real", "demo"):
            raise ValueError("env_dv 는 'real' 또는 'demo' 만 허용됩니다.")
        params = {
            "FID_COND_MRKT_DIV_CODE": fid_cond_mrkt_div_code,
            "FID_INPUT_ISCD": fid_input_iscd,
        }
        return self.get_json(API_INQUIRE_PRICE, TR_INQUIRE_PRICE, params)


def main() -> None:
    paper = _truthy(os.environ.get("KIS_PAPER"))
    client = KISClient(paper=paper)

    client.issue_access_token()
    print("접근토큰 발급 완료.")

    # 삼성전자 KRX 현재가 시세 (README 예제와 동일 파라미터)
    env_label = "demo" if paper else "real"
    body = client.inquire_domestic_price(
        env_dv=env_label,
        fid_cond_mrkt_div_code="J",
        fid_input_iscd="005930",
    )
    rt = body.get("rt_cd")
    msg = body.get("msg1", "")
    if rt != "0":
        print("API 오류:", rt, msg, file=sys.stderr)
        print(json.dumps(body, ensure_ascii=False, indent=2))
        sys.exit(1)

    out = body.get("output")
    print("조회 성공:", msg or "ok")
    print(json.dumps(out, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()