52733df235
pykrx/KRX 비정상 응답에서 traceback 포매팅 중 UnicodeDecodeError 가 발생해 API 500 을 만든 회귀 사례가 있었음. 새 pykrx fundamentals 경로에 동일 패턴이 다시 들어가지 않도록 %r 안전 직렬화로 통일.
123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
"""기업 펀더멘털 (시가총액 / PER / PBR / EPS / BPS / 배당수익률) — pykrx 기반.
|
|
|
|
KRX 가 일 단위로 산정한 멀티플 + 시총 스냅샷을 가져온다. pykrx 호출은 KRX 페이지 스크래핑
|
|
이라 종목당 1~2초가 든다. 같은 (code, snapshot_date) 는 인메모리 캐시로 중복 호출을 막는다.
|
|
|
|
API 호출이 자주 들어와도 같은 날 같은 종목은 한 번만 KRX 를 때린다. 컨테이너 재기동 시
|
|
캐시는 비어 다시 채워진다 (디스크 캐시까지는 안 만든다 — KRX 는 인증/쿼터가 없어서
|
|
재발급 비용이 없음).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
from dataclasses import dataclass
|
|
from datetime import date, timedelta
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 한국 시장 마지막 영업일 후보 — 토/일이면 금요일로, 공휴일은 KRX 가 빈 응답을 줘서
|
|
# 한 단계씩 거꾸로 시도 (최대 7일). 7일 모두 비면 fundamentals 데이터 없음.
|
|
_MAX_BACKOFF_DAYS = 7
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Fundamentals:
|
|
code: str
|
|
snapshot_date: date
|
|
market_cap: float | None # 원
|
|
shares_outstanding: int | None # 상장주식수
|
|
per: float | None
|
|
pbr: float | None
|
|
eps: float | None
|
|
bps: float | None
|
|
div: float | None # 배당금 (원)
|
|
dvd_yield: float | None # 배당수익률 (%)
|
|
|
|
|
|
# (code, today) → Fundamentals. None 도 캐시해서 KRX 없는 종목 반복 호출 방지.
|
|
_cache: dict[tuple[str, date], Fundamentals | None] = {}
|
|
_cache_lock = threading.Lock()
|
|
|
|
|
|
def _to_pykrx(d: date) -> str:
|
|
return d.strftime("%Y%m%d")
|
|
|
|
|
|
def _fetch_one(code: str, target: date) -> Fundamentals | None:
|
|
"""target 일 기준 KRX fundamentals 스냅샷. KRX 가 그 날 빈 응답이면 직전 영업일로 backoff.
|
|
|
|
KRX 는 토/일/공휴일에 데이터를 안 주므로 target 부터 최대 _MAX_BACKOFF_DAYS 일 이전까지
|
|
역순으로 시도. 시총 / 멀티플 두 호출을 같은 target_date 로 정렬.
|
|
"""
|
|
try:
|
|
from pykrx import stock as krx
|
|
except ImportError:
|
|
logger.warning("pykrx unavailable")
|
|
return None
|
|
|
|
cur = target
|
|
for _ in range(_MAX_BACKOFF_DAYS):
|
|
ds = _to_pykrx(cur)
|
|
try:
|
|
cap_df = krx.get_market_cap(ds, ds, code)
|
|
fund_df = krx.get_market_fundamental(ds, ds, code)
|
|
except Exception as exc: # noqa: BLE001
|
|
# logger.exception 은 예전에 pykrx/KRX 비정상 응답 traceback 포매팅 중
|
|
# UnicodeDecodeError 로 API 500 을 만든 적이 있음 → %r 로 안전하게 직렬화.
|
|
logger.error("pykrx fundamentals failed code=%s date=%s err=%r", code, ds, exc)
|
|
return None
|
|
|
|
cap_ok = cap_df is not None and not cap_df.empty
|
|
fund_ok = fund_df is not None and not fund_df.empty
|
|
if cap_ok or fund_ok:
|
|
# 한쪽이라도 있으면 그 날짜로 채택. 결측은 None.
|
|
def _val(df: Any, key: str) -> float | None:
|
|
if df is None or df.empty:
|
|
return None
|
|
v = df.iloc[0].get(key)
|
|
try:
|
|
if v is None:
|
|
return None
|
|
f = float(v)
|
|
if f != f: # NaN
|
|
return None
|
|
return f
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
# pykrx fundamental 컬럼: ['BPS', 'PER', 'PBR', 'EPS', 'DIV', 'DPS']
|
|
# DIV = 배당수익률 (%), DPS = 주당배당금 (원).
|
|
mc = _val(cap_df, "시가총액")
|
|
shares_f = _val(cap_df, "상장주식수")
|
|
shares: int | None = int(shares_f) if shares_f is not None else None
|
|
return Fundamentals(
|
|
code=code,
|
|
snapshot_date=cur,
|
|
market_cap=mc,
|
|
shares_outstanding=shares,
|
|
per=_val(fund_df, "PER"),
|
|
pbr=_val(fund_df, "PBR"),
|
|
eps=_val(fund_df, "EPS"),
|
|
bps=_val(fund_df, "BPS"),
|
|
div=_val(fund_df, "DPS"),
|
|
dvd_yield=_val(fund_df, "DIV"),
|
|
)
|
|
cur = cur - timedelta(days=1)
|
|
|
|
return None
|
|
|
|
|
|
def get_fundamentals(code: str, *, today: date | None = None) -> Fundamentals | None:
|
|
"""code 의 최신 영업일 fundamentals. 같은 (code, today) 는 인메모리 캐시."""
|
|
t = today or date.today()
|
|
key = (code, t)
|
|
with _cache_lock:
|
|
if key in _cache:
|
|
return _cache[key]
|
|
val = _fetch_one(code, t)
|
|
with _cache_lock:
|
|
_cache[key] = val
|
|
return val
|