키움 REST | 분봉차트 조회 실전 가이드 (Python, 2025-10 최신)

업데이트(2025-10-16): 토큰 캐싱/연속조회(next-key) 예제 추가, 분봉 수집 파이프라인(스케줄·리트라이) 보강, 에러 트러블슈팅 표 수록.

안녕하세요, TechOracle입니다. “분봉 데이터를 안정적으로 모아서 내 전략에 쓰고 싶다”는 요청이 정말 많았어요. 이 글은 키움 REST 기반 분봉 수집을 처음부터 끝까지 서술형으로 안내합니다. 특히 토큰 관리·연속조회·휴일 처리·저장 구조를 한 번에 세팅해, 내일 아침부터 바로 굴러가도록 만드는 데 초점을 맞췄습니다.

---

#요약 (바로 적용)

• 토큰: OAuth 발급→캐시 파일 저장→만료 전 갱신
• 연속조회: 응답 헤더의 cont-yn/next-key 처리로 페이지 단절 방지
• 스로틀링: 지수 백오프 + 최대 재시도 n회
• 스케줄(Asia/Seoul): EOD 20:00에 전 종목 분봉 수집(또는 특정 유니버스)
• 저장: D:\stock\data\ohlcv_min_raw\KRX\{심볼}_{YYYYMMDD}_m{unit}.csv

원칙: “주문은 분리(주문 엔진), 분석 데이터는 표준 스키마로 저장” — 장애에 강해지고, 백테스트 재현성이 높아집니다.

---

1) 준비물과 디렉터리 표준

• 파이썬 가상환경: .venv64 (없으면 .venv32)
• 베이스 디렉터리: D:\stock
• 비밀키/토큰: D:\stock\secrets\kiwoom_cred_prod.json, D:\stock\secrets\kiwoom_token_prod.json
• 결과 저장: D:\stock\data\ohlcv_min_raw\KRX
• 로깅: D:\stock\logs\minute_fetch.log

권한과 보안

• secrets 폴더는 읽기 권한 최소화
• 토큰 파일은 git ignore
• IP 화이트리스트 등록(필요 시) 및 요청 상한 준수

---

2) 토큰 발급·갱신 (OAuth) — 한 번 세팅하면 편해집니다

키움 REST는 OAuth 기반입니다. 아래는 발급→캐시→갱신의 최소 예시 흐름입니다. (엔드포인트·헤더 키 이름은 브로커 공식 문서 기준으로 맞춰 주세요.)

# D:\stock\app\brokers\kiwoom_token_manager.py
import json, time, requests, pathlib
from datetime import datetime, timedelta

AUTH_URL = "https://api.kiwoom.com/oauth2/token"  # 운영 도메인 예시
CRED = json.load(open(r"D:\stock\secrets\kiwoom_cred_prod.json", "r", encoding="utf-8"))
CACHE = pathlib.Path(r"D:\stock\secrets\kiwoom_token_prod.json")


def issue_token():
    r = requests.post(AUTH_URL, json={
        "grant_type": "client_credentials",
        "client_id": CRED["client_id"],
        "client_secret": CRED["client_secret"],
    }, timeout=20)
    r.raise_for_status()
    data = r.json()
    # 예시: {"token":"...","token_type":"Bearer","expires_dt":"20250914125726"}
    CACHE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    return data


def get_token():
    if CACHE.exists():
        data = json.loads(CACHE.read_text(encoding="utf-8"))
        # 여유 있게 만료 5분 전 재발급
        exp = datetime.strptime(data.get("expires_dt"), "%Y%m%d%H%M%S")
        if datetime.now() < exp - timedelta(minutes=5):
            return data["token_type"] + " " + data["token"]
    data = issue_token()
    return data["token_type"] + " " + data["token"]

팁: 운영/모의 토큰을 분리해 두면 테스트가 편합니다. (예: kiwoom_token_mock.json)

---

3) 분봉 조회 — 연속조회(페이징)와 스로틀링이 핵심

분봉 API는 요청 건수 제한이 있고, 응답에 연속 조회 키가 따라옵니다. 이를 무시하면 중간 구간이 누락됩니다. 아래는 **헤더의 cont-yn/next-key**를 처리하는 안전한 루프 예시입니다.

# D:\stock\app\jobs\fetch_minute_symbol.py
import csv, time, requests
from datetime import datetime

BASE = "https://api.kiwoom.com"  # 운영 도메인 예시
API_ID = "ka10xxx"                # 분봉 조회용 TR 코드(문서 기준으로 교체)
UNIT = 1                           # 1=1분, 5=5분 등
DATE = "2025-09-10"               # 기준 일자(KST)
SYMBOL = "005930"                  # 예: 삼성전자


