키움 REST 인증 — 토큰 발급·rate limit, 그리고 토큰이 갑자기 죽는 함정

키움 REST API는 모든 요청에 토큰을 요구한다. 그래서 자동매매의 첫 코드는 매매가 아니라 인증이다. 이 편에서 토큰을 받고 첫 조회를 성공시킨다.

OAuth2 — 토큰 받기

appkey/secretkey를 /oauth2/token에 보내면 토큰이 나온다.

import json, urllib.request

def get_token(base, appkey, secretkey):
    body = json.dumps({
        "grant_type": "client_credentials",
        "appkey": appkey,
        "secretkey": secretkey,
    }).encode("utf-8")
    req = urllib.request.Request(
        base + "/oauth2/token", data=body,
        headers={"Content-Type": "application/json;charset=UTF-8"},
        method="POST")
    with urllib.request.urlopen(req, timeout=15) as resp:
        d = json.loads(resp.read().decode("utf-8"))
    if d.get("return_code") != 0:
        raise RuntimeError(f"토큰 실패: {d.get('return_msg')}")
    return d["token"], d.get("expires_dt")   # 'YYYYMMDDHHMMSS'

표준 urllib만 썼다. 의존성 0.

토큰을 매번 받지 말 것 — 캐싱

토큰은 유효기간이 있다. 매 요청마다 재발급하면 키움이 차단한다. 받은 토큰을 저장해두고, 만료 60초 전에만 다시 받는다.

def ensure_token(self, force=False):
    if not force and self._token and time.time() < self._token_exp - 60:
        return self._token          # 아직 유효 → 그대로 사용
    # ... 재발급 후 self._token, self._token_exp 갱신

Rate limit — 초당 1회, 직렬화

TR(거래 요청)은 초당 약 1회 제한이 있다. 여러 스레드가 동시에 호출하면 막힌다. 그래서 락으로 호출을 한 줄로 세우고, 마지막 호출과의 간격을 지킨다.

with self._lock:
    gap = time.time() - self._last_call
    if gap < self.min_interval:
        time.sleep(self.min_interval - gap)   # 간격 확보
    # ... 실제 호출
    self._last_call = time.time()

TR 호출 패턴

모든 거래 요청은 같은 모양이다: URL + api-id 헤더(TR코드) + Bearer 토큰.

headers = {
    "authorization": f"Bearer {token}",
    "api-id": api_id,          # 예: "usa20100" (현재가)
}
# 페이징이 필요하면 cont-yn / next-key 헤더 추가

응답의 return_code가 0이면 성공. 참고로 20은 "조회 결과 없음"인데 에러가 아니라 정상 상황(예: 미체결 내역이 없음)이다. 이런 코드 구분이 실전에서 중요하다.

⚠️ 실전 함정 — 토큰이 갑자기 죽는다 (8005)

가장 골치 아픈 실장애다. 같은 appkey를 다른 프로그램이 써서 토큰을 새로 발급하면, 내 토큰이 즉시 무효화된다. 그런데 명목 만료시각은 아직 남아 있어서, 캐시만 믿으면 재시작 전까지 모든 요청이 실패한다.

해결: 응답에 8005 마크가 보이면 토큰을 강제 재발급하고 딱 한 번 재시도한다. 단, 재발급 엔드포인트 연타를 막으려 최소 30초 쿨다운을 둔다.

if "8005" in str(d.get("return_msg") or ""):
    # 강제 재발급 후 1회 재시도 (쿨다운 체크)
    ...

이런 "실제로 겪어봐야 아는" 처리들이 자동매매를 오래 죽지 않게 만든다. 문서엔 없다.

첫 호출 — 안전한 조회로 검증

계좌·주문을 건드리기 전에, 부작용 없는 조회로 연결을 확인하자. 예를 들어 ka10008(주식 외국인 매매동향)은 계좌·주문과 무관해서 첫 검증에 안전하다.

d = client.call("ka10008", "/api/dostk/frgnistt", {"stk_cd": "005930"})
print(d.get("return_code"))   # 0 이면 인증·통신 정상

여기까지 성공하면 토큰·인증·통신이 모두 정상이라는 뜻이다.

📦 여기까지를 한 파일로 — 복붙하면 돌아갑니다

