|
|
"""
|
|
|
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",
|
|
|
"2m": "2Min",
|
|
|
"5m": "5Min",
|
|
|
"15m": "15Min",
|
|
|
"30m": "30Min",
|
|
|
"1h": "1Hour",
|
|
|
"1d": "1Day",
|
|
|
"1w": "1Week",
|
|
|
"1mo": "1Month",
|
|
|
}
|
|
|
|
|
|
|
|
|
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/{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,
|
|
|
) -> Dict[str, List[Dict]]:
|
|
|
"""
|
|
|
Fetch bars for multiple symbols in one request with auto-pagination.
|
|
|
|
|
|
Returns:
|
|
|
Dict mapping symbol → list of bar dicts
|
|
|
"""
|
|
|
alpaca_tf = INTERVAL_MAP.get(timeframe, timeframe)
|
|
|
params: Dict = {
|
|
|
"symbols": ",".join(s.upper() for s in symbols),
|
|
|
"timeframe": alpaca_tf,
|
|
|
"limit": min(limit, 10000),
|
|
|
}
|
|
|
if start:
|
|
|
params["start"] = start
|
|
|
if end:
|
|
|
params["end"] = end
|
|
|
|
|
|
result: Dict[str, List[Dict]] = {s.upper(): [] for s in symbols}
|
|
|
path = "/v2/stocks/bars"
|
|
|
|
|
|
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 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,
|
|
|
}
|