""" Alpaca Market Data API client (v2) """ import asyncio import time import logging from typing import Dict, List, Optional import httpx from app.core.config import settings logger = logging.getLogger(__name__) # Interval mapping: internal format → Alpaca API format INTERVAL_MAP = { "1m": "1Min", "1min": "1Min", "2m": "2Min", "5m": "5Min", "5min": "5Min", "15m": "15Min", "15min": "15Min", "30m": "30Min", "30min": "30Min", "1h": "1Hour", "60min": "1Hour", "1d": "1Day", "1w": "1Week", "1mo": "1Month", } def normalize_ticker(symbol: str) -> str: """Normalize ticker symbol for Alpaca API. US stock tickers use dots for share classes (BF.B, BRK.B) while Yahoo Finance and other sources use hyphens (BF-B, BRK-B). """ return symbol.replace("-", ".") class AlpacaClient: """Alpaca Market Data API client (v2)""" def __init__( self, api_key: Optional[str] = None, secret_key: Optional[str] = None, base_url: Optional[str] = None, max_requests_per_min: int = 200, ): self.api_key = api_key or settings.ALPACA_API_KEY self.secret_key = secret_key or settings.ALPACA_SECRET_KEY self.base_url = (base_url or settings.ALPACA_BASE_URL).rstrip("/") self._semaphore = asyncio.Semaphore(max_requests_per_min) self._request_times: List[float] = [] self._max_rpm = max_requests_per_min self._client: Optional[httpx.AsyncClient] = None # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ def is_configured(self) -> bool: """Return True if both API key and secret are set.""" return bool(self.api_key) and bool(self.secret_key) async def _get_client(self) -> httpx.AsyncClient: if self._client is None or self._client.is_closed: self._client = httpx.AsyncClient( base_url=self.base_url, headers={ "APCA-API-KEY-ID": self.api_key, "APCA-API-SECRET-KEY": self.secret_key, }, timeout=30.0, ) return self._client async def close(self): if self._client and not self._client.is_closed: await self._client.aclose() # ------------------------------------------------------------------ # Rate limiting # ------------------------------------------------------------------ async def _wait_for_rate_limit(self): """Simple token-bucket style rate limiter (200 req/min).""" now = time.monotonic() # Remove entries older than 60 seconds self._request_times = [t for t in self._request_times if now - t < 60] if len(self._request_times) >= self._max_rpm: sleep_for = 60 - (now - self._request_times[0]) + 0.1 if sleep_for > 0: logger.debug(f"Alpaca rate limit reached, sleeping {sleep_for:.1f}s") await asyncio.sleep(sleep_for) self._request_times.append(time.monotonic()) # ------------------------------------------------------------------ # Core HTTP # ------------------------------------------------------------------ async def _request( self, method: str, path: str, params: Optional[Dict] = None, retries: int = 3 ) -> Dict: """Make an authenticated request with retry + rate limiting.""" await self._wait_for_rate_limit() client = await self._get_client() last_exc: Optional[Exception] = None for attempt in range(retries): try: resp = await client.request(method, path, params=params) if resp.status_code == 429: wait = 2 ** attempt logger.warning(f"Alpaca 429 – retrying in {wait}s (attempt {attempt + 1})") await asyncio.sleep(wait) continue if resp.status_code >= 500: wait = 2 ** attempt logger.warning(f"Alpaca {resp.status_code} – retrying in {wait}s (attempt {attempt + 1})") await asyncio.sleep(wait) continue resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as exc: last_exc = exc if attempt < retries - 1 and exc.response.status_code in (429, 500, 502, 503, 504): await asyncio.sleep(2 ** attempt) continue raise except (httpx.ConnectError, httpx.ReadTimeout) as exc: last_exc = exc if attempt < retries - 1: await asyncio.sleep(2 ** attempt) continue raise raise last_exc # type: ignore[misc] # ------------------------------------------------------------------ # Public API methods # ------------------------------------------------------------------ async def get_bars( self, symbol: str, timeframe: str = "1d", start: Optional[str] = None, end: Optional[str] = None, limit: int = 10000, ) -> List[Dict]: """ Fetch bars for a single symbol with automatic pagination. Args: symbol: Ticker symbol (e.g. "AAPL") timeframe: Internal interval string (e.g. "1d", "1h", "1m") start: RFC-3339 date/datetime (e.g. "2024-01-01") end: RFC-3339 date/datetime limit: Max bars per page (Alpaca max 10000) Returns: List of bar dicts with keys: t, o, h, l, c, v, n, vw """ alpaca_tf = INTERVAL_MAP.get(timeframe, timeframe) params: Dict = {"timeframe": alpaca_tf, "limit": min(limit, 10000)} if start: params["start"] = start if end: params["end"] = end all_bars: List[Dict] = [] path = f"/v2/stocks/{normalize_ticker(symbol).upper()}/bars" while True: data = await self._request("GET", path, params=params) bars = data.get("bars") or [] all_bars.extend(bars) next_token = data.get("next_page_token") if not next_token or not bars: break params["page_token"] = next_token logger.info(f"Alpaca: fetched {len(all_bars)} bars for {symbol} ({alpaca_tf})") return all_bars async def get_multi_bars( self, symbols: List[str], timeframe: str = "1d", start: Optional[str] = None, end: Optional[str] = None, limit: int = 10000, batch_size: int = 200, ) -> Dict[str, List[Dict]]: """ Fetch bars for multiple symbols with auto-pagination and transparent batching. Normalizes tickers (BF-B → BF.B) before calling Alpaca and returns results keyed by the *Alpaca* symbol (normalized). Callers that need the original symbol names should build their own reverse-map before calling. Args: symbols: List of ticker symbols (hyphens are auto-normalized to dots) timeframe: Internal interval string (e.g. "1d", "1h", "5min") start: RFC-3339 date/datetime string end: RFC-3339 date/datetime string limit: Max bars per page (Alpaca max 10000) batch_size: Max symbols per Alpaca request (default 200, ~1KB URL) Returns: Dict mapping normalized Alpaca symbol → list of bar dicts """ alpaca_tf = INTERVAL_MAP.get(timeframe, timeframe) path = "/v2/stocks/bars" # Normalize all symbols for Alpaca (BF-B → BF.B) normalized = [normalize_ticker(s).upper() for s in symbols] result: Dict[str, List[Dict]] = {s: [] for s in normalized} # Process in batches to stay within URL length limits for batch_start in range(0, len(normalized), batch_size): batch = normalized[batch_start: batch_start + batch_size] params: Dict = { "symbols": ",".join(batch), "timeframe": alpaca_tf, "limit": min(limit, 10000), } if start: params["start"] = start if end: params["end"] = end while True: data = await self._request("GET", path, params=params) bars_map = data.get("bars") or {} for sym, bars in bars_map.items(): result.setdefault(sym, []).extend(bars) next_token = data.get("next_page_token") if not next_token: break params["page_token"] = next_token total = sum(len(v) for v in result.values()) logger.info(f"Alpaca: fetched {total} bars for {len(symbols)} symbols ({alpaca_tf})") return result async def get_snapshot(self, symbol: str) -> Dict: """ Fetch a real-time snapshot for a single symbol. Returns Alpaca's snapshot object with keys: latestTrade, latestQuote, minuteBar, dailyBar, prevDailyBar """ data = await self._request("GET", f"/v2/stocks/{normalize_ticker(symbol).upper()}/snapshot") return data async def get_snapshots(self, symbols: List[str]) -> Dict[str, Dict]: """ Fetch real-time snapshots for multiple symbols in one request. Returns dict mapping symbol → snapshot object. """ params = {"symbols": ",".join(normalize_ticker(s).upper() for s in symbols)} data = await self._request("GET", "/v2/stocks/snapshots", params=params) return data # {SYMBOL: {...snapshot...}, ...} async def check_connection(self) -> Dict: """Verify API key validity by requesting a small amount of data.""" try: bars = await self.get_bars("AAPL", timeframe="1d", limit=1) return { "connected": True, "bars_returned": len(bars), "base_url": self.base_url, } except Exception as e: return { "connected": False, "error": str(e), "base_url": self.base_url, }