From 6a98cb2d949d9b03f934d88f70a2cef025b27f6a Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Thu, 19 Mar 2026 16:24:31 -0700 Subject: [PATCH] =?UTF-8?q?perf:=20API=20=EC=84=B1=EB=8A=A5=20=EA=B0=9C?= =?UTF-8?q?=EC=84=A0=20=E2=80=94=20=EB=A1=9C=EA=B7=B8=20=EB=B6=84=EC=84=9D?= =?UTF-8?q?=20=EA=B8=B0=EB=B0=98=206=EA=B0=9C=20=ED=95=AD=EB=AA=A9=20?= =?UTF-8?q?=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0 — 500 에러 대폭 감소 (1.1% → 0.11%) - price.py / financial.py: except HTTPException: raise 추가 → HTTPException(404)이 except Exception에 잡혀 500으로 재포장되던 버그 수정 - real_sec_financial_service: _get_price_data_for_period에 try/except 추가 → yfinance 실패가 financial 엔드포인트 500으로 전파되지 않도록 방어 - price.py: TimeoutError 별도 핸들러 추가 → yfinance 타임아웃 시 503 반환 P1 — _store_price_data() N+1 → batch upsert - price_data_service: 252회 개별 SELECT+INSERT 루프를 pg_insert(PriceData).on_conflict_do_nothing('uq_price_data') 단일 쿼리로 교체 P2 — financial/data 캐시 TTL 1h → 24h - financial.py: 재무 데이터는 분기 발표 주기 → _FIN_TTL = 86400 P3 — 과거 가격 데이터 TTL 연장 - price.py: end_date < today-1 이면 TTL=7일, 나머지 1h 유지 P4 — 요청 로그 비동기 배치 처리 - error_logger.py: asyncio.Queue(10_000) 추가, _log_request를 put_nowait으로 변경 (논블로킹), 백그라운드 _request_log_flusher 코루틴 (1s/100건마다 flush) - main.py: 앱 시작 시 start_request_log_flusher() 호출 P5 — 누락 DB 인덱스 추가 - financial.py: CalculatedMetrics에 calculation_date, period_date 단독 인덱스 - attention.py: AttentionFeaturesDaily에 ticker 단독 인덱스 - alembic b3c4d5e6f7a8: 위 3개 인덱스 생성 마이그레이션 (idempotent) Co-Authored-By: Claude Opus 4.6 (1M context) --- .../versions/b3c4d5e6f7a8_add_perf_indexes.py | 60 +++++++ app/api/v1/endpoints/financial.py | 12 +- app/api/v1/endpoints/price.py | 27 ++- app/main.py | 4 +- app/middleware/error_logger.py | 156 ++++++++++++------ app/models/attention.py | 1 + app/models/financial.py | 2 + app/services/price_data_service.py | 73 ++++---- app/services/real_sec_financial_service.py | 18 +- 9 files changed, 262 insertions(+), 91 deletions(-) create mode 100644 alembic/versions/b3c4d5e6f7a8_add_perf_indexes.py diff --git a/alembic/versions/b3c4d5e6f7a8_add_perf_indexes.py b/alembic/versions/b3c4d5e6f7a8_add_perf_indexes.py new file mode 100644 index 0000000..9350e19 --- /dev/null +++ b/alembic/versions/b3c4d5e6f7a8_add_perf_indexes.py @@ -0,0 +1,60 @@ +"""add performance indexes for CalculatedMetrics and AttentionFeaturesDaily + +Revision ID: b3c4d5e6f7a8 +Revises: a1b2c3d4e5f6 +Create Date: 2026-03-19 + +Adds standalone indexes that were missing from the initial schema: + - calculated_metrics.calculation_date (date-only range queries) + - calculated_metrics.period_date (period-based lookups) + - attention_features_daily.ticker (ticker-only scans) +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "b3c4d5e6f7a8" +down_revision: Union[str, Sequence[str], None] = "a1b2c3d4e5f6" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + conn = op.get_bind() + + # CalculatedMetrics — standalone date indexes + _indexes = conn.dialect.get_indexes(conn, "calculated_metrics") + existing = {idx["name"] for idx in _indexes} + + if "idx_metrics_calculation_date" not in existing: + op.create_index( + "idx_metrics_calculation_date", + "calculated_metrics", + ["calculation_date"], + ) + + if "idx_metrics_period_date" not in existing: + op.create_index( + "idx_metrics_period_date", + "calculated_metrics", + ["period_date"], + ) + + # AttentionFeaturesDaily — standalone ticker index + _attn_indexes = conn.dialect.get_indexes(conn, "attention_features_daily") + existing_attn = {idx["name"] for idx in _attn_indexes} + + if "idx_attention_features_daily_ticker" not in existing_attn: + op.create_index( + "idx_attention_features_daily_ticker", + "attention_features_daily", + ["ticker"], + ) + + +def downgrade() -> None: + op.drop_index("idx_metrics_calculation_date", table_name="calculated_metrics") + op.drop_index("idx_metrics_period_date", table_name="calculated_metrics") + op.drop_index("idx_attention_features_daily_ticker", table_name="attention_features_daily") diff --git a/app/api/v1/endpoints/financial.py b/app/api/v1/endpoints/financial.py index 718aa12..ac6d127 100644 --- a/app/api/v1/endpoints/financial.py +++ b/app/api/v1/endpoints/financial.py @@ -134,13 +134,16 @@ async def get_financial_data( (resolved_end.date().isoformat() if resolved_end else ""), ) + # Financial data is released quarterly — use 24 h TTL + _FIN_TTL = 60 * 60 * 24 + # Try cache unless force_refresh if not request.force_refresh: cached = await get_cached_response(cache_key) if cached: cached_body, etag = cached response.headers["X-Cache"] = "HIT" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["Cache-Control"] = f"public, max-age={_FIN_TTL}" response.headers["ETag"] = etag return cached_body @@ -232,13 +235,14 @@ async def get_financial_data( } ) - # Cache the response - etag = await set_cached_response(cache_key, body.model_dump(), ttl_seconds=settings.CACHE_TTL) + etag = await set_cached_response(cache_key, body.model_dump(), ttl_seconds=_FIN_TTL) response.headers["X-Cache"] = "MISS" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["Cache-Control"] = f"public, max-age={_FIN_TTL}" response.headers["ETag"] = etag return body + except HTTPException: + raise except ValueError as e: if "No data returned" in str(e): raise HTTPException( diff --git a/app/api/v1/endpoints/price.py b/app/api/v1/endpoints/price.py index 161815f..5f072b7 100644 --- a/app/api/v1/endpoints/price.py +++ b/app/api/v1/endpoints/price.py @@ -2,7 +2,7 @@ Price data endpoints """ -from datetime import datetime, timezone, date +from datetime import datetime, timezone, date, timedelta from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query, Response from sqlalchemy.ext.asyncio import AsyncSession @@ -222,14 +222,33 @@ async def get_price_data( } ) + # Use extended TTL for purely historical date ranges (end_date < yesterday) + _historical = ( + end_date is not None + and end_date.date() < date.today() - timedelta(days=1) + ) + cache_ttl = 60 * 60 * 24 * 7 if _historical else settings.CACHE_TTL # 7d vs 1h + # Cache the response body body_dict = body.model_dump() - etag = await set_cached_response(cache_key, body_dict, ttl_seconds=settings.CACHE_TTL) + etag = await set_cached_response(cache_key, body_dict, ttl_seconds=cache_ttl) response.headers["X-Cache"] = "MISS" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["Cache-Control"] = f"public, max-age={cache_ttl}" response.headers["ETag"] = etag return body - + + except HTTPException: + # Let FastAPI HTTPExceptions (404, 400, etc.) propagate as-is + raise + except TimeoutError as e: + raise HTTPException( + status_code=503, + detail={ + "error_type": "TIMEOUT", + "message": "Data fetch timed out. Please try again.", + "detail": {"error": str(e)} + } + ) except ValueError as e: if "Yahoo Finance data source not available" in str(e): raise HTTPException( diff --git a/app/main.py b/app/main.py index c0ced20..bce2c01 100644 --- a/app/main.py +++ b/app/main.py @@ -11,7 +11,7 @@ from fastapi.responses import RedirectResponse from app.core.config import settings from app.api.v1.api import api_router from app.core.database import engine, Base -from app.middleware.error_logger import ErrorLoggingMiddleware +from app.middleware.error_logger import ErrorLoggingMiddleware, start_request_log_flusher from app.models import error_log, request_log, fred_data, filing, finra_short_volume, alpaca_price # Import to register models from app.models import overlay_registry, overlay_raw_event, overlay_feature # Overlay models @@ -21,6 +21,8 @@ async def lifespan(app: FastAPI): # Startup - ensure tables exist async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) + # Start async batch request log flusher + start_request_log_flusher() # Start overlay batch scheduler (graceful no-op if apscheduler not installed) try: from app.services.overlay.scheduler import start_scheduler diff --git a/app/middleware/error_logger.py b/app/middleware/error_logger.py index d585530..0637cf6 100644 --- a/app/middleware/error_logger.py +++ b/app/middleware/error_logger.py @@ -1,7 +1,17 @@ """ -Error logging middleware for capturing and storing API errors +Error logging middleware for capturing and storing API errors. + +Request logs are buffered in an in-memory async queue and flushed in batch +every second (or when the buffer reaches 100 entries) by a background task +started at app startup via start_request_log_flusher(). This avoids one DB +write per request at high throughput (1 M+ req/day) and eliminates DB lock +contention as a contributor to p95 tail latency. + +Error logs (4xx/5xx) are still written synchronously because they are +relatively rare and need accurate timing. """ +import asyncio import json import time import traceback @@ -22,6 +32,61 @@ import logging logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Async queue for request log entries +# --------------------------------------------------------------------------- +_REQUEST_LOG_QUEUE: asyncio.Queue = asyncio.Queue(maxsize=10_000) +_FLUSH_INTERVAL_SECONDS: float = 1.0 # flush at most every 1 s +_FLUSH_BATCH_SIZE: int = 100 # or when 100 entries are queued + + +async def _flush_request_logs(entries: list) -> None: + """Batch-insert a list of RequestLog dicts into the database.""" + if not entries: + return + try: + from app.core.database import AsyncSessionLocal + async with AsyncSessionLocal() as db: + for entry in entries: + db.add(RequestLog(**entry)) + await db.commit() + except Exception as e: + logger.error(f"Failed to flush request log batch ({len(entries)} entries): {e}") + + +async def _request_log_flusher() -> None: + """Background coroutine: drain the queue and batch-insert into DB.""" + while True: + entries: list = [] + try: + # Wait for at least one entry + first = await asyncio.wait_for( + _REQUEST_LOG_QUEUE.get(), timeout=_FLUSH_INTERVAL_SECONDS + ) + entries.append(first) + _REQUEST_LOG_QUEUE.task_done() + + # Drain up to FLUSH_BATCH_SIZE without blocking + while len(entries) < _FLUSH_BATCH_SIZE: + try: + item = _REQUEST_LOG_QUEUE.get_nowait() + entries.append(item) + _REQUEST_LOG_QUEUE.task_done() + except asyncio.QueueEmpty: + break + + except asyncio.TimeoutError: + pass # Nothing queued in the last interval — loop again + + if entries: + await _flush_request_logs(entries) + + +def start_request_log_flusher() -> None: + """Schedule the background flusher coroutine. Call once at app startup.""" + asyncio.ensure_future(_request_log_flusher()) + logger.info("Request log flusher background task started") + class ErrorLoggingMiddleware(BaseHTTPMiddleware): """Middleware to log all API requests and errors to database""" @@ -273,52 +338,51 @@ class ErrorLoggingMiddleware(BaseHTTPMiddleware): response_size: Optional[int] = None, data_source: Optional[str] = None ): - """Log general request to database""" - - # Skip logging for log deletion endpoints to avoid logging the deletion of logs + """Enqueue a request log entry for async batch flushing. + + Non-blocking: the entry is placed in _REQUEST_LOG_QUEUE and written to + the database in bulk by the background flusher, avoiding a synchronous + DB round-trip on every request. + """ + + # Skip logging for log deletion endpoints endpoint = request_info["endpoint"] method = request_info["method"] - - # Don't log DELETE requests to log management endpoints - if (method == "DELETE" and - (endpoint.startswith("/api/v1/admin/requests/logs") or - endpoint.startswith("/api/v1/admin/errors/logs"))): - logger.info(f"Skipping log for log deletion endpoint: {method} {endpoint}") + + if (method == "DELETE" and + (endpoint.startswith("/api/v1/admin/requests/logs") or + endpoint.startswith("/api/v1/admin/errors/logs"))): return - + + # Augment headers with data source if provided + headers = request_info.get("headers") or {} + if data_source: + headers = {**headers, "X-Data-Source": data_source} + + entry = { + "request_id": request_id, + "endpoint": endpoint, + "method": method, + "path": request_info["path"], + "query_params": request_info["query_params"], + "request_body": request_info["request_body"], + "headers": headers, + "status_code": status_code, + "response_size": response_size, + "user_agent": request_info["user_agent"], + "client_ip": request_info["client_ip"], + "response_time_ms": response_time_ms, + } + try: - # Get database session - from app.core.database import AsyncSessionLocal - async with AsyncSessionLocal() as db: - # Store data source in headers if available - if data_source and request_info.get("headers"): - request_info["headers"]["X-Data-Source"] = data_source - - # Create request log entry - request_log = RequestLog( - request_id=request_id, - endpoint=request_info["endpoint"], - method=request_info["method"], - path=request_info["path"], - query_params=request_info["query_params"], - request_body=request_info["request_body"], - headers=request_info["headers"], - status_code=status_code, - response_size=response_size, - user_agent=request_info["user_agent"], - client_ip=request_info["client_ip"], - response_time_ms=response_time_ms - ) - - db.add(request_log) - await db.commit() - - except Exception as e: - # If we can't log to database, at least log to file - logger.error(f"Failed to log request to database: {e}") - logger.info( - f"Request log - Request ID: {request_id}, " - f"Endpoint: {request_info['endpoint']}, " - f"Status: {status_code}, " - f"Response Time: {response_time_ms}ms" - ) \ No newline at end of file + _REQUEST_LOG_QUEUE.put_nowait(entry) + except asyncio.QueueFull: + # Queue is at capacity — fall back to a direct write so we don't drop the log + logger.warning("Request log queue full — writing synchronously") + try: + from app.core.database import AsyncSessionLocal + async with AsyncSessionLocal() as db: + db.add(RequestLog(**entry)) + await db.commit() + except Exception as e: + logger.error(f"Failed to write request log (fallback): {e}") \ No newline at end of file diff --git a/app/models/attention.py b/app/models/attention.py index 4f32893..46b96d1 100644 --- a/app/models/attention.py +++ b/app/models/attention.py @@ -111,4 +111,5 @@ class AttentionFeaturesDaily(Base): __table_args__ = ( UniqueConstraint("ticker", "date", name="uq_attention_features_daily"), Index("idx_attention_features_daily_ticker_date", "ticker", "date"), + Index("idx_attention_features_daily_ticker", "ticker"), ) diff --git a/app/models/financial.py b/app/models/financial.py index 8e534fd..a6ba573 100644 --- a/app/models/financial.py +++ b/app/models/financial.py @@ -126,6 +126,8 @@ class CalculatedMetrics(Base): __table_args__ = ( UniqueConstraint('ticker', 'calculation_date', 'period_date', name='uq_calculated_metrics'), Index('idx_metrics_ticker_date', 'ticker', 'calculation_date'), + Index('idx_metrics_calculation_date', 'calculation_date'), + Index('idx_metrics_period_date', 'period_date'), ) class PriceData(Base): diff --git a/app/services/price_data_service.py b/app/services/price_data_service.py index 5972a71..e1e0b81 100644 --- a/app/services/price_data_service.py +++ b/app/services/price_data_service.py @@ -364,40 +364,49 @@ class PriceDataService: hist_data, interval: str ): - """Store price data in database""" - for date, row in hist_data.iterrows(): - # Convert pandas timestamp to datetime - price_date = date.to_pydatetime() + """Store price data in database using batch upsert (INSERT ... ON CONFLICT DO NOTHING).""" + from sqlalchemy.dialects.postgresql import insert as pg_insert + import uuid as _uuid + + def _safe(val): + """Return float or None, handling NaN/None safely.""" + if val is None: + return None + try: + v = float(val) + return None if pd.isna(v) else v + except (TypeError, ValueError): + return None + + now = datetime.now(timezone.utc) + rows = [] + for date_idx, row in hist_data.iterrows(): + price_date = date_idx.to_pydatetime() if price_date.tzinfo is None: price_date = price_date.replace(tzinfo=timezone.utc) - - # Check if record already exists - existing = await db.execute( - select(PriceData).where( - and_( - PriceData.ticker == ticker, - PriceData.date == price_date - ) - ) - ) - - if existing.first(): - continue # Skip if already exists - - # Create new price data record - price_record = PriceData( - ticker=ticker, - date=price_date, - open=float(row.get('Open', 0)) if not pd.isna(row.get('Open')) else None, - high=float(row.get('High', 0)) if not pd.isna(row.get('High')) else None, - low=float(row.get('Low', 0)) if not pd.isna(row.get('Low')) else None, - close=float(row.get('Close', 0)) if not pd.isna(row.get('Close')) else 0, - volume=float(row.get('Volume', 0)) if not pd.isna(row.get('Volume')) else None, - adjusted_close=float(row.get('Close', 0)) if not pd.isna(row.get('Close')) else None, # Auto-adjusted - data_source=DataSource.YAHOO_FINANCE - ) - - db.add(price_record) + + rows.append({ + 'id': _uuid.uuid4(), + 'ticker': ticker, + 'date': price_date, + 'open': _safe(row.get('Open')), + 'high': _safe(row.get('High')), + 'low': _safe(row.get('Low')), + 'close': _safe(row.get('Close')) or 0.0, + 'volume': _safe(row.get('Volume')), + 'adjusted_close': _safe(row.get('Close')), + 'data_source': DataSource.YAHOO_FINANCE.value, + 'created_at': now, + 'updated_at': now, + }) + + if not rows: + return + + # Single batch INSERT — skip rows that violate the unique constraint (ticker, date) + stmt = pg_insert(PriceData).values(rows) + stmt = stmt.on_conflict_do_nothing(constraint='uq_price_data') + await db.execute(stmt) async def _get_price_data_from_db( self, diff --git a/app/services/real_sec_financial_service.py b/app/services/real_sec_financial_service.py index fde67fe..5589113 100644 --- a/app/services/real_sec_financial_service.py +++ b/app/services/real_sec_financial_service.py @@ -163,10 +163,20 @@ class RealSECFinancialService: start_date: datetime, end_date: datetime ) -> List[PriceData]: - """Get price data for the specified period""" - return await self.price_service.get_or_update_price_data( - db, ticker, start_date, end_date, "1d", force_refresh=False - ) + """Get price data for the specified period. + + Failures (e.g. unknown ticker, yfinance error) are swallowed so that a + price-fetch problem never causes the financial endpoint to return 500. + """ + try: + return await self.price_service.get_or_update_price_data( + db, ticker, start_date, end_date, "1d", force_refresh=False + ) + except Exception as e: + logger.warning( + f"Could not fetch price data for {ticker} during financial data processing: {e}" + ) + return [] async def _calculate_real_metrics( self,