perf: API 성능 개선 Round 3 — 8개 항목 (P15-P22)

P15: FINRA ingest N+1 쿼리 → chunked batch upsert (on_conflict_do_nothing)
P16: Request log 통계 4개 COUNT 쿼리 → 단일 case() 집계 쿼리
P17: stocks/52-week-gainers (1h), stocks/trending (30m) 캐시 추가
P18: Alpaca bars/data 캐시 TTL None→86400 (과거 불변 데이터)
P19: Request log flush 배치 100→500, 간격 1s→2s
P20: Overlay feature_builder matched_symbols Python 필터 → DB JSONB @> 연산자
P21: with_cache Pydantic 모델 캐시 키에 model_dump_json() 사용
P22: Alpaca price 저장 SELECT+filter → batch upsert (on_conflict_do_nothing)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

app/api/v1/endpoints/alpaca.py

@@ -71,7 +71,7 @@ async def alpaca_status():
     summary="Get Alpaca bars (raw, no DB)",
     description="Fetch historical bars directly from Alpaca without storing in DB.",
 )
-@with_cache(namespace="alpaca:bars", ttl=None, key_params=["ticker", "interval", "start_date", "end_date", "limit"])
+@with_cache(namespace="alpaca:bars", ttl=86400, key_params=["ticker", "interval", "start_date", "end_date", "limit"])
 async def get_alpaca_bars(
     ticker: str,
     response: Response,
@@ -125,7 +125,7 @@ async def get_alpaca_bars(
     - Supports: 1m, 5m, 15m, 1h, 1d, 1w, 1mo intervals
     """,
 )
-@with_cache(namespace="alpaca:data", ttl=None, key_params=["ticker", "interval", "start_date", "end_date"])
+@with_cache(namespace="alpaca:data", ttl=86400, key_params=["ticker", "interval", "start_date", "end_date"])
 async def get_alpaca_price_data(
     ticker: str,
     response: Response,

app/api/v1/endpoints/request_logs.py

@@ -6,7 +6,7 @@ from datetime import datetime, timedelta, timezone
 from typing import List, Optional
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import select, desc, and_, or_, func
+from sqlalchemy import select, desc, and_, or_, func, case
 from sqlalchemy.orm import selectinload
 
 from app.core.database import get_db
@@ -186,35 +186,20 @@ async def get_request_stats(
         RequestLog.created_at <= end_date
     )
     
-    # Get total count
-    total_result = await db.execute(
-        select(func.count()).select_from(RequestLog).where(date_filter)
-    )
-    total_requests = total_result.scalar()
-    
-    # Get success count (2xx status codes)
-    success_result = await db.execute(
-        select(func.count()).select_from(RequestLog).where(
-            and_(date_filter, RequestLog.status_code.between(200, 299))
-        )
-    )
-    success_requests = success_result.scalar()
-    
-    # Get client error count (4xx status codes)
-    client_error_result = await db.execute(
-        select(func.count()).select_from(RequestLog).where(
-            and_(date_filter, RequestLog.status_code.between(400, 499))
-        )
-    )
-    client_error_requests = client_error_result.scalar()
-    
-    # Get server error count (5xx status codes)
-    server_error_result = await db.execute(
-        select(func.count()).select_from(RequestLog).where(
-            and_(date_filter, RequestLog.status_code.between(500, 599))
-        )
+    # Single query: total + success + client_error + server_error counts
+    counts_result = await db.execute(
+        select(
+            func.count().label('total'),
+            func.sum(case((RequestLog.status_code.between(200, 299), 1), else_=0)).label('success'),
+            func.sum(case((RequestLog.status_code.between(400, 499), 1), else_=0)).label('client_error'),
+            func.sum(case((RequestLog.status_code.between(500, 599), 1), else_=0)).label('server_error'),
+        ).select_from(RequestLog).where(date_filter)
     )
-    server_error_requests = server_error_result.scalar()
+    counts = counts_result.one()
+    total_requests = counts.total or 0
+    success_requests = counts.success or 0
+    client_error_requests = counts.client_error or 0
+    server_error_requests = counts.server_error or 0
     
     # Get requests by method
     method_result = await db.execute(

app/api/v1/endpoints/stocks.py

@@ -185,9 +185,12 @@ async def get_most_active_stocks(
 
 
 @router.get("/52-week-gainers", summary="Top 52-week gaining stocks")
+@with_cache(namespace="stocks:52-week-gainers", ttl=3600, key_params=["limit", "max_pages"])
 async def get_52week_gainers(
+    response: Response,
     limit: Optional[int] = Query(None, ge=1, le=1000, description="Maximum number of stocks to return (1-1000). If not specified, returns first 600 stocks (3 pages) for performance."),
-    max_pages: Optional[int] = Query(3, ge=1, le=10, description="Maximum pages to fetch (1-10). Each page has ~200 stocks. Higher values may cause rate limiting.")
+    max_pages: Optional[int] = Query(3, ge=1, le=10, description="Maximum pages to fetch (1-10). Each page has ~200 stocks. Higher values may cause rate limiting."),
+    force_refresh: bool = Query(False, description="Bypass cache"),
 ):
     """
     Get 52-week top gaining stocks from Yahoo Finance
@@ -281,10 +284,13 @@ async def get_52week_gainers(
 
 
 @router.get("/trending", summary="Trending stocks combining most active and 52-week gainers")
+@with_cache(namespace="stocks:trending", ttl=1800, key_params=["n", "most_active_limit", "gainers_limit"])
 async def get_trending_stocks(
+    response: Response,
     n: Optional[int] = Query(500, ge=1, description="Total number of trending stocks to return after combining most active + gainers (default: 500)"),
     most_active_limit: Optional[int] = Query(None, ge=1, description="Number of most active stocks to include. If not specified, returns all available stocks (~170)."),
-    gainers_limit: Optional[int] = Query(None, ge=1, description="Number of 52-week gainers to fetch. If not specified, fetches enough to reach target 'n' after combining with most active.")
+    gainers_limit: Optional[int] = Query(None, ge=1, description="Number of 52-week gainers to fetch. If not specified, fetches enough to reach target 'n' after combining with most active."),
+    force_refresh: bool = Query(False, description="Bypass cache"),
 ):
     """
     Get trending stocks combining most active and 52-week gainers

app/middleware/error_logger.py

@@ -36,8 +36,8 @@ 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
+_FLUSH_INTERVAL_SECONDS: float = 2.0   # flush at most every 2 s
+_FLUSH_BATCH_SIZE: int = 500           # or when 500 entries are queued
 
 
 async def _flush_request_logs(entries: list) -> None:

app/services/alpaca_price_service.py

@@ -8,6 +8,7 @@ from typing import Dict, List, Optional
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select, and_
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from app.models.alpaca_price import AlpacaPriceData
 from app.services.alpaca_client import AlpacaClient
@@ -53,36 +54,34 @@ class AlpacaPriceService:
             logger.warning(f"Alpaca returned 0 bars for {ticker}")
             return 0
 
-        # Fetch existing dates for dedup
-        result = await db.execute(
-            select(AlpacaPriceData.date).where(AlpacaPriceData.ticker == ticker)
-        )
-        existing_dates = {row[0].date() for row in result.fetchall()}
-
-        inserted = 0
+        # Build rows for batch insert
+        rows = []
         for bar in bars:
             bar_dt = _parse_bar_timestamp(bar["t"])
-            if bar_dt.date() in existing_dates:
-                continue
-
-            record = AlpacaPriceData(
-                ticker=ticker,
-                date=bar_dt,
-                open=float(bar.get("o", 0)),
-                high=float(bar.get("h", 0)),
-                low=float(bar.get("l", 0)),
-                close=float(bar.get("c", 0)),
-                volume=float(bar.get("v", 0)),
-                vwap=float(bar["vw"]) if bar.get("vw") else None,
-                trade_count=int(bar["n"]) if bar.get("n") else None,
-                data_source="ALPACA",
-            )
-            db.add(record)
-            existing_dates.add(bar_dt.date())
-            inserted += 1
+            rows.append({
+                "ticker": ticker,
+                "date": bar_dt,
+                "open": float(bar.get("o", 0)),
+                "high": float(bar.get("h", 0)),
+                "low": float(bar.get("l", 0)),
+                "close": float(bar.get("c", 0)),
+                "volume": float(bar.get("v", 0)),
+                "vwap": float(bar["vw"]) if bar.get("vw") else None,
+                "trade_count": int(bar["n"]) if bar.get("n") else None,
+                "data_source": "ALPACA",
+            })
+
+        if not rows:
+            return 0
+
+        # Batch insert — skip duplicates via ON CONFLICT DO NOTHING
+        stmt = pg_insert(AlpacaPriceData).values(rows)
+        stmt = stmt.on_conflict_do_nothing(constraint='uq_alpaca_price_data')
+        result = await db.execute(stmt)
+        await db.commit()
 
+        inserted = result.rowcount
         if inserted:
-            await db.commit()
             logger.info(f"Alpaca: inserted {inserted} bars for {ticker}")
 
         return inserted

app/services/finra_short_volume_service.py

@@ -9,6 +9,7 @@ from typing import Dict, List, Optional, Tuple
 import aiohttp
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select, and_, func, desc
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from app.core.http_client import get_http_session
 
@@ -128,26 +129,18 @@ class FinraShortVolumeService:
         if not records:
             return 0
 
-        # Bulk insert – skip duplicates
+        # Batch insert in chunks — asyncpg has a 32767 bind-parameter limit
+        CHUNK_SIZE = 3000
         inserted = 0
-        for rec in records:
-            existing = await db.execute(
-                select(FinraShortVolume.id).where(
-                    and_(
-                        FinraShortVolume.symbol == rec["symbol"],
-                        FinraShortVolume.date == rec["date"],
-                        FinraShortVolume.market == rec["market"],
-                    )
-                )
-            )
-            if existing.first():
-                continue
-
-            db.add(FinraShortVolume(**rec))
-            inserted += 1
+        for i in range(0, len(records), CHUNK_SIZE):
+            chunk = records[i:i + CHUNK_SIZE]
+            stmt = pg_insert(FinraShortVolume).values(chunk)
+            stmt = stmt.on_conflict_do_nothing(constraint='uq_finra_short_volume')
+            result = await db.execute(stmt)
+            inserted += result.rowcount
+        await db.commit()
 
         if inserted:
-            await db.commit()
             logger.info(f"FINRA: ingested {inserted} records for {target_date}")
 
         return inserted

app/services/overlay/feature_builder.py

@@ -8,7 +8,8 @@ from datetime import datetime, timedelta, timezone
 from typing import Dict, List, Optional
 
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import select, and_, func
+from sqlalchemy import select, and_, func, cast
+from sqlalchemy.dialects.postgresql import JSONB
 
 from app.models.overlay_raw_event import (
     OverlayHeadlineEvent,
@@ -72,23 +73,22 @@ class FeatureBuilder:
         cutoff_6h = as_of - timedelta(hours=6)
         window_cutoff = as_of - timedelta(days=ZSCORE_WINDOW_DAYS)
 
-        # Fetch all headlines in z-score window (we need matched_symbols for filtering)
+        # Fetch headlines matching symbol in z-score window (DB-level JSON filter)
+        symbol_json = cast([symbol], JSONB)
         result = await db.execute(
             select(
                 OverlayHeadlineEvent.publisher,
                 OverlayHeadlineEvent.published_at,
-                OverlayHeadlineEvent.matched_symbols,
             ).where(
                 and_(
                     OverlayHeadlineEvent.published_at >= window_cutoff,
                     OverlayHeadlineEvent.published_at <= as_of,
+                    cast(OverlayHeadlineEvent.matched_symbols, JSONB).op('@>')(symbol_json),
                 )
             )
         )
-        all_rows = result.fetchall()
+        sym_rows_hist = result.fetchall()
 
-        # Filter by symbol
-        sym_rows_hist = [r for r in all_rows if symbol in (r.matched_symbols or [])]
         sym_rows_24h = [r for r in sym_rows_hist if r.published_at >= cutoff_24h]
         sym_rows_6h = [r for r in sym_rows_24h if r.published_at >= cutoff_6h]
 
@@ -123,21 +123,21 @@ class FeatureBuilder:
         cutoff_24h = as_of - timedelta(hours=24)
         window_cutoff = as_of - timedelta(days=ZSCORE_WINDOW_DAYS)
 
+        symbol_json = cast([symbol], JSONB)
         result = await db.execute(
             select(
                 OverlayVideoEvent.view_count,
                 OverlayVideoEvent.channel_weight,
                 OverlayVideoEvent.published_at,
-                OverlayVideoEvent.matched_symbols,
             ).where(
                 and_(
                     OverlayVideoEvent.published_at >= window_cutoff,
                     OverlayVideoEvent.published_at <= as_of,
+                    cast(OverlayVideoEvent.matched_symbols, JSONB).op('@>')(symbol_json),
                 )
             )
         )
-        all_rows = result.fetchall()
-        sym_rows = [r for r in all_rows if symbol in (r.matched_symbols or [])]
+        sym_rows = result.fetchall()
         sym_rows_24h = [r for r in sym_rows if r.published_at >= cutoff_24h]
 
         mentions_24h = len(sym_rows_24h)

app/utils/cache.py

@@ -179,7 +179,11 @@ def with_cache(namespace: str, ttl: int = None, key_params: list[str] = None):
             for p in (key_params or []):
                 val = arguments.get(p)
                 if val is not None:
-                    parts.append(str(val))
+                    # Use deterministic JSON for Pydantic models instead of str()
+                    if hasattr(val, 'model_dump_json'):
+                        parts.append(val.model_dump_json())
+                    else:
+                        parts.append(str(val))
             cache_key = build_cache_key(namespace, *parts)
 
             # Check force_refresh