378 lines
13 KiB
Python
378 lines
13 KiB
Python
"""
|
|
공공데이터포털 금융위원회 주식시세 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()
|