""" 공공데이터포털 금융위원회 주식시세 API 기반 일봉 데이터 수집 스크립트 API: GetStockSecuritiesInfoService (apis.data.go.kr) - API 키 필요 (data.go.kr 에서 무료 신청) - KOSPI / KOSDAQ 전종목 일별 OHLCV 제공 - KRX / pykrx / FinanceDataReader 의존성 없음 사용법: python collect_krx_data.py # 전체 수집 (2020-01-01~) python collect_krx_data.py --limit 5 # 최근 5거래일 (테스트) python collect_krx_data.py --start-date 20240101 # 시작일 지정 python collect_krx_data.py --update # 증분 업데이트 python collect_krx_data.py --market kospi # KOSPI만 """ import argparse import asyncio import logging import sys import time from datetime import date, datetime, timedelta from pathlib import Path import httpx import pandas as pd # 프로젝트 루트를 Python 경로에 추가 sys.path.insert(0, str(Path(__file__).parent)) from sqlalchemy import text from sqlalchemy.dialects.postgresql import insert as pg_insert from app.core.config import settings from app.core.database import AsyncSessionLocal, engine, Base from app.models.models import StockDailyPrice # ETF/특수 종목 제외 키워드 SKIP_KEYWORDS = [ "ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG", "KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", "선물", "스팩", "리츠", "SPAC", ] # 공공데이터포털 API 설정 DATAGOKR_BASE_URL = ( "https://apis.data.go.kr/1160100/service" "/GetStockSecuritiesInfoService/getStockPriceInfo" ) ROWS_PER_PAGE = 1000 # API 최대 허용값 # 날짜별 요청 간격 (초) REQUEST_INTERVAL = 0.2 logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) logger = logging.getLogger(__name__) # ─── 유틸 ───────────────────────────────────────────────────────── def generate_trading_dates(start: str, end: str) -> list[str]: """start ~ end 사이 평일(월~금) 날짜 문자열 목록 생성 (YYYYMMDD).""" d = datetime.strptime(start, "%Y%m%d").date() e = datetime.strptime(end, "%Y%m%d").date() dates: list[str] = [] while d <= e: if d.weekday() < 5: dates.append(d.strftime("%Y%m%d")) d += timedelta(days=1) return dates # ─── 공공데이터포털 API 수집 ─────────────────────────────────────── async def fetch_daily_prices(market: str, bas_dd: str) -> pd.DataFrame: """공공데이터포털 금융위원회 주식시세 API로 전종목 OHLCV 조회. market: 'kospi' | 'kosdaq' bas_dd: 'YYYYMMDD' 반환: DataFrame (stock_code, stock_name, date, open, high, low, close, volume, change_rate) """ mrkt_cls = "KOSPI" if market == "kospi" else "KOSDAQ" api_key = settings.DATAGOKR_API_KEY if not api_key: raise RuntimeError( "DATAGOKR_API_KEY 가 설정되지 않았습니다. " ".env 에 DATAGOKR_API_KEY=<발급키> 를 추가하세요." ) trade_date = datetime.strptime(bas_dd, "%Y%m%d").date() rows: list[dict] = [] page = 1 async with httpx.AsyncClient(timeout=30.0) as client: while True: params = { "serviceKey": api_key, "numOfRows": str(ROWS_PER_PAGE), "pageNo": str(page), "resultType": "json", "basDt": bas_dd, "mrktCls": mrkt_cls, } try: resp = await client.get(DATAGOKR_BASE_URL, params=params) resp.raise_for_status() data = resp.json() except Exception as exc: logger.error("API 요청 실패 [%s %s page%d]: %s", market, bas_dd, page, exc) break body = data.get("response", {}).get("body", {}) total_count = int(body.get("totalCount", 0)) items_wrap = body.get("items", {}) # 데이터 없음 (휴장일) if not items_wrap: break items = items_wrap.get("item", []) if not isinstance(items, list): items = [items] # 1건이면 dict로 올 수 있음 for item in items: name = item.get("itmsNm", "") if any(kw in name for kw in SKIP_KEYWORDS): continue volume = int(item.get("trqu", 0) or 0) if volume == 0: continue # 거래 정지 close = int(item.get("clpr", 0) or 0) if close == 0: continue try: change_rate = float(item.get("fltRt", 0) or 0) / 100.0 except (TypeError, ValueError): change_rate = 0.0 # srtnCd: 단축코드(6자리), 앞에 공백 있을 수 있어 strip stock_code = str(item.get("srtnCd", "")).strip().zfill(6) rows.append({ "stock_code": stock_code, "stock_name": name, "date": trade_date, "open": int(item.get("mkp", 0) or 0), "high": int(item.get("hipr", 0) or 0), "low": int(item.get("lopr", 0) or 0), "close": close, "volume": volume, "change_rate": change_rate, "market_cap": int(item.get("mrktTotAmt", 0) or 0) or None, }) # 페이지네이션 종료 조건 fetched_so_far = (page - 1) * ROWS_PER_PAGE + len(items) if fetched_so_far >= total_count or len(items) < ROWS_PER_PAGE: break page += 1 await asyncio.sleep(REQUEST_INTERVAL) return pd.DataFrame(rows) if rows else pd.DataFrame() # ─── DB 헬퍼 ────────────────────────────────────────────────────── async def create_tables(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("DB 테이블 확인/생성 완료") async def get_last_trade_date(session) -> date | None: result = await session.execute( text("SELECT MAX(trade_date) FROM stock_daily_prices") ) return result.scalar() async def upsert_daily_prices(session, df: pd.DataFrame) -> int: """DB에 일봉 데이터 UPSERT (배치 500건).""" if df.empty: return 0 records = [] for _, row in df.iterrows(): trade_date = row["date"] if isinstance(trade_date, pd.Timestamp): trade_date = trade_date.date() records.append({ "stock_code": row["stock_code"], "stock_name": row["stock_name"], "trade_date": trade_date, "open_price": int(row["open"]), "high_price": int(row["high"]), "low_price": int(row["low"]), "close_price": int(row["close"]), "volume": int(row["volume"]), "change_rate": float(row["change_rate"]) if pd.notna(row["change_rate"]) else None, "market_cap": int(row["market_cap"]) if pd.notna(row.get("market_cap")) and row.get("market_cap") else None, }) batch_size = 500 inserted = 0 for i in range(0, len(records), batch_size): batch = records[i: i + batch_size] stmt = pg_insert(StockDailyPrice).values(batch) stmt = stmt.on_conflict_do_update( constraint="uq_stock_daily", set_={ "stock_name": stmt.excluded.stock_name, "open_price": stmt.excluded.open_price, "high_price": stmt.excluded.high_price, "low_price": stmt.excluded.low_price, "close_price": stmt.excluded.close_price, "volume": stmt.excluded.volume, "change_rate": stmt.excluded.change_rate, "market_cap": stmt.excluded.market_cap, }, ) await session.execute(stmt) inserted += len(batch) await session.commit() return inserted # ─── 날짜별 수집 ────────────────────────────────────────────────── async def collect_date( i: int, total: int, bas_dd: str, market: str, semaphore: asyncio.Semaphore, ) -> tuple[bool, int]: """단일 날짜 수집 (세마포어 적용).""" async with semaphore: try: markets = [] if market in ("kospi", "all"): markets.append("kospi") if market in ("kosdaq", "all"): markets.append("kosdaq") frames: list[pd.DataFrame] = [] for mkt in markets: df = await fetch_daily_prices(mkt, bas_dd) if not df.empty: frames.append(df) await asyncio.sleep(REQUEST_INTERVAL) if not frames: logger.info("[%d/%d] %s: 데이터 없음 (휴장일?)", i, total, bas_dd) return True, 0 combined = pd.concat(frames, ignore_index=True) combined = combined.drop_duplicates(subset=["stock_code", "date"]) async with AsyncSessionLocal() as db: count = await upsert_daily_prices(db, combined) logger.info("[%d/%d] %s: %d종목 저장", i, total, bas_dd, count) return True, count except Exception as exc: logger.error("[%d/%d] %s 실패: %s", i, total, bas_dd, exc) return False, 0 # ─── 메인 수집 로직 ─────────────────────────────────────────────── async def collect_all( market: str = "all", limit: int = 0, start_date: str = "20200101", update_mode: bool = False, ): """메인 수집 로직 (공공데이터포털 금융위원회 주식시세 API).""" await create_tables() end_date = datetime.now().strftime("%Y%m%d") actual_start = start_date if update_mode: async with AsyncSessionLocal() as db: last_date = await get_last_trade_date(db) if last_date: next_day = last_date + timedelta(days=1) actual_start = next_day.strftime("%Y%m%d") if next_day >= date.today(): logger.info("이미 최신 데이터입니다 (마지막: %s)", last_date) return logger.info("증분 업데이트: %s ~", actual_start) dates = generate_trading_dates(actual_start, end_date) if not dates: logger.warning("수집할 날짜가 없습니다.") return if limit > 0: dates = dates[-limit:] total = len(dates) logger.info("수집 날짜: %d일 (%s ~ %s)", total, dates[0], dates[-1]) # 공공데이터 API: 동시 3 요청까지 허용 semaphore = asyncio.Semaphore(3) results = await asyncio.gather( *[ collect_date(i, total, d, market, semaphore) for i, d in enumerate(dates, 1) ], return_exceptions=True, ) success = sum(1 for r in results if isinstance(r, tuple) and r[0]) fail = total - success total_rows = sum(r[1] for r in results if isinstance(r, tuple)) logger.info("=" * 60) logger.info("수집 완료!") logger.info(" 성공: %d/%d일", success, total) logger.info(" 실패: %d/%d일", fail, total) logger.info(" 총 저장 행: %s", f"{total_rows:,}") logger.info("=" * 60) def main(): parser = argparse.ArgumentParser( description="일봉 데이터 수집 (공공데이터포털 금융위원회 주식시세 API)" ) parser.add_argument( "--market", choices=["kospi", "kosdaq", "all"], default="all", help="수집 시장 (기본: all)", ) parser.add_argument( "--limit", type=int, default=0, help="최근 N거래일만 수집 (0=전체, 테스트용)", ) parser.add_argument( "--start-date", default="20200101", help="수집 시작일 YYYYMMDD (기본: 20200101)", ) parser.add_argument( "--update", action="store_true", help="증분 업데이트 모드 (마지막 저장일 이후만 수집)", ) args = parser.parse_args() logger.info("일봉 데이터 수집 시작 (공공데이터포털)") logger.info(" 시장: %s", args.market) logger.info(" 날짜 제한: %s", "전체" if args.limit == 0 else f"최근 {args.limit}거래일") logger.info(" 시작일: %s", args.start_date) logger.info(" 증분 모드: %s", args.update) start_time = time.time() asyncio.run(collect_all( market=args.market, limit=args.limit, start_date=args.start_date, update_mode=args.update, )) elapsed = time.time() - start_time logger.info("총 소요시간: %.1f초 (%.1f분)", elapsed, elapsed / 60) if __name__ == "__main__": main()