코드 변경

backend/app/services/naver_finance.py

@@ -0,0 +1,121 @@
+"""네이버 파이낸스 종목 목록 크롤러
+
+KIS volume-rank API 대신 네이버 시가총액 순위 페이지에서
+KOSPI/KOSDAQ 종목 목록을 가져온다.
+
+사용처:
+  - batch_daily_analysis.get_all_tradable_stocks() DB 폴백 이전 1차 시도
+  - market_analyzer.get_all_tradable_stocks() 대체
+
+URL: https://finance.naver.com/sise/sise_market_sum.naver?sosok=0&page=N
+  sosok=0 : KOSPI
+  sosok=1 : KOSDAQ
+"""
+
+import asyncio
+import logging
+from typing import List, Tuple
+
+import httpx
+from bs4 import BeautifulSoup
+
+logger = logging.getLogger(__name__)
+
+_HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; autostock-batch/1.0)"}
+_BASE_URL = "https://finance.naver.com/sise/sise_market_sum.naver"
+
+SKIP_KEYWORDS = [
+    "ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG",
+    "KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", "RISE",
+    "선물", "스팩", "리츠", "SPAC",
+]
+
+_MIN_PRICE = 5_000  # 5,000원 미만 제외
+
+
+def _parse_page(html: bytes) -> List[Tuple[str, str, int]]:
+    """HTML 파싱 → [(종목코드, 종목명, 현재가), ...]"""
+    soup = BeautifulSoup(html, "html.parser", from_encoding="euc-kr")
+    table = soup.find("table", {"class": "type_2"})
+    if not table:
+        return []
+
+    result = []
+    for row in table.find_all("tr"):
+        a = row.find("a", class_="tltle")
+        if not a:
+            continue
+        href = a.get("href", "")
+        if "code=" not in href:
+            continue
+        code = href.split("code=")[1]
+        name = a.text.strip()
+
+        # 현재가
+        tds = row.find_all("td", class_="number")
+        try:
+            price = int(tds[0].text.strip().replace(",", "")) if tds else 0
+        except (ValueError, IndexError):
+            price = 0
+
+        result.append((code, name, price))
+    return result
+
+
+async def _fetch_page(client: httpx.AsyncClient, sosok: int, page: int) -> List[Tuple[str, str, int]]:
+    try:
+        resp = await client.get(
+            _BASE_URL,
+            params={"sosok": sosok, "page": page},
+            headers=_HEADERS,
+            timeout=10,
+        )
+        resp.raise_for_status()
+        return _parse_page(resp.content)
+    except Exception as exc:
+        logger.warning("네이버 파이낸스 조회 실패 sosok=%d page=%d: %s", sosok, page, exc)
+        return []
+
+
+async def get_naver_stock_list(
+    max_pages_per_market: int = 10,
+    min_price: int = _MIN_PRICE,
+) -> List[Tuple[str, str]]:
+    """KOSPI + KOSDAQ 시가총액 순위 종목 조회.
+
+    Args:
+        max_pages_per_market: 시장별 최대 페이지 수 (페이지당 ~50종목)
+        min_price: 최소 현재가 필터 (기본 5,000원)
+
+    Returns:
+        [(종목코드, 종목명), ...] — ETF/레버리지/스팩 제외, 가격 필터 적용
+    """
+    stocks: List[Tuple[str, str]] = []
+    seen_codes: set = set()
+
+    async with httpx.AsyncClient() as client:
+        for sosok, market_name in ((0, "KOSPI"), (1, "KOSDAQ")):
+            market_stocks = []
+            for page in range(1, max_pages_per_market + 1):
+                rows = await _fetch_page(client, sosok, page)
+                if not rows:
+                    break  # 마지막 페이지
+
+                for code, name, price in rows:
+                    if code in seen_codes:
+                        continue
+                    if price < min_price:
+                        continue
+                    name_upper = name.upper()
+                    if any(kw.upper() in name_upper for kw in SKIP_KEYWORDS):
+                        continue
+                    seen_codes.add(code)
+                    market_stocks.append((code, name))
+
+                await asyncio.sleep(0.3)  # 서버 부하 방지
+
+            logger.info("네이버 %s: %d개 종목 수집", market_name, len(market_stocks))
+            stocks.extend(market_stocks)
+
+    logger.info("네이버 파이낸스 총 %d개 종목 수집", len(stocks))
+    return stocks