web-page-backend/stock-lab/app/price_fetcher.py

import time
import requests
from bs4 import BeautifulSoup
from typing import Optional

# 캐시는 detail 단위(가격+세션+as_of)로 보관. 호환용 단순 가격은 여기서 추출.
_cache: dict[str, tuple[Optional[dict], float]] = {}  # ticker -> (detail | None, 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 _parse_price_str(value) -> Optional[int]:
    if value is None:
        return None
    s = str(value).replace(",", "").strip()
    if not s:
        return None
    # 음수/소수점도 일단 정수 라운드(국내 주식은 정수)
    try:
        return int(float(s))
    except ValueError:
        return None


def _select_price_from_response(payload: dict) -> dict:
    """네이버 모바일 주식 API 응답 dict에서 (price, session, as_of)를 결정.

    세션 분류:
      - "REGULAR"   : 정규장(KRX) 운영중 — closePrice가 실시간
      - "NXT_PRE"   : 정규장 마감 + NXT 프리마켓 운영중 → overPrice 사용
      - "NXT_AFTER" : 정규장 마감 + NXT 애프터마켓 운영중 → overPrice 사용
      - "CLOSED"    : 정규장 마감 + NXT 비운영/거래중지 → closePrice 사용

    반환 dict: {"price": int | None, "session": str, "as_of": str | None}
    """
    close_price = _parse_price_str(payload.get("closePrice") or payload.get("stockEndPrice"))
    top_as_of = payload.get("localTradedAt")

    market_status = (payload.get("marketStatus") or "").upper()
    if market_status == "OPEN":
        return {"price": close_price, "session": "REGULAR", "as_of": top_as_of}

    over = payload.get("overMarketPriceInfo")
    if isinstance(over, dict):
        over_status = (over.get("overMarketStatus") or "").upper()
        trade_stop_name = ((over.get("tradeStopType") or {}).get("name") or "").upper()
        if over_status == "OPEN" and trade_stop_name == "TRADING":
            over_price = _parse_price_str(over.get("overPrice"))
            if over_price is not None:
                session_type = (over.get("tradingSessionType") or "").upper()
                if session_type == "PRE_MARKET":
                    session = "NXT_PRE"
                elif session_type == "AFTER_MARKET":
                    session = "NXT_AFTER"
                else:
                    # 알 수 없는 NXT 세션은 보수적으로 AFTER 취급
                    session = "NXT_AFTER"
                return {
                    "price": over_price,
                    "session": session,
                    "as_of": over.get("localTradedAt") or top_as_of,
                }

    return {"price": close_price, "session": "CLOSED", "as_of": top_as_of}


def _fetch_mobile_api_payload(ticker: str) -> Optional[dict]:
    """네이버 모바일 주식 API 응답 dict 반환."""
    url = f"https://m.stock.naver.com/api/stock/{ticker}/basic"
    try:
        resp = requests.get(url, headers=_HEADERS, timeout=5)
        resp.raise_for_status()
        return resp.json()
    except Exception:
        return None


def _fetch_close_price_from_html(ticker: str) -> Optional[int]:
    """네이버 금융 HTML 폴백 (정규장 종가만 가능, NXT 정보 없음)."""
    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:
            return _parse_price_str(tag.get_text(strip=True))
        return None
    except Exception:
        return None


def get_current_price_info(ticker: str) -> Optional[dict]:
    """단건 상세 가격 정보 조회 (3분 캐시).

    반환: {"price": int | None, "session": str, "as_of": str | None} | None
    """
    now = time.time()
    cached = _cache.get(ticker)
    if cached and (now - cached[1]) < _CACHE_TTL:
        return cached[0]

    detail: Optional[dict] = None
    payload = _fetch_mobile_api_payload(ticker)
    if isinstance(payload, dict):
        detail = _select_price_from_response(payload)
        if detail.get("price") is None:
            detail = None  # 폴백 시도

    if detail is None:
        fallback_price = _fetch_close_price_from_html(ticker)
        if fallback_price is not None:
            detail = {"price": fallback_price, "session": "CLOSED", "as_of": None}

    _cache[ticker] = (detail, now)
    return detail


def get_current_prices_detail(tickers: list[str]) -> dict[str, Optional[dict]]:
    """배치 상세 가격 조회 (캐시 미스 종목만 실제 호출)."""
    return {ticker: get_current_price_info(ticker) for ticker in tickers}


def get_current_price(ticker: str) -> Optional[int]:
    """단건 현재가 조회 — 호환용. detail에서 price만 추출."""
    detail = get_current_price_info(ticker)
    return detail["price"] if detail else None


def get_current_prices(tickers: list[str]) -> dict[str, Optional[int]]:
    """배치 현재가 조회 — 호환용."""
    return {ticker: get_current_price(ticker) for ticker in tickers}