a79e784550
backend:
- fetch/fundamentals.py: get_fundamentals(code) — pykrx 의 get_market_cap +
get_market_fundamental 을 같은 영업일로 정렬해서 호출. 토/일/공휴일이면 KRX 가 빈
응답을 줘서 최대 7일까지 역순 backoff. (code, today) 인메모리 daily 캐시로 중복 호출 방지.
- api/fundamentals.py: GET /api/fundamentals/{code} — 없는 코드 404, KRX 가 데이터를
안 주는 케이스는 status="no_data" + null 들. 200 응답으로 단순화 (호출자 분기 편의).
- main.py: fundamentals_router 등록.
frontend:
- lib/api.ts: FundamentalsResponse 타입 + api.fundamentals(code).
- components/SymbolSidebar.tsx: 1Y 차트와 별도로 fundamentals 호출. ok 일 때만
'기업 지표' 섹션 노출 — 시가총액(조/억) / 상장주식수 / PER / PBR / EPS / BPS /
배당수익률(%) / 주당배당금. PER·PBR 음수면 '적자' 표기 (이익 음수 → 멀티플 무의미).
- 기존 시세/52주 게이지 코멘트에 fundamentals 추가됨을 반영.
pykrx 종목당 호출 1~2초 → 첫 페이지 로딩이 살짝 느려질 수 있으나 daily 캐시로
재방문은 즉시. 펀더멘털 실패는 silent — 시세 표시는 영향 없음.
121 lines
4.4 KiB
Python
121 lines
4.4 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: # noqa: BLE001
|
|
logger.exception("pykrx fundamentals failed code=%s date=%s", code, ds)
|
|
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
|