yfinance로 국내 주식 데이터 받기: 수정주가·결측·티커를 먼저 확인하는 법

python -m pip install yfinance pandas pyarrow

#!/usr/bin/env python3
"""yfinance로 국내 주식 일봉을 받고 기본 품질 검사를 거쳐 Parquet로 저장한다."""

from __future__ import annotations

import argparse
from pathlib import Path

import pandas as pd
import yfinance as yf


REQUIRED_COLUMNS = ("Open", "High", "Low", "Close", "Adj Close", "Volume")


def normalize_columns(price: pd.DataFrame) -> pd.DataFrame:
    if not isinstance(price.columns, pd.MultiIndex):
        return price
    if price.columns.nlevels != 2:
        raise ValueError("예상하지 못한 다중 열 구조입니다.")
    first_level = price.columns.get_level_values(0)
    if set(REQUIRED_COLUMNS).issubset(first_level):
        price.columns = first_level
        return price
    second_level = price.columns.get_level_values(1)
    if set(REQUIRED_COLUMNS).issubset(second_level):
        price.columns = second_level
        return price
    raise ValueError("OHLCV 열을 찾지 못했습니다.")


def build_ticker(code: str, market: str) -> str:
    clean_code = code.strip().zfill(6)
    suffix_by_market = {"KOSPI": ".KS", "KOSDAQ": ".KQ"}
    normalized_market = market.strip().upper()
    if normalized_market not in suffix_by_market:
        raise ValueError("시장 구분은 KOSPI 또는 KOSDAQ만 사용할 수 있습니다.")
    if not clean_code.isdigit() or len(clean_code) != 6:
        raise ValueError("종목 코드는 여섯 자리 숫자로 입력해야 합니다.")
    return f"{clean_code}{suffix_by_market[normalized_market]}"


def download_daily_price(ticker: str, period: str) -> pd.DataFrame:
    price = yf.download(
        tickers=ticker,
        period=period,
        interval="1d",
        auto_adjust=False,
        actions=True,
        group_by="column",
        multi_level_index=False,
        progress=False,
        threads=False,
        timeout=20,
    )
    if price.empty:
        raise RuntimeError(f"가격 데이터를 받지 못했습니다: {ticker}")
    price = normalize_columns(price).copy()
    missing_columns = [column for column in REQUIRED_COLUMNS if column not in price.columns]
    if missing_columns:
        raise ValueError(f"필수 열이 없습니다: {', '.join(missing_columns)}")
    price.index = pd.to_datetime(price.index).tz_localize(None)
    price.index.name = "trade_date"
    price = price.loc[:, [*REQUIRED_COLUMNS, "Dividends", "Stock Splits"]].copy()
    price["Dividends"] = price["Dividends"].fillna(0.0)
    price["Stock Splits"] = price["Stock Splits"].fillna(0.0)
    price["adjustment_factor"] = price["Adj Close"].div(price["Close"]).where(price["Close"].ne(0))
    return price


def validate_daily_price(price: pd.DataFrame) -> dict[str, int]:
    missing_core = int(price.loc[:, REQUIRED_COLUMNS].isna().any(axis=1).sum())
    duplicated_dates = int(price.index.duplicated().sum())
    invalid_ohlc_mask = (price["Low"] > price[["Open", "Close", "High"]].min(axis=1)) | (price["High"] < price[["Open", "Close", "Low"]].max(axis=1))
    invalid_ohlc = int(invalid_ohlc_mask.sum())
    negative_volume = int((price["Volume"] < 0).sum())
    report = {
        "rows": len(price),
        "missing_core_rows": missing_core,
        "duplicated_dates": duplicated_dates,
        "invalid_ohlc_rows": invalid_ohlc,
        "negative_volume_rows": negative_volume,
    }
    if any(report[key] for key in ("missing_core_rows", "duplicated_dates", "invalid_ohlc_rows", "negative_volume_rows")):
        invalid_dates = ", ".join(price.index[invalid_ohlc_mask].strftime("%Y-%m-%d").tolist()[:5])
        raise ValueError(f"품질 검사 실패: {report}; OHLC 확인 거래일: {invalid_dates or '-'}")
    return report


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--code", default="035720", help="여섯 자리 종목 코드")
    parser.add_argument("--market", choices=("KOSPI", "KOSDAQ"), default="KOSPI")
    parser.add_argument("--period", default="5y", help="yfinance 기간 문자열")
    parser.add_argument("--output", type=Path, default=Path("krx_daily_price.parquet"))
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    ticker = build_ticker(args.code, args.market)
    price = download_daily_price(ticker, args.period)
    report = validate_daily_price(price)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    price.to_parquet(args.output)
    print(f"ticker: {ticker}")
    print(f"rows: {report['rows']:,}")
    print(f"date_range: {price.index.min().date()} ~ {price.index.max().date()}")
    print(f"quality: missing={report['missing_core_rows']}, duplicate={report['duplicated_dates']}, ohlc={report['invalid_ohlc_rows']}, volume={report['negative_volume_rows']}")
    print(f"saved: {args.output}")


if __name__ == "__main__":
    main()

python yfinance_krx_daily_data.py --code 005930 --market KOSPI --period 5y