69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
import time
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from typing import Optional
|
|
|
|
_cache: dict[str, tuple[Optional[int], float]] = {} # ticker -> (price, timestamp)
|
|
_CACHE_TTL = 180 # 3분
|
|
|
|
_HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/90.0.4430.93 Safari/537.36"
|
|
)
|
|
}
|
|
|
|
|
|
def _fetch_from_mobile_api(ticker: str) -> Optional[int]:
|
|
"""네이버 모바일 주식 API로 현재가 조회"""
|
|
url = f"https://m.stock.naver.com/api/stock/{ticker}/basic"
|
|
try:
|
|
resp = requests.get(url, headers=_HEADERS, timeout=5)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
price_str = data.get("closePrice") or data.get("stockEndPrice") or ""
|
|
price_str = str(price_str).replace(",", "").strip()
|
|
return int(price_str) if price_str.isdigit() else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _fetch_from_html_fallback(ticker: str) -> Optional[int]:
|
|
"""네이버 금융 HTML 폴백 (.no_today .blind 파싱)"""
|
|
url = f"https://finance.naver.com/item/main.naver?code={ticker}"
|
|
try:
|
|
resp = requests.get(url, headers=_HEADERS, timeout=5)
|
|
resp.raise_for_status()
|
|
soup = BeautifulSoup(resp.content, "html.parser", from_encoding="cp949")
|
|
tag = soup.select_one(".no_today .blind")
|
|
if tag:
|
|
price_str = tag.get_text(strip=True).replace(",", "")
|
|
return int(price_str) if price_str.isdigit() else None
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_current_price(ticker: str) -> Optional[int]:
|
|
"""단건 현재가 조회 (3분 캐시)"""
|
|
now = time.time()
|
|
cached = _cache.get(ticker)
|
|
if cached and (now - cached[1]) < _CACHE_TTL:
|
|
return cached[0]
|
|
|
|
price = _fetch_from_mobile_api(ticker)
|
|
if price is None:
|
|
price = _fetch_from_html_fallback(ticker)
|
|
|
|
_cache[ticker] = (price, now)
|
|
return price
|
|
|
|
|
|
def get_current_prices(tickers: list[str]) -> dict[str, Optional[int]]:
|
|
"""배치 현재가 조회 (캐시 미스 종목만 실제 호출)"""
|
|
result: dict[str, Optional[int]] = {}
|
|
for ticker in tickers:
|
|
result[ticker] = get_current_price(ticker)
|
|
return result
|