hyowons
116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
"""기준지수(KOSPI/KOSDAQ) 일별 종가 캐시 + 벤치마크 수익률·알파.
|
||
|
||
네이버 m.stock 지수 일별시세 API에서 받아 state/sim/index_history.json 에 누적.
|
||
sim/백테스트 수익률을 '같은 기간 지수 매수후보유' 대비(알파)로 평가하기 위한 데이터원.
|
||
실패해도 raise 하지 않고 None 으로 흘려보낸다 (지표 부재 ≠ 엔진 중단).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import urllib.request
|
||
|
||
from . import config
|
||
|
||
INDEX_PATH = config.STATE_DIR / 'index_history.json'
|
||
INDICES = ('KOSPI', 'KOSDAQ')
|
||
INDEX_LABEL = {'KOSPI': '코스피', 'KOSDAQ': '코스닥'}
|
||
|
||
|
||
def norm_date(d: str) -> str:
|
||
"""'2026-06-09', '2026-06-09T...', '20260609' → 'YYYYMMDD'."""
|
||
return (d or '').replace('-', '')[:8]
|
||
|
||
|
||
def _load() -> dict:
|
||
try:
|
||
return json.loads(INDEX_PATH.read_text())
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _save(data: dict) -> None:
|
||
config.STATE_DIR.mkdir(parents=True, exist_ok=True)
|
||
INDEX_PATH.write_text(json.dumps(data, ensure_ascii=False))
|
||
|
||
|
||
def _fetch_page(index: str, page: int, size: int) -> list[dict]:
|
||
url = f'https://m.stock.naver.com/api/index/{index}/price?pageSize={size}&page={page}'
|
||
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
|
||
with urllib.request.urlopen(req, timeout=6.0) as r:
|
||
return json.loads(r.read().decode('utf-8', 'ignore')) or []
|
||
|
||
|
||
def backfill(pages: int = 8, size: int = 50) -> dict:
|
||
"""KOSPI/KOSDAQ 일별 종가를 pages×size 만큼 받아 캐시에 병합(idempotent). {index: 신규건수}."""
|
||
data = _load()
|
||
added: dict[str, int] = {}
|
||
for idx in INDICES:
|
||
store = data.setdefault(idx, {})
|
||
n0 = len(store)
|
||
for p in range(1, pages + 1):
|
||
try:
|
||
rows = _fetch_page(idx, p, size)
|
||
except Exception:
|
||
break
|
||
if not rows:
|
||
break
|
||
for row in rows:
|
||
d = norm_date(row.get('localTradedAt'))
|
||
c = row.get('closePrice')
|
||
if not d or not c:
|
||
continue
|
||
try:
|
||
store[d] = float(str(c).replace(',', ''))
|
||
except (ValueError, TypeError):
|
||
pass
|
||
added[idx] = len(store) - n0
|
||
_save(data)
|
||
return added
|
||
|
||
|
||
def update_today() -> None:
|
||
"""최신 1페이지만 받아 오늘 종가 갱신 (스캔마다 저비용 호출, 실패 무시)."""
|
||
try:
|
||
backfill(pages=1, size=10)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def series(index: str) -> dict:
|
||
return _load().get(index, {})
|
||
|
||
|
||
def _nearest(store: dict, date: str, after: bool):
|
||
"""date 기준 on-or-after(after=True) / on-or-before 가장 가까운 (date, close). 없으면 None."""
|
||
if not store:
|
||
return None
|
||
d = norm_date(date)
|
||
keys = sorted(store)
|
||
if after:
|
||
cand = [k for k in keys if k >= d]
|
||
k = cand[0] if cand else None
|
||
else:
|
||
cand = [k for k in keys if k <= d]
|
||
k = cand[-1] if cand else None
|
||
return (k, store[k]) if k else None
|
||
|
||
|
||
def benchmark_return(index: str, date_from: str, date_to: str):
|
||
"""[from, to] 구간 지수 등락률(%). from=on-or-after, to=on-or-before 종가. 데이터 없으면 None."""
|
||
store = series(index)
|
||
a = _nearest(store, date_from, after=True)
|
||
b = _nearest(store, date_to, after=False)
|
||
if not a or not b or a[1] <= 0 or b[0] <= a[0]:
|
||
return None
|
||
return round((b[1] / a[1] - 1) * 100, 2)
|
||
|
||
|
||
def compare(return_pct, date_from: str, date_to: str) -> dict:
|
||
"""기간 수익률(return_pct)을 지수 대비로 비교. {index: {'pct':지수등락, 'alpha':초과}}."""
|
||
out: dict[str, dict] = {}
|
||
for idx in INDICES:
|
||
b = benchmark_return(idx, date_from, date_to)
|
||
alpha = round(return_pct - b, 2) if (b is not None and return_pct is not None) else None
|
||
out[idx] = {'pct': b, 'alpha': alpha}
|
||
return out
|