shuim/auto-stock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

배치 수정

Browse Source
This commit is contained in:
shuim
2026-03-06 14:12:54 +09:00
parent 6d39069d3f
commit ac98bb856f
8 changed files with 321 additions and 351 deletions
Download Patch File Download Diff File Expand all files Collapse all files

279
backend/collect_krx_data.py
View File

@@ -1,11 +1,10 @@
"""
KRX 히스토리컬 데이터 수집 스크립트
공공데이터포털 금융위원회 주식시세 API 기반 일봉 데이터 수집 스크립트
KRX OpenAPI (data-dbg.krx.co.kr)를 사용하여 KOSPI/KOSDAQ 종목의
일봉 OHLCV 데이터를 DB에 저장합니다.
- KIS API 의존성 완전 제거
- 날짜 1건 호출 → 해당일 전 종목 OHLCV 일괄 수신 (효율적)
API: GetStockSecuritiesInfoService (apis.data.go.kr)
- API 키 필요 (data.go.kr 에서 무료 신청)
- KOSPI / KOSDAQ 전종목 일별 OHLCV 제공
- KRX / pykrx / FinanceDataReader 의존성 없음
사용법:
python collect_krx_data.py # 전체 수집 (2020-01-01~)
@@ -18,13 +17,12 @@ KRX OpenAPI (data-dbg.krx.co.kr)를 사용하여 KOSPI/KOSDAQ 종목의
import argparse
import asyncio
import logging
import os
import sys
import time
from datetime import date, datetime, timedelta
from pathlib import Path
import aiohttp
import httpx
import pandas as pd
# 프로젝트 루트를 Python 경로에 추가
@@ -33,31 +31,26 @@ 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
# ─── KRX OpenAPI 설정 ─────────────────────────────────────────────
# 실제 데이터 엔드포인트 (openapi.krx.co.kr 는 포털, 데이터는 data-dbg)
KRX_DATA_BASE = "https://data-dbg.krx.co.kr/svc/apis"
try:
from app.core.config import settings as _cfg
KRX_API_KEY: str = _cfg.KRX_API_KEY
except Exception:
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent.parent / ".env")
KRX_API_KEY = os.environ.get("KRX_API_KEY", "")
# ─── ETF/특수 종목 제외 키워드 ────────────────────────────────────
# ETF/특수 종목 제외 키워드
SKIP_KEYWORDS = [
"ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG",
"KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX",
"선물", "스팩", "리츠", "SPAC",
]
# 날짜별 동시 요청 수 (KRX API 쓰로틀링 고려)
COLLECT_CONCURRENCY = 3
REQUEST_INTERVAL = 0.5 # 요청 간 최소 대기(초)
# 공공데이터포털 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,
@@ -74,130 +67,108 @@ def generate_trading_dates(start: str, end: str) -> list[str]:
e = datetime.strptime(end, "%Y%m%d").date()
dates: list[str] = []
while d <= e:
if d.weekday() < 5: # 0=Mon … 4=Fri
if d.weekday() < 5:
dates.append(d.strftime("%Y%m%d"))
d += timedelta(days=1)
return dates
def _int(val) -> int:
"""문자열/숫자를 int로 안전 변환 (쉼표 포함 처리)."""
try:
return int(str(val).replace(",", "").strip() or "0")
except (ValueError, TypeError):
return 0
# ─── 공공데이터포털 API 수집 ───────────────────────────────────────
async def fetch_daily_prices(market: str, bas_dd: str) -> pd.DataFrame:
"""공공데이터포털 금융위원회 주식시세 API로 전종목 OHLCV 조회.
def _float(val) -> float:
"""문자열/숫자를 float로 안전 변환."""
try:
return float(str(val).replace(",", "").strip() or "0")
except (ValueError, TypeError):
return 0.0
# ─── KRX API 호출 ─────────────────────────────────────────────────
async def _fetch_krx(
session: aiohttp.ClientSession,
endpoint: str,
bas_dd: str,
) -> list:
"""KRX OpenAPI 단건 호출 → OutBlock_1 리스트 반환.
endpoint 예: "sto/stk_bydd_trd"
market: 'kospi' | 'kosdaq'
bas_dd: 'YYYYMMDD'
반환: DataFrame (stock_code, stock_name, date, open, high, low, close, volume, change_rate)
"""
url = f"{KRX_DATA_BASE}/{endpoint}"
try:
async with session.get(
url,
params={"basDd": bas_dd},
headers={"AUTH_KEY": KRX_API_KEY},
timeout=aiohttp.ClientTimeout(total=30),
) as resp:
if resp.status != 200:
logger.warning("KRX %s %s → HTTP %s", endpoint, bas_dd, resp.status)
return []
data = await resp.json(content_type=None)
return data.get("OutBlock_1") or []
except Exception as exc:
logger.error("KRX API 오류 [%s %s]: %s", endpoint, bas_dd, exc)
return []
async def fetch_daily_prices(
session: aiohttp.ClientSession,
market: str,
bas_dd: str,
) -> pd.DataFrame:
"""특정 날짜의 전 종목 OHLCV 조회.
KRX 응답 필드:
ISU_SRT_CD : 단축 종목코드 (e.g. "005930")
ISU_ABBRV : 종목 약칭
TDD_OPNPRC : 시가
TDD_HGPRC : 고가
TDD_LWPRC : 저가
TDD_CLSPRC : 종가
ACC_TRDVOL : 누적 거래량
FLUC_RT : 등락률 (%)
"""
endpoint = "sto/stk_bydd_trd" if market == "kospi" else "sto/ksq_bydd_trd"
items = await _fetch_krx(session, endpoint, bas_dd)
if not items:
return pd.DataFrame()
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
for item in items:
name = (item.get("ISU_ABBRV") or item.get("ISU_NM") or "").strip()
if any(kw in name for kw in SKIP_KEYWORDS):
continue
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
code = item.get("ISU_SRT_CD", "").strip()
if not code:
continue
body = data.get("response", {}).get("body", {})
total_count = int(body.get("totalCount", 0))
items_wrap = body.get("items", {})
volume = _int(item.get("ACC_TRDVOL"))
if volume == 0:
continue # 거래정지일 제외
# 데이터 없음 (휴장일)
if not items_wrap:
break
close = _int(item.get("TDD_CLSPRC"))
if close == 0:
continue
items = items_wrap.get("item", [])
if not isinstance(items, list):
items = [items] # 1건이면 dict로 올 수 있음
rows.append({
"stock_code": code,
"stock_name": name,
"date": trade_date,
"open": _int(item.get("TDD_OPNPRC")),
"high": _int(item.get("TDD_HGPRC")),
"low": _int(item.get("TDD_LWPRC")),
"close": close,
"volume": volume,
"change_rate": _float(item.get("FLUC_RT")) / 100.0,
})
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()
async def get_stock_codes(
session: aiohttp.ClientSession,
market: str,
bas_dd: str,
) -> set[str]:
"""특정 날짜 기준 상장 종목 코드 집합 조회 (선택적 필터용)."""
endpoint = "sto/stk_isu_base_info" if market == "kospi" else "sto/ksq_isu_base_info"
items = await _fetch_krx(session, endpoint, bas_dd)
codes: set[str] = set()
for item in items:
code = item.get("ISU_SRT_CD", "").strip()
if code:
codes.add(code)
return codes
# ─── DB 헬퍼 ──────────────────────────────────────────────────────
async def create_tables():
@@ -207,7 +178,6 @@ async def create_tables():
async def get_last_trade_date(session) -> date | None:
"""DB 내 마지막 거래일 조회 (전 종목 공통)."""
result = await session.execute(
text("SELECT MAX(trade_date) FROM stock_daily_prices")
)
@@ -234,13 +204,13 @@ async def upsert_daily_prices(session, df: pd.DataFrame) -> int:
"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": 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]
batch = records[i: i + batch_size]
stmt = pg_insert(StockDailyPrice).values(batch)
stmt = stmt.on_conflict_do_update(
constraint="uq_stock_daily",
@@ -270,7 +240,6 @@ async def collect_date(
bas_dd: str,
market: str,
semaphore: asyncio.Semaphore,
http_session: aiohttp.ClientSession,
) -> tuple[bool, int]:
"""단일 날짜 수집 (세마포어 적용)."""
async with semaphore:
@@ -283,17 +252,16 @@ async def collect_date(
frames: list[pd.DataFrame] = []
for mkt in markets:
df = await fetch_daily_prices(http_session, mkt, bas_dd)
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 # 휴장일은 성공으로 처리
return True, 0
combined = pd.concat(frames, ignore_index=True)
# 같은 날짜 내 코드 중복 제거 (KOSPI/KOSDAQ 겹침 방지)
combined = combined.drop_duplicates(subset=["stock_code", "date"])
async with AsyncSessionLocal() as db:
@@ -315,16 +283,7 @@ async def collect_all(
start_date: str = "20200101",
update_mode: bool = False,
):
"""메인 수집 로직 (KRX OpenAPI 기반, 날짜별 전종목 일괄 수집)."""
if not KRX_API_KEY:
logger.error(
"KRX_API_KEY가 설정되지 않았습니다.\n"
" 1) https://openapi.krx.co.kr 에서 API 키 발급\n"
" 2) .env 파일에 KRX_API_KEY=<발급받은키> 추가\n"
" 3) 다시 실행하세요."
)
sys.exit(1)
"""메인 수집 로직 (공공데이터포털 금융위원회 주식시세 API)."""
await create_tables()
end_date = datetime.now().strftime("%Y%m%d")
@@ -346,27 +305,25 @@ async def collect_all(
logger.warning("수집할 날짜가 없습니다.")
return
# --limit N → 최근 N거래일만 (테스트용)
if limit > 0:
dates = dates[-limit:]
total = len(dates)
logger.info("수집 날짜: %d일 (%s ~ %s)", total, dates[0], dates[-1])
semaphore = asyncio.Semaphore(COLLECT_CONCURRENCY)
conn = aiohttp.TCPConnector(limit=10)
# 공공데이터 API: 동시 3 요청까지 허용
semaphore = asyncio.Semaphore(3)
async with aiohttp.ClientSession(connector=conn) as http_session:
results = await asyncio.gather(
*[
collect_date(i, total, d, market, semaphore, http_session)
for i, d in enumerate(dates, 1)
],
return_exceptions=True,
)
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
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)
@@ -378,7 +335,9 @@ async def collect_all(
def main():
parser = argparse.ArgumentParser(description="KRX 히스토리컬 데이터 수집 (KRX OpenAPI)")
parser = argparse.ArgumentParser(
description="일봉 데이터 수집 (공공데이터포털 금융위원회 주식시세 API)"
)
parser.add_argument(
"--market", choices=["kospi", "kosdaq", "all"], default="all",
help="수집 시장 (기본: all)",
@@ -397,7 +356,7 @@ def main():
)
args = parser.parse_args()
logger.info("KRX 데이터 수집 시작 (KRX OpenAPI)")
logger.info("일봉 데이터 수집 시작 (공공데이터포털)")
logger.info(" 시장: %s", args.market)
logger.info(" 날짜 제한: %s", "전체" if args.limit == 0 else f"최근 {args.limit}거래일")
logger.info(" 시작일: %s", args.start_date)