키움증권 REST API로 주식 DB 만들기: Python·SQLite 일봉 수집

export KIWOOM_APP_KEY="발급받은_앱키"
export KIWOOM_SECRET_KEY="발급받은_시크릿키"
pip install requests

$env:KIWOOM_APP_KEY="발급받은_앱키"
$env:KIWOOM_SECRET_KEY="발급받은_시크릿키"
pip install requests

CREATE TABLE IF NOT EXISTS daily_price (
    code        TEXT    NOT NULL,
    trade_date  TEXT    NOT NULL,
    open        INTEGER NOT NULL,
    high        INTEGER NOT NULL,
    low         INTEGER NOT NULL,
    close       INTEGER NOT NULL,
    volume      INTEGER NOT NULL,
    trade_value INTEGER NOT NULL,
    PRIMARY KEY (code, trade_date)
) WITHOUT ROWID;

import os
import time
import requests

BASE_URL = "https://api.kiwoom.com"


def issue_token():
    response = requests.post(
        BASE_URL + "/oauth2/token",
        json={
            "grant_type": "client_credentials",
            "appkey": os.environ["KIWOOM_APP_KEY"],
            "secretkey": os.environ["KIWOOM_SECRET_KEY"],
        },
        timeout=15,
    )
    response.raise_for_status()
    return response.json()["token"]


def fetch_daily(token, code, base_date, max_pages=10):
    rows = []
    cont_yn, next_key = "N", ""

    for page_no in range(max_pages):
        response = requests.post(
            BASE_URL + "/api/dostk/chart",
            headers={
                "Content-Type": "application/json;charset=UTF-8",
                "authorization": f"Bearer {token}",
                "cont-yn": cont_yn,
                "next-key": next_key,
                "api-id": "ka10081",
            },
            json={
                "stk_cd": code,
                "base_dt": base_date,
                "upd_stkpc_tp": "1",
            },
            timeout=15,
        )
        response.raise_for_status()
        payload = response.json()
        if int(payload.get("return_code", -1)) != 0:
            raise RuntimeError(payload.get("return_msg", payload))

        rows.extend(payload.get("stk_dt_pole_chart_qry", []))
        cont_yn = response.headers.get("cont-yn", "N").upper()
        next_key = response.headers.get("next-key", "")
        if cont_yn != "Y" or not next_key:
            break
        time.sleep(0.22)

    return rows

import sqlite3


def to_record(code, row):
    to_int = lambda value: abs(int(str(value or "0").replace(",", "")))
    return (
        code,
        row["dt"],
        to_int(row["open_pric"]),
        to_int(row["high_pric"]),
        to_int(row["low_pric"]),
        to_int(row["cur_prc"]),
        to_int(row["trde_qty"]),
        to_int(row["trde_prica"]),
    )


UPSERT_SQL = """
INSERT INTO daily_price
    (code, trade_date, open, high, low, close, volume, trade_value)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(code, trade_date) DO UPDATE SET
    open=excluded.open, high=excluded.high, low=excluded.low,
    close=excluded.close, volume=excluded.volume,
    trade_value=excluded.trade_value
"""

with sqlite3.connect("data/kiwoom_daily.db") as connection:
    connection.executemany(
        UPSERT_SQL,
        [to_record("005930", row) for row in daily_rows],
    )

-- 종목별 행 수와 날짜 범위
SELECT code, COUNT(*), MIN(trade_date), MAX(trade_date)
FROM daily_price
GROUP BY code;

-- 고가와 저가가 OHLC 범위를 벗어난 행
SELECT * FROM daily_price
WHERE high < MAX(open, close, low)
   OR low > MIN(open, close, high);

-- 음수 거래량
SELECT * FROM daily_price WHERE volume < 0;