diff --git a/app/api/v1/endpoints/alpaca.py b/app/api/v1/endpoints/alpaca.py index 63119ed..ea0a7e7 100644 --- a/app/api/v1/endpoints/alpaca.py +++ b/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, diff --git a/app/api/v1/endpoints/request_logs.py b/app/api/v1/endpoints/request_logs.py index 7cd878b..7077193 100644 --- a/app/api/v1/endpoints/request_logs.py +++ b/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( diff --git a/app/api/v1/endpoints/stocks.py b/app/api/v1/endpoints/stocks.py index 9df8eb3..cbf2459 100644 --- a/app/api/v1/endpoints/stocks.py +++ b/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 diff --git a/app/middleware/error_logger.py b/app/middleware/error_logger.py index c578961..de61a24 100644 --- a/app/middleware/error_logger.py +++ b/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: diff --git a/app/services/alpaca_price_service.py b/app/services/alpaca_price_service.py index 6f07027..600bfa1 100644 --- a/app/services/alpaca_price_service.py +++ b/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 diff --git a/app/services/finra_short_volume_service.py b/app/services/finra_short_volume_service.py index 4ae7b4c..1aa0dff 100644 --- a/app/services/finra_short_volume_service.py +++ b/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 diff --git a/app/services/overlay/feature_builder.py b/app/services/overlay/feature_builder.py index e16760a..87fd0e3 100644 --- a/app/services/overlay/feature_builder.py +++ b/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) diff --git a/app/utils/cache.py b/app/utils/cache.py index c5dce35..752ecd6 100644 --- a/app/utils/cache.py +++ b/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