From 20816118848ad8f54bb43f989dd3045c8cb83c84 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Mon, 13 Apr 2026 11:16:12 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20multi-ticker=20daily=20bars=20=E2=80=94?= =?UTF-8?q?=20DB=20storage=20+=20DB-first=20fetch=20logic?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - AlpacaPriceService.get_or_fetch_multi_bars(): checks DB max_date per ticker, only fetches missing ranges from Alpaca, upserts with chunking (3000 rows/chunk, asyncpg 32767-param limit) then reads back from DB - GET /price/data endpoint: now uses service + Depends(get_db); subsequent calls for same date range skip Alpaca entirely - force_refresh=true bypasses DB check and re-fetches all from Alpaca Co-Authored-By: Claude Sonnet 4.6 --- app/api/v1/endpoints/price.py | 65 +++++++--------- app/services/alpaca_price_service.py | 108 ++++++++++++++++++++++++++- 2 files changed, 134 insertions(+), 39 deletions(-) diff --git a/app/api/v1/endpoints/price.py b/app/api/v1/endpoints/price.py index 9353e48..1f75be2 100644 --- a/app/api/v1/endpoints/price.py +++ b/app/api/v1/endpoints/price.py @@ -23,7 +23,7 @@ from app.schemas.financial import ( TodayOHLCResponse, ) from app.services.price_data_service import PriceDataService -from app.services.alpaca_client import AlpacaClient, normalize_ticker +from app.services.alpaca_price_service import AlpacaPriceService from app.schemas.financial import AlpacaMultiBarsResponse from app.core.config import settings from app.utils.date_utils import quarters_to_date_range @@ -291,13 +291,14 @@ async def get_price_data( @router.get( "/data", response_model=AlpacaMultiBarsResponse, - summary="Get daily bars for multiple tickers via Alpaca", + summary="Get daily bars for multiple tickers via Alpaca (DB-backed)", description=( - "Fetch OHLCV daily bars for up to ~500 tickers in one call using Alpaca's " - "multi-bar endpoint. Data is **not** cached (live trading use-case).\n\n" + "Fetch OHLCV daily bars for up to ~500 tickers. Results are stored in DB so " + "subsequent calls only fetch new/missing dates from Alpaca.\n\n" "- `tickers`: comma-separated list, e.g. `AAPL,MSFT,BF-B`\n" - "- Ticker normalization: `BF-B` → `BF.B` is handled automatically; " - "response keys use the *original* symbol names you passed in.\n" + "- Ticker normalization: `BF-B` → `BF.B` handled automatically; " + "response keys use the original symbol names.\n" + "- `force_refresh=true`: re-fetch all from Alpaca regardless of DB state.\n" "- Requires `ALPACA_API_KEY` / `ALPACA_SECRET_KEY`." ), tags=["price", "alpaca"], @@ -307,56 +308,46 @@ async def get_multi_ticker_daily_bars( start_date: date = Query(..., description="Start date (YYYY-MM-DD)"), end_date: date = Query(..., description="End date (YYYY-MM-DD)"), interval: str = Query("1d", description="Bar interval: 1d, 1w, 1mo"), - force_refresh: bool = Query(False, description="Ignored — no caching for this endpoint"), + force_refresh: bool = Query(False, description="Re-fetch from Alpaca even if DB has data"), + db: AsyncSession = Depends(get_db), ): - """Multi-ticker daily bars via Alpaca (ORB engine interface).""" + """Multi-ticker daily bars via Alpaca with DB storage (ORB engine interface).""" symbols = [s.strip().upper() for s in tickers.split(",") if s.strip()] if not symbols: raise HTTPException(status_code=400, detail="No tickers provided.") if len(symbols) > 1000: raise HTTPException(status_code=400, detail="Maximum 1000 tickers per request.") - client = AlpacaClient() - if not client.is_configured(): + svc = AlpacaPriceService() + if not svc.is_available(): raise HTTPException(status_code=503, detail="Alpaca API keys not configured.") - # Build reverse-map: normalized_alpaca_symbol → original_input_symbol - reverse_map = {normalize_ticker(s): s for s in symbols} - - start_str = start_date.isoformat() - end_str = end_date.isoformat() + start_dt = datetime.combine(start_date, datetime.min.time()).replace(tzinfo=timezone.utc) + end_dt = datetime.combine(end_date, datetime.min.time()).replace(tzinfo=timezone.utc) try: - raw = await client.get_multi_bars( - symbols=symbols, - timeframe=interval, - start=start_str, - end=end_str, + data = await svc.get_or_fetch_multi_bars( + db, symbols, start_dt, end_dt, interval, force_refresh ) except Exception as e: raise HTTPException(status_code=502, detail=f"Alpaca API error: {e}") finally: - await client.close() + await svc.client.close() - # Re-key results back to original input symbols and reshape bar dicts - bars: dict = {} - for alpaca_sym, bar_list in raw.items(): - original_sym = reverse_map.get(alpaca_sym, alpaca_sym) - bars[original_sym] = [ + bars = { + ticker: [ { - "date": b["t"][:10], # "2026-02-07T05:00:00Z" → "2026-02-07" - "open": b.get("o"), - "high": b.get("h"), - "low": b.get("l"), - "close": b.get("c"), - "volume": b.get("v"), + "date": row.date.date().isoformat(), + "open": row.open, + "high": row.high, + "low": row.low, + "close": row.close, + "volume": row.volume, } - for b in bar_list + for row in rows ] - - # Include symbols with no data as empty lists - for sym in symbols: - bars.setdefault(sym, []) + for ticker, rows in data.items() + } return AlpacaMultiBarsResponse( interval=interval, diff --git a/app/services/alpaca_price_service.py b/app/services/alpaca_price_service.py index 600bfa1..23c4777 100644 --- a/app/services/alpaca_price_service.py +++ b/app/services/alpaca_price_service.py @@ -7,11 +7,11 @@ from datetime import datetime, timezone from typing import Dict, List, Optional from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy import select, and_ +from sqlalchemy import select, and_, func from sqlalchemy.dialects.postgresql import insert as pg_insert from app.models.alpaca_price import AlpacaPriceData -from app.services.alpaca_client import AlpacaClient +from app.services.alpaca_client import AlpacaClient, normalize_ticker logger = logging.getLogger(__name__) @@ -86,6 +86,110 @@ class AlpacaPriceService: return inserted + async def get_or_fetch_multi_bars( + self, + db: AsyncSession, + tickers: List[str], + start_dt: datetime, + end_dt: datetime, + interval: str = "1d", + force_refresh: bool = False, + ) -> Dict[str, List[AlpacaPriceData]]: + """ + DB-first multi-ticker daily bars. + + 1. Check which tickers are missing data (max_date < end_dt) in DB. + 2. Fetch only missing tickers from Alpaca and upsert. + 3. Read all data from DB and return as Dict[ticker → rows]. + """ + upper_tickers = [t.upper() for t in tickers] + + if not force_refresh: + result = await db.execute( + select( + AlpacaPriceData.ticker, + func.max(AlpacaPriceData.date).label("max_date"), + ) + .where( + and_( + AlpacaPriceData.ticker.in_(upper_tickers), + AlpacaPriceData.date >= start_dt, + AlpacaPriceData.date <= end_dt, + ) + ) + .group_by(AlpacaPriceData.ticker) + ) + coverage = {row.ticker: row.max_date for row in result} + need_fetch = [ + t for t in upper_tickers + if coverage.get(t) is None or coverage[t].date() < end_dt.date() + ] + else: + need_fetch = upper_tickers + + if need_fetch: + start_str = start_dt.strftime("%Y-%m-%d") + end_str = end_dt.strftime("%Y-%m-%d") + + # reverse-map: normalized Alpaca symbol → original input ticker + reverse_map = {normalize_ticker(t): t for t in need_fetch} + + raw = await self.client.get_multi_bars( + symbols=need_fetch, + timeframe=interval, + start=start_str, + end=end_str, + ) + + rows = [] + for alpaca_sym, bar_list in raw.items(): + original = reverse_map.get(alpaca_sym, alpaca_sym) + for bar in bar_list: + bar_dt = _parse_bar_timestamp(bar["t"]) + rows.append({ + "ticker": original, + "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 rows: + # Chunk to stay under asyncpg's 32767-param limit (~10 params/row → 3000/chunk) + CHUNK = 3000 + for i in range(0, len(rows), CHUNK): + stmt = pg_insert(AlpacaPriceData).values(rows[i : i + CHUNK]) + stmt = stmt.on_conflict_do_nothing(constraint="uq_alpaca_price_data") + await db.execute(stmt) + await db.commit() + logger.info( + f"Alpaca multi-bars: stored {len(rows)} rows for {len(need_fetch)} tickers" + ) + + # Read all data back from DB + result = await db.execute( + select(AlpacaPriceData) + .where( + and_( + AlpacaPriceData.ticker.in_(upper_tickers), + AlpacaPriceData.date >= start_dt, + AlpacaPriceData.date <= end_dt, + ) + ) + .order_by(AlpacaPriceData.ticker, AlpacaPriceData.date) + ) + db_rows = result.scalars().all() + + data: Dict[str, List[AlpacaPriceData]] = {t: [] for t in upper_tickers} + for row in db_rows: + data[row.ticker].append(row) + return data + async def fetch_bars_raw( self, ticker: str,