#!/usr/bin/env python3 """Run a minimal KIS daily-price smoke without printing secrets.""" from __future__ import annotations import argparse import json import os import sys import urllib.error import urllib.parse import urllib.request ENV_CONFIG = { "paper": { "base_url": "https://openapivts.koreainvestment.com:29443", "app_key": "KIS_PAPER_APP_KEY", "app_secret": "KIS_PAPER_APP_SECRET", "cano": "KIS_PAPER_CANO", "product": "KIS_PAPER_ACNT_PRDT_CD", }, "real": { "base_url": "https://openapi.koreainvestment.com:9443", "app_key": "KIS_REAL_APP_KEY", "app_secret": "KIS_REAL_APP_SECRET", "cano": "KIS_REAL_CANO", "product": "KIS_REAL_ACNT_PRDT_CD", }, } TOKEN_PATH = "/oauth2/tokenP" DAILY_PATH = "/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice" DAILY_TR_ID = "FHKST03010100" USER_AGENT = ( "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36" ) class SmokeError(Exception): pass def required_env(name: str) -> str: value = os.environ.get(name, "") if value == "": raise SmokeError(f"missing required env: {name}") return value def redact(text: str, secrets: list[str]) -> str: redacted = text for secret in secrets: if secret: redacted = redacted.replace(secret, "[redacted]") return redacted def request_json( method: str, url: str, *, headers: dict[str, str], body: dict[str, str] | None = None, params: dict[str, str] | None = None, timeout: float, secrets: list[str], ) -> dict: if params: url = f"{url}?{urllib.parse.urlencode(params)}" data = None if body is not None: data = json.dumps(body).encode("utf-8") req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode("utf-8", errors="replace") except urllib.error.HTTPError as err: raw = err.read().decode("utf-8", errors="replace") raise SmokeError(f"HTTP {err.code}: {redact(raw, secrets)}") from err except urllib.error.URLError as err: raise SmokeError(f"network error: {err.reason}") from err try: return json.loads(raw) except json.JSONDecodeError as err: raise SmokeError(f"non-JSON response: {redact(raw[:1000], secrets)}") from err def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Run a KIS daily item chart price smoke.", ) parser.add_argument("--symbol", default=os.environ.get("KIS_SMOKE_SYMBOL", "005930")) parser.add_argument("--start", default=os.environ.get("KIS_SMOKE_START", "20240527")) parser.add_argument("--end", default=os.environ.get("KIS_SMOKE_END", "20240528")) parser.add_argument("--timeout", type=float, default=20.0) return parser.parse_args(argv) def validate_yyyymmdd(name: str, value: str) -> None: if len(value) != 8 or not value.isdigit(): raise SmokeError(f"{name} must be YYYYMMDD") def main(argv: list[str]) -> int: args = parse_args(argv) validate_yyyymmdd("--start", args.start) validate_yyyymmdd("--end", args.end) if args.start > args.end: raise SmokeError("--start must be before or equal to --end") if args.symbol == "" or not args.symbol.isdigit(): raise SmokeError("--symbol must be a numeric KRX symbol") active_env = os.environ.get("KIS_ACTIVE_ENV", "") env_config = ENV_CONFIG.get(active_env) if env_config is None: raise SmokeError('KIS_ACTIVE_ENV must be "paper" or "real" for this smoke') app_key = required_env(env_config["app_key"]) app_secret = required_env(env_config["app_secret"]) account_no = required_env(env_config["cano"]) product = required_env(env_config["product"]) if product != "01": raise SmokeError(f"{env_config['product']} must be 01 for domestic stock smoke") base_url = env_config["base_url"] static_secrets = [app_key, app_secret, account_no] base_headers = { "Content-Type": "application/json; charset=utf-8", "Accept": "application/json", "User-Agent": USER_AGENT, } token_payload = request_json( "POST", base_url + TOKEN_PATH, headers=base_headers, body={ "grant_type": "client_credentials", "appkey": app_key, "appsecret": app_secret, }, timeout=args.timeout, secrets=static_secrets, ) access_token = token_payload.get("access_token", "") if access_token == "": raise SmokeError( "token response missing access_token: " + redact(json.dumps(token_payload, ensure_ascii=False), static_secrets) ) query_payload = request_json( "GET", base_url + DAILY_PATH, headers={ **base_headers, "authorization": f"Bearer {access_token}", "appkey": app_key, "appsecret": app_secret, "tr_id": DAILY_TR_ID, "custtype": "P", "tr_cont": "", }, params={ "FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": args.symbol, "FID_INPUT_DATE_1": args.start, "FID_INPUT_DATE_2": args.end, "FID_PERIOD_DIV_CODE": "D", "FID_ORG_ADJ_PRC": "0", }, timeout=args.timeout, secrets=static_secrets + [access_token], ) if query_payload.get("rt_cd") != "0": raise SmokeError( "KIS daily query failed: " + redact(json.dumps(query_payload, ensure_ascii=False), static_secrets + [access_token]) ) output1 = query_payload.get("output1") or {} output2 = query_payload.get("output2") or [] if not isinstance(output2, list) or len(output2) == 0: raise SmokeError("KIS daily query returned no output2 rows") dates = sorted( row.get("stck_bsop_date", "") for row in output2 if isinstance(row, dict) and row.get("stck_bsop_date") ) if not dates: raise SmokeError("KIS daily rows did not include stck_bsop_date") print(f"KIS {active_env} daily smoke ok") print(f"env={active_env}") print("provider=kis") print("endpoint=inquire-daily-itemchartprice") print(f"symbol={args.symbol}") print(f"name={output1.get('hts_kor_isnm', '')}") print(f"bar_count={len(output2)}") print(f"first_date={dates[0]}") print(f"last_date={dates[-1]}") print(f"msg_cd={query_payload.get('msg_cd', '')}") print(f"msg={query_payload.get('msg1', '')}") return 0 if __name__ == "__main__": try: raise SystemExit(main(sys.argv[1:])) except SmokeError as err: print(f"KIS daily smoke failed: {err}", file=sys.stderr) raise SystemExit(1)