한국투자증권 API 인증과 현재가 조회

if (-not (Test-Path -LiteralPath 'config/.env')) {
    Copy-Item -LiteralPath 'config/example.env' -Destination 'config/.env'
}

from pathlib import Path
import re

import requests
from dotenv import dotenv_values

ROOT = Path(__file__).resolve().parents[1]
SERVERS = {
    "mock": "https://openapivts.koreainvestment.com:29443",
    "live": "https://openapi.koreainvestment.com:9443",
}
QUOTATIONS = {
    "price": (
        "/uapi/domestic-stock/v1/quotations/inquire-price",
        "FHKST01010100",
    ),
    "daily": (
        "/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice",
        "FHKST03010100",
    ),
}


def load_config():
    path = ROOT / "config" / ".env"
    if not path.is_file():
        raise FileNotFoundError("config/.env is missing")
    return dict(dotenv_values(path, encoding="utf-8-sig"))


class ReadOnlyKis:
    def __init__(self, config):
        mode = config.get("RUN_MODE", "read_only")
        if mode not in {"read_only", "paper"}:
            raise ValueError("This client is for read_only or local paper mode")
        env = config.get("KIS_ENV", "mock")
        if env not in SERVERS:
            raise ValueError("KIS_ENV must be mock or live")
        self.base_url = SERVERS[env]
        self.key = config.get("KIS_APP_KEY") or ""
        self.secret = config.get("KIS_APP_SECRET") or ""
        if any(not v or v.startswith("REPLACE_") for v in (self.key, self.secret)):
            raise ValueError("KIS credentials are not configured")
        self.session = requests.Session()
        self.token = None

    @staticmethod
    def _json(response):
        if response.status_code != 200:
            raise RuntimeError(f"KIS HTTP error: {response.status_code}")
        try:
            body = response.json()
        except ValueError as exc:
            raise RuntimeError("KIS returned non-JSON content") from exc
        if not isinstance(body, dict):
            raise RuntimeError("KIS returned an unexpected JSON structure")
        return body

    def authenticate(self):
        if self.token:
            return
        response = self.session.post(
            self.base_url + "/oauth2/tokenP",
            json={
                "grant_type": "client_credentials",
                "appkey": self.key,
                "appsecret": self.secret,
            },
            timeout=(5, 20),
        )
        body = self._json(response)
        token = body.get("access_token")
        if not isinstance(token, str) or not token:
            raise RuntimeError("KIS authentication failed; check the API portal")
        self.token = token

    def quotations(self, kind, params):
        if kind not in QUOTATIONS:
            raise ValueError("Unsupported quotation kind")
        self.authenticate()
        path, tr_id = QUOTATIONS[kind]
        response = self.session.get(
            self.base_url + path,
            headers={
                "authorization": f"Bearer {self.token}",
                "appkey": self.key,
                "appsecret": self.secret,
                "tr_id": tr_id,
                "custtype": "P",
            },
            params=params,
            timeout=(5, 20),
        )
        body = self._json(response)
        if body.get("rt_cd") != "0":
            code = str(body.get("msg_cd", "unknown"))[:40]
            raise RuntimeError(f"KIS quotation rejected: {code}")
        return body

    def current_price(self, symbol):
        if not re.fullmatch(r"[A-Z0-9]{6}", symbol):
            raise ValueError("A six-character uppercase ASCII stock symbol is required")
        body = self.quotations("price", {
            "FID_COND_MRKT_DIV_CODE": "J",
            "FID_INPUT_ISCD": symbol,
        })
        output = body.get("output")
        if not isinstance(output, dict):
            raise RuntimeError("Price output is missing")
        price = output.get("stck_prpr")
        if not isinstance(price, str) or not re.fullmatch(r"[0-9]+", price) or int(price) <= 0:
            raise RuntimeError("Price must be a positive integer")
        return output

    def close(self):
        self.session.close()

from src.kis_readonly import ReadOnlyKis, load_config


def main():
    config = load_config()
    client = ReadOnlyKis(config)
    try:
        output = client.current_price("005930")
        price = int(output["stck_prpr"])
        print(f"env={config.get('KIS_ENV', 'mock')} symbol=005930 price={price}")
        print("quotation OK; credentials and token omitted")
    finally:
        client.close()


if __name__ == "__main__":
    main()

.\.venv\Scripts\python.exe -m py_compile src/kis_readonly.py src/check_kis.py
.\.venv\Scripts\python.exe -m src.check_kis