stock 실계좌 정보 표출 추가

stock-lab/app/price_fetcher.py

@@ -0,0 +1,68 @@
+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