지금까지의 조각을 모으면, 그대로 실행 가능한 최소 클라이언트가 된다. hero/kiwoom_rest.py로 저장하자. (표준 라이브러리만 쓰므로 pip install 불필요.)

# hero/kiwoom_rest.py
import json, time, threading, urllib.request, urllib.error

REAL_BASE = "https://api.kiwoom.com"
MOCK_BASE = "https://mockapi.kiwoom.com"

class KiwoomRestClient:
    def __init__(self, appkey, secretkey, mock=True, min_interval=1.0, timeout=15):
        self.appkey, self.secretkey = appkey, secretkey
        self.base = MOCK_BASE if mock else REAL_BASE
        self.min_interval, self.timeout = min_interval, timeout
        self._token, self._token_exp = None, 0.0
        self._lock = threading.Lock()
        self._last_call = 0.0

    def _post(self, path, body, headers):
        data = json.dumps(body).encode("utf-8")
        req = urllib.request.Request(
            self.base + path, data=data, method="POST",
            headers={"Content-Type": "application/json;charset=UTF-8", **headers})
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as r:
                return r.getcode(), json.loads(r.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            try:    return e.code, json.loads(e.read().decode("utf-8"))
            except Exception: return e.code, {}

    def _ensure_token(self):
        if self._token and time.time() < self._token_exp - 60:
            return self._token                       # 캐시된 유효 토큰 재사용
        _, d = self._post("/oauth2/token", {
            "grant_type": "client_credentials",
            "appkey": self.appkey, "secretkey": self.secretkey}, {})
        if d.get("return_code") != 0 or not d.get("token"):
            raise RuntimeError(f"토큰 실패: {d.get('return_msg')}")
        self._token = d["token"]
        try:
            self._token_exp = time.mktime(time.strptime(d["expires_dt"], "%Y%m%d%H%M%S"))
        except Exception:
            self._token_exp = time.time() + 3600
        print("토큰 발급 OK (만료", d.get("expires_dt"), ")")
        return self._token

    def call(self, api_id, path, body):
        with self._lock:                             # rate limit 대비 직렬화
            gap = time.time() - self._last_call
            if gap < self.min_interval:
                time.sleep(self.min_interval - gap)
            token = self._ensure_token()
            code, d = self._post(path, body,
                {"authorization": f"Bearer {token}", "api-id": api_id})
            self._last_call = time.time()
        rc = d.get("return_code")
        if code >= 400 or rc not in (0, None):
            raise RuntimeError(f"{api_id} 실패 (HTTP {code}, rc={rc}): {d.get('return_msg')}")
        return d


if __name__ == "__main__":
    import pathlib
    cfg = json.loads(pathlib.Path("config/settings.json").read_text("utf-8"))["kiwoom_rest"]
    client = KiwoomRestClient(cfg["appkey"], cfg["secretkey"], mock=cfg.get("mock", True))
    token = client._ensure_token()      # 토큰 발급이 되면 인증·통신 정상 (실전·모의 공통)
    print("토큰 앞 12자리:", token[:12], "…")

실전 클라이언트에는 여기에 8005 강제 재발급 재시도(위 참고)와 페이징(cont-yn/next-key)이 더 붙는다. 하지만 "돌아가는 것"은 이게 전부다.

✅ 확인

python hero/kiwoom_rest.py

정상이면 이렇게 출력된다:

토큰 발급 OK (만료 20260827xxxxxx )
토큰 앞 12자리: eyJhbGciOi …

토큰 발급은 실전·모의 양쪽에서 되므로, 여기까지 뜨면 인증·통신은 확실히 통과한 것이다. (첫 시세 조회는 4편에서.) 토큰 실패가 뜨면 키가 맞는지 / API 사용 신청이 승인됐는지만 확인하면 된다 — 에러의 return_msg를 그대로 댓글에 남겨주면 같이 봐드릴게요.

다음 편 예고

4편에서는 진짜 매매에 필요한 데이터 — 미국주식 현재가·잔고·체결내역을 조회합니다. 그리고 초보자가 반드시 헷갈리는 "원장일" 개념(한국시간 자정에 안 바뀌는 미국 거래일)을 정리합니다.

---

토큰 발급이나 첫 호출에서 에러가 나면, 응답의 return_msg를 그대로 댓글에 남겨주세요. 같이 봐드릴게요.