fix: universe snapshot + gainer 수집 안정성 개선

- universe_service: yf.download() → Ticker.history() (최신 날짜 버그 수정)
- universe_service: 이미 완료된 ticker skip 로직 추가 (재실행 시 중복 HTTP 방지)
- universe_service: SEC 캐시 클리어 주기 20→5배치 (OOM 방지)
- yahoo_client: 성공 fetch 후 세션 즉시 교체 (Yahoo soft rate-limit 대응)
- scripts/build_snapshots_batched.py: 재작성 — 날짜 자동화, GC 명시, 배치 크기 30

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app/services/gainers/yahoo_client.py

@@ -144,6 +144,14 @@ class _YahooScreenerClient:
         with self._lock:
             self._sessions[self._idx % len(self._sessions)] = new_s
 
+    def _rotate_used(self):
+        """Replace the slot that was just used (idx already incremented after _next_session)."""
+        profile = random.choice(_BROWSER_PROFILES)
+        new_s = self._make_session(profile)
+        with self._lock:
+            slot = (self._idx - 1) % len(self._sessions)
+            self._sessions[slot] = new_s
+
     def fetch_sync(
         self,
         preset: str = "day_gainers",
@@ -184,6 +192,8 @@ class _YahooScreenerClient:
                 result = (resp.json().get("finance", {}).get("result") or [{}])[0]
                 quotes = result.get("quotes", [])
                 total = result.get("total", len(quotes))
+                # Always replace the just-used session so each 5-min fetch starts fresh
+                self._rotate_used()
                 return quotes, total, None
             except Exception as e:
                 last_exc = e

app/services/universe_service.py

@@ -275,6 +275,9 @@ class UniverseService:
                 yield _session
 
         # ---- Resolve ticker list ----
+        start_dt = datetime.fromisoformat(start_date).replace(tzinfo=timezone.utc)
+        end_dt   = datetime.fromisoformat(end_date).replace(tzinfo=timezone.utc)
+
         async with _db_ctx() as db:
             if tickers:
                 ticker_list = [t.upper() for t in tickers]
@@ -290,6 +293,24 @@ class UniverseService:
                 logger.warning("Universe: no tickers to process")
                 return {"tickers_processed": 0, "tickers_failed": 0, "snapshots_created": 0}
 
+            # Skip tickers that already have complete snapshots for this date range
+            # (unless force_rebuild). Avoids redundant yfinance + SEC HTTP calls.
+            if not force_rebuild:
+                done_result = await db.execute(
+                    select(UniverseSnapshot.ticker)
+                    .where(
+                        UniverseSnapshot.snapshot_date >= start_dt,
+                        UniverseSnapshot.snapshot_date < end_dt,
+                    )
+                    .distinct()
+                )
+                done_tickers = {r[0] for r in done_result.fetchall()}
+                before = len(ticker_list)
+                ticker_list = [t for t in ticker_list if t not in done_tickers]
+                skipped = before - len(ticker_list)
+                if skipped:
+                    logger.info(f"Universe: skipping {skipped} tickers already in DB for this range")
+
             logger.info(
                 f"Universe: building snapshots for {len(ticker_list)} tickers "
                 f"({start_date} → {end_date})"
@@ -326,7 +347,7 @@ class UniverseService:
         total_snapshots = 0
         total_failed = 0
         total_batches = (len(ticker_list) + _PRICE_BATCH - 1) // _PRICE_BATCH
-        _CACHE_CLEAR_EVERY = 20   # clear SEC in-memory cache every N batches to prevent OOM
+        _CACHE_CLEAR_EVERY = 5   # clear SEC in-memory cache every N batches to prevent OOM
 
         # ---- Process in batches — each batch uses its own DB session ----
         for batch_start in range(0, len(ticker_list), _PRICE_BATCH):
@@ -487,12 +508,15 @@ class UniverseService:
         end: str,
     ) -> Dict[str, Dict[date, float]]:
         """
-        Synchronous bulk monthly price fetch via yfinance.
+        Synchronous bulk monthly price fetch via yfinance Ticker.history().
         Returns {ticker: {date: close_price}}.
+
+        Uses per-ticker history() instead of download() because yfinance's
+        download() returns empty DataFrames for recent date ranges (post-2025).
         """
         try:
             import math as _math
-            import yfinance_plus as yf
+            import yfinance as _yf
         except ImportError:
             logger.error("yfinance not available for price download")
             return {}
@@ -500,26 +524,6 @@ class UniverseService:
         if not tickers:
             return {}
 
-        # Always use group_by='ticker' for consistent MultiIndex column structure:
-        # data[ticker] -> DataFrame with ['Close', 'Adj Close', ...]
-        try:
-            data = yf.download(
-                tickers=tickers,
-                start=start,
-                end=end,
-                interval="1mo",
-                auto_adjust=False,
-                progress=False,
-                threads=False,   # disable internal thread pool — prevents OS thread explosion
-                group_by="ticker",
-            )
-        except Exception as e:
-            logger.warning(f"Universe: bulk price download failed: {e}")
-            return {}
-
-        if data is None or data.empty:
-            return {}
-
         result: Dict[str, Dict[date, float]] = {}
 
         def _safe_close(val) -> Optional[float]:
@@ -541,15 +545,15 @@ class UniverseService:
         for ticker in tickers:
             result[ticker] = {}
             try:
-                # With group_by='ticker', columns are MultiIndex (ticker, price_type)
+                hist = _yf.Ticker(ticker).history(
-                # data[ticker] gives a flat DataFrame with price columns
+                    start=start,
-                lvl0 = data.columns.get_level_values(0)
+                    end=end,
-                if ticker not in lvl0:
+                    interval="1mo",
+                    auto_adjust=False,
+                )
+                if hist is None or hist.empty:
                     continue
-                ticker_df = data[ticker]
+                close_col = hist.get("Close")
-                close_col = ticker_df.get("Close")
-                if close_col is None:
-                    close_col = ticker_df.get("Adj Close")
                 if close_col is None:
                     continue
                 for dt, val in close_col.items():
@@ -558,7 +562,7 @@ class UniverseService:
                     if d and c:
                         result[ticker][d] = c
             except Exception as e:
-                logger.debug(f"Universe: price parse error for {ticker}: {e}")
+                logger.debug(f"Universe: price fetch error for {ticker}: {e}")
 
         return result
 

scripts/build_snapshots_batched.py

@@ -0,0 +1,123 @@
+"""
+Universe snapshot backfill script.
+
+Usage:
+    python3 build_snapshots_batched.py [start_date] [end_date]
+
+Defaults:
+    start_date: 2015-01-01
+    end_date:   today (first day of current month)
+
+Runs inside the stock_oracle_api container via:
+    docker exec -d stock_oracle_api bash -c "python3 /app/scripts/build_snapshots_batched.py > /tmp/build.log 2>&1"
+
+Features:
+- Skips tickers already fully covered in the date range (no redundant HTTP calls)
+- Processes registry tickers in batches of BATCH_SIZE to limit memory usage
+- Logs progress with batch index, snapshots created/failed, elapsed time
+"""
+import asyncio
+import logging
+import sys
+import os
+import time
+from datetime import date, datetime, timezone
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(message)s",
+    datefmt="%H:%M:%S",
+)
+logger = logging.getLogger(__name__)
+
+BATCH_SIZE = 30
+
+def _resolve_dates():
+    today = date.today()
+    end = date(today.year, today.month, 1)   # first day of current month
+    start = date(2015, 1, 1)
+
+    if len(sys.argv) >= 2:
+        start = date.fromisoformat(sys.argv[1])
+    if len(sys.argv) >= 3:
+        end = date.fromisoformat(sys.argv[2])
+
+    return start.isoformat(), end.isoformat()
+
+
+async def get_all_tickers(engine):
+    from sqlalchemy import text
+    from sqlalchemy.ext.asyncio import AsyncSession
+    async with AsyncSession(engine) as session:
+        result = await session.execute(
+            text("SELECT ticker FROM universe_ticker_registry WHERE is_active = true ORDER BY ticker")
+        )
+        return [r[0] for r in result.fetchall()]
+
+
+async def run_batch(svc, session_factory, tickers, batch_num, total_batches, start_date, end_date):
+    import gc
+    t0 = time.time()
+    try:
+        result = await svc.build_snapshots(
+            session_factory,
+            tickers=tickers,
+            start_date=start_date,
+            end_date=end_date,
+            force_rebuild=False,
+        )
+        elapsed = time.time() - t0
+        created = result.get("snapshots_created", "?")
+        failed  = result.get("tickers_failed", "?")
+        processed = result.get("tickers_processed", len(tickers))
+        logger.info(
+            "[%d/%d] done in %.1fs — created=%s failed=%s processed=%d tickers=%s..%s",
+            batch_num, total_batches, elapsed, created, failed, processed,
+            tickers[0], tickers[-1],
+        )
+    except Exception as e:
+        logger.error("[%d/%d] FAILED (%s..%s): %s", batch_num, total_batches, tickers[0], tickers[-1], e)
+    finally:
+        gc.collect()
+
+
+async def main():
+    sys.path.insert(0, "/app")
+    os.environ.setdefault("ENV", "production")
+
+    start_date, end_date = _resolve_dates()
+
+    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+    from app.core.config import settings
+    from app.services.universe_service import UniverseService
+
+    engine = create_async_engine(
+        settings.DATABASE_URL,
+        echo=False,
+        pool_size=3,
+        max_overflow=0,
+        pool_timeout=60,
+        pool_pre_ping=True,
+        connect_args={"server_settings": {"jit": "off"}},
+    )
+    session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+    all_tickers = await get_all_tickers(engine)
+    total = len(all_tickers)
+    batches = [all_tickers[i:i+BATCH_SIZE] for i in range(0, total, BATCH_SIZE)]
+    total_batches = len(batches)
+
+    logger.info("Backfill range: %s → %s", start_date, end_date)
+    logger.info("Registry tickers: %d  |  Batch size: %d  |  Total batches: %d", total, BATCH_SIZE, total_batches)
+    logger.info("Already-covered tickers will be skipped automatically per batch.")
+
+    svc = UniverseService()
+    for idx, batch in enumerate(batches, start=1):
+        await run_batch(svc, session_factory, batch, idx, total_batches, start_date, end_date)
+
+    await engine.dispose()
+    logger.info("All batches complete.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())