def fetch_all_minutes(symbol: str, unit: int, base_date: str):
    url = f"{BASE}/uapi/domestic-stock/v1/quotations/minute"  # 예시 경로; 실제 문서 기준으로 교체
    headers = {
        "Authorization": get_token(),            # 위 섹션에서 구현
        "Content-Type": "application/json",
        "api-id": API_ID,                        # TR 코드
    }
    payload = {
        "symbol": symbol,
        "unit": unit,
        "base_date": base_date.replace("-", ""),  # 20250910
        "count": 400,                            # 브로커 허용 범위 내 최대치
    }

    out = []
    next_key = None
    retry = 0
    while True:
        if next_key:
            payload["next_key"] = next_key
        r = requests.post(url, headers=headers, json=payload, timeout=20)
        if r.status_code >= 500 and retry < 3:
            retry += 1
            time.sleep(2 ** retry)  # 지수 백오프
            continue
        r.raise_for_status()
        data = r.json()
        rows = data.get("output", [])  # 실 응답 스키마에 맞춰 조정
        out.extend(rows)
        # 연속조회 헤더 처리 (예시 키 이름)
        cont = r.headers.get("cont-yn", "N")
        next_key = r.headers.get("next-key")
        if cont == "N" or not next_key:
            break
        time.sleep(0.3)  # 레이트 제한 보호
    return out


if __name__ == "__main__":
    rows = fetch_all_minutes(SYMBOL, UNIT, DATE)
    fn = fr"D:\stock\data\ohlcv_min_raw\KRX\{SYMBOL}_{DATE.replace('-', '')}_m{UNIT}.csv"
    with open(fn, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["dt","o","h","l","c","v"])  # 스키마 표준화
        for r in rows:
            # 실제 키 이름에 맞춰 매핑하세요
            w.writerow([r["datetime"], r["open"], r["high"], r["low"], r["close"], r["volume"]])
    print("[OK] rows=", len(rows), "->", fn)

중요: 엔드포인트 경로·응답 스키마·헤더 키 이름은 브로커 공식 문서를 기준으로 맞추세요. 위 코드는 안전한 패턴을 보여주는 템플릿입니다.

---

4) 유니버스 일괄 수집 — EOD 20:00 크론으로 굴리기

하루에 한 번, 장 마감 이후 전 유니버스를 훑어 저장하면 T+1 전략(갭 가드·오버나이트 필터)에 즉시 쓸 수 있습니다.

# D:\stock\ops\run_minute_eod.ps1
$py = "D:\stock\.venv64\Scripts\python.exe"
$base = (Get-Date -Format yyyy-MM-dd)
$syms = Get-Content "D:\stock\data\symbols_krx_active.txt"
foreach ($s in $syms) {
  & $py "D:\stock\app\jobs\fetch_minute_symbol.py" --symbol $s --unit 1 --base-date $base
  Start-Sleep -Milliseconds 400
}

휴일·반일장 처리

• app\utils\krx_calendar.py를 호출해 거래일만 수집
• 반일장은 타임 윈도우를 줄여 불필요한 재시도를 방지

---

5) 저장 스키마와 재현성 — 백테스트를 염두에 둔 표준화

• 분봉 스키마는 dt, o, h, l, c, v를 고정
• 파일명 규칙으로 심볼/일자/분해능 노출 → 빠른 탐색·재처리
• DB 사용 시 SQLite prices_min 테이블에 인덱스 (symbol, dt)

팁: “분석 데이터(C 레이어)”는 주문 엔진과 분리하세요. 실패 영역을 분리하면 운영이 편해집니다.

---

6) 트러블슈팅 (실패를 계획하세요)

증상 원인 해결

중간 구간 누락	cont-yn/next-key 미처리	헤더 기반 루프 구현, 마지막 페이지까지 수집
401/403 오류	토큰 만료/화이트리스트 미설정	토큰 갱신 시각 여유 5분 규칙, 고정 IP 등록
429/5xx 빈번	레이트 제한·간헐 장애	지수 백오프·쿨다운, 재시도 상한·서킷브레이커
시간 틀어짐	KST/UTC 혼선	저장 전 KST 기준 통일, 파일명·dt 모두 KST
파일 파편화	예외 종료·중복 실행	임시파일→원자적 rename, 중복 덮어쓰기 금지 옵션

---

7) 다음 단계 — 전략으로 연결하기

• T+1 스크리너: 전일 분봉의 마지막 60분 흐름 + 캔들 패턴 + 체결강도 → 다음날 감시리스트
• 갭 가드: “최대 허용 시가” 초과 시 진입 스킵, 대안은 당일 MOC/ATC
• 브래킷 주문: 체결 후 자동 손절·익절 지정 + ATR/PSAR 기반 트레일링

관련 글도 이어서 읽어보세요:

• 키움 REST API로 주식 주문 시스템 구현하기 (Python) — 브래킷·정정/취소 포함
• 한국 주식 API 생태계 완전 가이드 (2025-10 최신판) — 데이터·브로커 조합 맵

 

 

해시태그

#키움증권 #키움REST #분봉차트 #분봉데이터 #한국주식API #연속조회 #nextkey #자동매매 #pykrx #KRX #TechOracle