코스피200 데이터 수집하기

import argparse
from datetime import date, timedelta
from pathlib import Path
import time

import pandas as pd

from src.kis_readonly import ROOT, ReadOnlyKis, load_config

FIELDS = {
    "stck_bsop_date": "trade_date",
    "stck_oprc": "open",
    "stck_hgpr": "high",
    "stck_lwpr": "low",
    "stck_clpr": "close",
    "acml_vol": "volume",
}
PRICE_COLUMNS = ["open", "high", "low", "close"]
COLUMNS = ["symbol", "trade_date", *PRICE_COLUMNS, "volume"]


def normalize(symbol, records):
    if not records:
        raise ValueError("No daily records returned")
    frame = pd.DataFrame(records)
    missing = set(FIELDS) - set(frame.columns)
    if missing:
        raise ValueError(f"Missing daily fields: {sorted(missing)}")
    frame = frame.loc[frame["stck_bsop_date"].fillna("").ne(""), list(FIELDS)]
    frame = frame.rename(columns=FIELDS).copy()
    if frame.empty:
        raise ValueError("All returned dates are empty")
    frame["trade_date"] = pd.to_datetime(
        frame["trade_date"], format="%Y%m%d", errors="raise"
    ).dt.date
    for column in [*PRICE_COLUMNS, "volume"]:
        values = pd.to_numeric(frame[column], errors="raise")
        if values.isna().any() or values.mod(1).ne(0).any():
            raise ValueError(f"Non-integer value in {column}")
        frame[column] = values.astype("int64")
    invalid = (
        frame[PRICE_COLUMNS].le(0).any(axis=1)
        | frame["volume"].lt(0)
        | frame["high"].lt(frame[["open", "close", "low"]].max(axis=1))
        | frame["low"].gt(frame[["open", "close", "high"]].min(axis=1))
    )
    if invalid.any():
        raise ValueError(f"Invalid OHLCV rows: {int(invalid.sum())}")
    frame["symbol"] = symbol
    frame = frame.drop_duplicates()
    if frame.duplicated(["symbol", "trade_date"]).any():
        raise ValueError("Conflicting rows for the same symbol and date")
    return frame[COLUMNS].sort_values(["symbol", "trade_date"]).reset_index(drop=True)


def collect_symbol(client, symbol, start, end):
    records = []
    cursor = start
    while cursor <= end:
        window_end = min(cursor + timedelta(days=79), end)
        body = client.quotations("daily", {
            "FID_COND_MRKT_DIV_CODE": "J",
            "FID_INPUT_ISCD": symbol,
            "FID_INPUT_DATE_1": cursor.strftime("%Y%m%d"),
            "FID_INPUT_DATE_2": window_end.strftime("%Y%m%d"),
            "FID_PERIOD_DIV_CODE": "D",
            "FID_ORG_ADJ_PRC": "0",
        })
        output = body.get("output2")
        if not isinstance(output, list):
            raise ValueError("output2 must be a list")
        records.extend(output)
        cursor = window_end + timedelta(days=1)
        time.sleep(1.0)
    frame = normalize(symbol, records)
    if frame["trade_date"].lt(start).any() or frame["trade_date"].gt(end).any():
        raise ValueError("The response contains a date outside the requested range")
    return frame


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--start", type=date.fromisoformat, required=True)
    parser.add_argument("--end", type=date.fromisoformat, required=True)
    parser.add_argument("--limit", type=int, default=1)
    args = parser.parse_args()
    if args.start > args.end or args.limit < 1:
        parser.error("Check the date range and limit")
    universe = pd.read_csv(
        ROOT / "outputs/universe/kospi200.csv", dtype={"symbol": "string"}
    )
    symbols = universe["symbol"]
    if symbols.isna().any() or not symbols.str.fullmatch(r"[A-Z0-9]{6}").all():
        raise ValueError("Universe contains invalid symbols")
    if symbols.duplicated().any():
        raise ValueError("Universe contains duplicate symbols")
    target = ROOT / "outputs/daily"
    target.mkdir(parents=True, exist_ok=True)
    manifest = []
    client = ReadOnlyKis(load_config())
    try:
        for symbol in symbols.head(args.limit):
            try:
                frame = collect_symbol(client, symbol, args.start, args.end)
                name = f"{symbol}_{args.start}_{args.end}.csv"
                frame.to_csv(target / name, index=False, encoding="utf-8-sig")
                manifest.append({"symbol": symbol, "status": "ok", "rows": len(frame),
                                 "first": frame.trade_date.min(), "last": frame.trade_date.max()})
            except Exception as exc:
                manifest.append({"symbol": symbol, "status": "failed", "rows": 0,
                                 "error_type": type(exc).__name__})
    finally:
        client.close()
    summary = pd.DataFrame(manifest)
    summary.to_csv(target / "collection_manifest.csv", index=False, encoding="utf-8-sig")
    print(summary.to_string(index=False))
    if summary["status"].ne("ok").any():
        raise SystemExit(1)


if __name__ == "__main__":
    main()

.\.venv\Scripts\python.exe -m py_compile src/collect_daily.py
.\.venv\Scripts\python.exe -m src.collect_daily --start 2026-01-01 --end 2026-10-16 --limit 1