From 2fb6ec0f0c09365f7a8baf625338875ccaf4e932 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Sun, 31 May 2026 10:10:34 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20price=5Fdata=5Fservice=20null/zero=20clo?= =?UTF-8?q?se=20=EC=9E=90=EB=8F=99=20=EC=9E=AC=EC=88=98=EC=A7=91=20+=20Yah?= =?UTF-8?q?oo=20rate=20limit=20429=20+=20earnings=20=EC=BA=90=EC=8B=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - price_data_service: null/zero close를 missing으로 처리 → 재수집 트리거 UTC 자정 정규화로 yf.history / yf.download 중복 방지 get_quote fallback: .info 실패 시 fast_info + history('2d') 체인 historical upsert: close IS NULL/0인 경우만 덮어쓰기 (정상 데이터 보호) - price.py: Yahoo rate limit → HTTP 429 + Retry-After: 30 응답 - earnings.py: bulk calendar 엔드포인트 캐시 1시간 추가 Co-Authored-By: Claude Opus 4.8 --- app/api/v1/endpoints/earnings.py | 1 + app/api/v1/endpoints/price.py | 46 ++++++-- app/services/price_data_service.py | 180 ++++++++++++++++++++++------- 3 files changed, 175 insertions(+), 52 deletions(-) diff --git a/app/api/v1/endpoints/earnings.py b/app/api/v1/endpoints/earnings.py index 8e241fa..f1375be 100644 --- a/app/api/v1/endpoints/earnings.py +++ b/app/api/v1/endpoints/earnings.py @@ -102,6 +102,7 @@ async def get_earnings_calendar( "Useful for checking upcoming earnings of sector peers or candidates." ), ) +@with_cache(namespace="earnings:calendar_bulk", ttl=3600, key_params=["request"]) async def get_bulk_earnings_calendar( request: BulkEarningsCalendarRequest, response: Response, diff --git a/app/api/v1/endpoints/price.py b/app/api/v1/endpoints/price.py index 995cebe..d378399 100644 --- a/app/api/v1/endpoints/price.py +++ b/app/api/v1/endpoints/price.py @@ -641,8 +641,18 @@ async def get_quote( use_prepost: bool = Query(True, description="Include pre/post market prices if available"), ): svc = PriceDataService() - data = await svc.get_quote(ticker, use_prepost=use_prepost) - return QuoteResponse(**data) + try: + data = await svc.get_quote(ticker, use_prepost=use_prepost) + return QuoteResponse(**data) + except Exception as e: + err = str(e).lower() + if "rate limit" in err or "too many requests" in err or "429" in err or "ratelimit" in err: + raise HTTPException( + status_code=429, + detail={"error_type": "RATE_LIMIT_ERROR", "message": "Yahoo Finance rate limit. Retry after a short delay."}, + headers={"Retry-After": "30"}, + ) + raise HTTPException(status_code=500, detail={"error_type": "SERVER_ERROR", "message": str(e)}) @router.get( "/intraday", @@ -729,14 +739,20 @@ async def get_intraday( period: str = Query("1d"), ): svc = PriceDataService() - candles = await svc.get_intraday(ticker, interval=interval, period=period) - return IntradayResponse( - ticker=ticker.upper(), - interval=interval, - period=period, - candles=[IntradayCandle(**c) for c in candles], - metadata={"count": len(candles)} - ) + try: + candles = await svc.get_intraday(ticker, interval=interval, period=period) + return IntradayResponse( + ticker=ticker.upper(), + interval=interval, + period=period, + candles=[IntradayCandle(**c) for c in candles], + metadata={"count": len(candles)} + ) + except Exception as e: + err = str(e).lower() + if "rate limit" in err or "too many requests" in err or "429" in err or "ratelimit" in err: + raise HTTPException(status_code=429, detail={"error_type": "RATE_LIMIT_ERROR", "message": "Yahoo Finance rate limit. Retry after a short delay."}, headers={"Retry-After": "30"}) + raise HTTPException(status_code=500, detail={"error_type": "SERVER_ERROR", "message": str(e)}) @router.get( "/today/{ticker}", @@ -748,5 +764,11 @@ async def get_today_ohlc( ticker: str, ): svc = PriceDataService() - data = await svc.get_today_ohlc(ticker) - return TodayOHLCResponse(**data) \ No newline at end of file + try: + data = await svc.get_today_ohlc(ticker) + return TodayOHLCResponse(**data) + except Exception as e: + err = str(e).lower() + if "rate limit" in err or "too many requests" in err or "429" in err or "ratelimit" in err: + raise HTTPException(status_code=429, detail={"error_type": "RATE_LIMIT_ERROR", "message": "Yahoo Finance rate limit. Retry after a short delay."}, headers={"Retry-After": "30"}) + raise HTTPException(status_code=500, detail={"error_type": "SERVER_ERROR", "message": str(e)}) \ No newline at end of file diff --git a/app/services/price_data_service.py b/app/services/price_data_service.py index cb9b273..70b1877 100644 --- a/app/services/price_data_service.py +++ b/app/services/price_data_service.py @@ -6,7 +6,7 @@ from datetime import datetime, timezone, timedelta, date from typing import Dict, List, Optional, Tuple, Union import logging from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy import select, and_, desc +from sqlalchemy import select, and_, desc, or_ import asyncio import sys import os @@ -120,14 +120,16 @@ class PriceDataService: if end_date.date() >= today: return [datetime.now()] - # Check if we have any data for this ticker and interval + # Count only rows with valid close (null/0 rows are treated as missing) result = await db.execute( select(PriceData.date) .where( and_( PriceData.ticker == ticker, PriceData.date >= start_date, - PriceData.date <= end_date + PriceData.date <= end_date, + PriceData.close.is_not(None), + PriceData.close > 0, ) ) .order_by(PriceData.date) @@ -228,22 +230,28 @@ class PriceDataService: return hist_data async def get_quote(self, ticker: str, use_prepost: bool = True) -> Dict: - """Get latest quote using yfinance-plus .info fields with fallback to fast history last row.""" + """Get latest quote using yfinance-plus .info fields with fallback to fast_info + history.""" if not self.yf_available: raise ValueError("Yahoo Finance (yfinance-plus) data source not available") + + yf_ticker = yf.Ticker(ticker) + loop = asyncio.get_event_loop() + + # Try .info first; fall back to fast_info + history on timeout/error + info = None try: - yf_ticker = yf.Ticker(ticker) - loop = asyncio.get_event_loop() info = await _run_with_timeout( loop.run_in_executor(None, lambda: yf_ticker.info), timeout_seconds=20, description=f"info {ticker}" ) - # Prefer regular/post/pre values + except Exception as e: + logger.warning(f"get_quote .info failed for {ticker}: {e}, falling back to fast_info") + + if info: regular = info.get("regularMarketPrice") post = info.get("postMarketPrice") if use_prepost else None pre = info.get("preMarketPrice") if use_prepost else None - price = post or pre or regular currency = info.get("currency") exchange = info.get("exchange") or info.get("fullExchangeName") market_state = info.get("marketState") @@ -255,22 +263,67 @@ class PriceDataService: ts = ts.replace(tzinfo=timezone.utc) else: ts = datetime.now(timezone.utc) - return { - "ticker": ticker.upper(), - "price": float(price) if price is not None else None, - "regular_price": float(regular) if regular is not None else None, - "pre_market_price": float(pre) if pre is not None else None, - "post_market_price": float(post) if post is not None else None, - "currency": currency, - "exchange": exchange, - "market_state": market_state, - "timestamp": ts, - "source": DataSource.YAHOO_FINANCE, - "delayed": True, - } - except Exception as e: - logger.error(f"Error fetching quote for {ticker}: {str(e)}") - raise + pre_val = float(pre) if pre is not None else None + post_val = float(post) if post is not None else None + regular_val = float(regular) if regular is not None else None + price_val = post_val or pre_val or regular_val + else: + # Fallback: fast_info for price/currency/exchange, history for timestamp + try: + fast = await _run_with_timeout( + loop.run_in_executor(None, lambda: yf_ticker.fast_info), + timeout_seconds=10, + description=f"fast_info {ticker}" + ) + regular_val = getattr(fast, "last_price", None) + if regular_val is None and isinstance(fast, dict): + regular_val = fast.get("lastPrice") or fast.get("last_price") + currency = getattr(fast, "currency", None) or (fast.get("currency") if isinstance(fast, dict) else None) + exchange = getattr(fast, "exchange", None) or (fast.get("exchange") if isinstance(fast, dict) else None) + except Exception as e2: + logger.error(f"get_quote fast_info also failed for {ticker}: {e2}") + raise + + # Try to get last close from recent history for timestamp + try: + df = await _run_with_timeout( + loop.run_in_executor( + None, + lambda: yf_ticker.history(period="2d", interval="1d", auto_adjust=True) + ), + timeout_seconds=15, + description=f"history fallback {ticker}" + ) + if not df.empty: + last_row = df.iloc[-1] + if regular_val is None: + regular_val = float(last_row.get("Close", 0)) or None + ts = df.index[-1].to_pydatetime() + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + else: + ts = datetime.now(timezone.utc) + except Exception: + ts = datetime.now(timezone.utc) + + pre_val = None + post_val = None + price_val = regular_val + market_state = None + + return { + "ticker": ticker.upper(), + "price": float(price_val) if price_val is not None else None, + "regular_price": float(regular_val) if regular_val is not None else None, + "pre_market_price": pre_val, + "post_market_price": post_val, + "currency": currency, + "exchange": exchange, + "market_state": market_state, + "timestamp": ts, + "source": DataSource.YAHOO_FINANCE, + "delayed": True, + } async def get_intraday(self, ticker: str, interval: str = "1m", period: str = "1d") -> List[Dict]: """Get intraday candles using yfinance-plus history with period/interval.""" @@ -487,6 +540,15 @@ class PriceDataService: price_date = date_idx.to_pydatetime() if price_date.tzinfo is None: price_date = price_date.replace(tzinfo=timezone.utc) + # Normalize to UTC midnight so uq_price_data(ticker, date) deduplicates + # correctly regardless of whether data came from yf.Ticker().history() + # (returns Eastern midnight = UTC 04:00) or yf.download() (returns UTC 00:00). + price_date = price_date.replace(hour=0, minute=0, second=0, microsecond=0, + tzinfo=timezone.utc) + + close_val = _safe(row.get('Close')) + if close_val is None: # Skip rows with no valid close price + continue target_rows = live_rows if price_date.date() >= today else historical_rows target_rows.append({ @@ -496,9 +558,9 @@ class PriceDataService: 'open': _safe(row.get('Open')), 'high': _safe(row.get('High')), 'low': _safe(row.get('Low')), - 'close': _safe(row.get('Close')) or 0.0, + 'close': close_val, 'volume': _safe(row.get('Volume')), - 'adjusted_close': _safe(row.get('Close')), + 'adjusted_close': close_val, 'data_source': DataSource.YAHOO_FINANCE.value, 'created_at': now, 'updated_at': now, @@ -507,13 +569,23 @@ class PriceDataService: if not historical_rows and not live_rows: return - # Yahoo-adjusted historical prices are not point-in-time stable: future - # dividends/splits can rewrite old OHLC values. Preserve existing - # historical rows and only update today's row, where intraday partials - # legitimately need EOD replacement. + # Historical rows: preserve valid data but overwrite null/zero-close garbage + # (yf.download MultiIndex parse failures leave close=0 rows that must self-heal). if historical_rows: stmt = pg_insert(PriceData).values(historical_rows) - stmt = stmt.on_conflict_do_nothing(constraint='uq_price_data') + stmt = stmt.on_conflict_do_update( + constraint='uq_price_data', + set_={ + 'open': stmt.excluded.open, + 'high': stmt.excluded.high, + 'low': stmt.excluded.low, + 'close': stmt.excluded.close, + 'volume': stmt.excluded.volume, + 'adjusted_close': stmt.excluded.adjusted_close, + 'updated_at': stmt.excluded.updated_at, + }, + where=or_(PriceData.close.is_(None), PriceData.close == 0.0), + ) await db.execute(stmt) if live_rows: @@ -794,9 +866,9 @@ class PriceDataService: logger.debug(f"_batch_check_missing_periods: end_date includes today — forcing re-fetch for all {len(tickers)} tickers") return list(tickers) - # Single query to check all tickers at once + # Single query to check all tickers at once — count only valid rows from sqlalchemy import func, case - + result = await db.execute( select( PriceData.ticker, @@ -808,7 +880,9 @@ class PriceDataService: and_( PriceData.ticker.in_(tickers), PriceData.date >= start_date, - PriceData.date <= end_date + PriceData.date <= end_date, + PriceData.close.is_not(None), + PriceData.close > 0, ) ) .group_by(PriceData.ticker) @@ -918,8 +992,15 @@ class PriceDataService: # Handle different data structures from yfinance bulk download if len(tickers) == 1: - # Single ticker - data is a simple DataFrame - await self._store_ticker_data(db, tickers[0], bulk_data, interval) + # yf.download with group_by='ticker' returns MultiIndex columns even for a + # single ticker: [('SGOV', 'Open'), ('SGOV', 'Close'), ...]. Extract the + # ticker slice so _store_ticker_data receives a flat DataFrame. + ticker = tickers[0] + if hasattr(bulk_data.columns, 'levels') and ticker in bulk_data.columns.get_level_values(0): + ticker_data = bulk_data[ticker] + else: + ticker_data = bulk_data + await self._store_ticker_data(db, ticker, ticker_data, interval) else: # Multiple tickers - data is grouped by ticker for ticker in tickers: @@ -960,6 +1041,13 @@ class PriceDataService: price_date = date_idx.to_pydatetime() if price_date.tzinfo is None: price_date = price_date.replace(tzinfo=timezone.utc) + # Normalize to UTC midnight (same logic as _store_price_data) + price_date = price_date.replace(hour=0, minute=0, second=0, microsecond=0, + tzinfo=timezone.utc) + + close_val = _safe(row.get('Close')) + if close_val is None: # Skip rows with no valid close price + continue target_rows = live_rows if price_date.date() >= today else historical_rows target_rows.append({ @@ -969,9 +1057,9 @@ class PriceDataService: 'open': _safe(row.get('Open')), 'high': _safe(row.get('High')), 'low': _safe(row.get('Low')), - 'close': _safe(row.get('Close')) or 0.0, + 'close': close_val, 'volume': _safe(row.get('Volume')), - 'adjusted_close': _safe(row.get('Close')), + 'adjusted_close': close_val, 'data_source': DataSource.YAHOO_FINANCE.value, 'created_at': now, 'updated_at': now, @@ -982,7 +1070,19 @@ class PriceDataService: if historical_rows: stmt = pg_insert(PriceData).values(historical_rows) - stmt = stmt.on_conflict_do_nothing(constraint='uq_price_data') + stmt = stmt.on_conflict_do_update( + constraint='uq_price_data', + set_={ + 'open': stmt.excluded.open, + 'high': stmt.excluded.high, + 'low': stmt.excluded.low, + 'close': stmt.excluded.close, + 'volume': stmt.excluded.volume, + 'adjusted_close': stmt.excluded.adjusted_close, + 'updated_at': stmt.excluded.updated_at, + }, + where=or_(PriceData.close.is_(None), PriceData.close == 0.0), + ) await db.execute(stmt) if live_rows: @@ -1002,7 +1102,7 @@ class PriceDataService: ) await db.execute(stmt) logger.info( - "Stored price records for %s: historical_insert_only=%d live_upsert=%d", + "Stored price records for %s: historical=%d live=%d", ticker, len(historical_rows), len(live_rows),