From 79164ecb6673d512ae7c99d6f6d74679c9c42213 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Sun, 17 May 2026 06:47:43 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20universe=20snapshot=20+=20gainer=20?= =?UTF-8?q?=EC=88=98=EC=A7=91=20=EC=95=88=EC=A0=95=EC=84=B1=20=EA=B0=9C?= =?UTF-8?q?=EC=84=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- app/services/gainers/yahoo_client.py | 10 +++ app/services/universe_service.py | 68 ++++++++------- scripts/build_snapshots_batched.py | 123 +++++++++++++++++++++++++++ 3 files changed, 169 insertions(+), 32 deletions(-) create mode 100644 scripts/build_snapshots_batched.py diff --git a/app/services/gainers/yahoo_client.py b/app/services/gainers/yahoo_client.py index c84e8d9..76d3ab8 100644 --- a/app/services/gainers/yahoo_client.py +++ b/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 diff --git a/app/services/universe_service.py b/app/services/universe_service.py index 7803c2b..3f68e66 100644 --- a/app/services/universe_service.py +++ b/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) - # data[ticker] gives a flat DataFrame with price columns - lvl0 = data.columns.get_level_values(0) - if ticker not in lvl0: + hist = _yf.Ticker(ticker).history( + start=start, + end=end, + interval="1mo", + auto_adjust=False, + ) + if hist is None or hist.empty: continue - ticker_df = data[ticker] - close_col = ticker_df.get("Close") - if close_col is None: - close_col = ticker_df.get("Adj Close") + close_col = hist.get("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 diff --git a/scripts/build_snapshots_batched.py b/scripts/build_snapshots_batched.py new file mode 100644 index 0000000..424d5b8 --- /dev/null +++ b/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())