alt/bin/kis-paper-daily-smoke

#!/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)