"""Live market data provider — Alpaca (stocks) + Tradier (options). Activated when DATA_PROVIDER=live and API keys are present. Every call degrades gracefully to the mock provider on missing keys, HTTP errors, or timeouts, so the platform never breaks when a vendor is down. PAPER TRADING ONLY: these are read-only *data* APIs. No order endpoints are used anywhere in this module. """ import logging import os from datetime import date, timedelta import httpx from .market_data import MarketDataProvider, MockMarketDataProvider, UNIVERSE, is_crypto logger = logging.getLogger(__name__) ALPACA_DATA_URL = "https://data.alpaca.markets" TRADIER_URL = "https://api.tradier.com" TIMEOUT = 8.0 class LiveMarketDataProvider(MarketDataProvider): """Alpaca for bars/quotes, Tradier for options chains, mock as safety net.""" def __init__(self): self.alpaca_key = os.getenv("ALPACA_API_KEY", "") self.alpaca_secret = os.getenv("ALPACA_API_SECRET", "") self.tradier_token = os.getenv("TRADIER_API_TOKEN", "") self._fallback = MockMarketDataProvider() # ── helpers ────────────────────────────────────────── def _alpaca_headers(self) -> dict: return { "APCA-API-KEY-ID": self.alpaca_key, "APCA-API-SECRET-KEY": self.alpaca_secret, } def _tradier_headers(self) -> dict: return {"Authorization": f"Bearer {self.tradier_token}", "Accept": "application/json"} # ── interface ──────────────────────────────────────── def get_universe(self) -> list[dict]: # Static curated universe in Phase 2; screener-driven in Phase 3. return [{"symbol": s, "company": c, "base_price": p} for s, c, p in UNIVERSE] def _crypto_history(self, symbol: str, days: int) -> list[dict]: """Alpaca crypto daily bars (v1beta3). Falls back to mock on failure.""" try: start = (date.today() - timedelta(days=days + 5)).isoformat() r = httpx.get( f"{ALPACA_DATA_URL}/v1beta3/crypto/us/bars", params={"symbols": symbol, "timeframe": "1Day", "start": start, "limit": days}, headers=self._alpaca_headers(), timeout=TIMEOUT, ) r.raise_for_status() bars = (r.json().get("bars") or {}).get(symbol) or [] out = [ {"date": b["t"][:10], "open": float(b["o"]), "high": float(b["h"]), "low": float(b["l"]), "close": float(b["c"]), "volume": int(b["v"])} for b in bars ] if len(out) < 30: return self._fallback.get_history(symbol, days) return out[-days:] except Exception as exc: # noqa: BLE001 logger.warning("Alpaca crypto history failed for %s: %s — using mock", symbol, exc) return self._fallback.get_history(symbol, days) def get_history(self, symbol: str, days: int = 60) -> list[dict]: if is_crypto(symbol): return self._crypto_history(symbol, days) if not (self.alpaca_key and self.alpaca_secret): return self._fallback.get_history(symbol, days) try: start = (date.today() - timedelta(days=days * 2)).isoformat() params = {"timeframe": "1Day", "start": start, "limit": days, "adjustment": "split"} bars = [] # Feed quality order: sip (official consolidated) → delayed_sip # (official, 15-min delayed — fine for daily bars) → iex (single # exchange, prices can deviate from the official close). preferred = os.getenv("ALPACA_DATA_FEED", "sip") feeds = [preferred] + [f for f in ("sip", "delayed_sip", "iex") if f != preferred] for feed in feeds: r = httpx.get( f"{ALPACA_DATA_URL}/v2/stocks/{symbol}/bars", params={**params, "feed": feed}, headers=self._alpaca_headers(), timeout=TIMEOUT, ) if r.status_code in (400, 401, 403, 422): logger.warning("Alpaca '%s' feed rejected for %s (HTTP %s); trying next feed", feed, symbol, r.status_code) continue r.raise_for_status() bars = r.json().get("bars") or [] if bars: if feed == "iex": logger.warning("Alpaca served %s from IEX feed — prices may deviate " "from official closes", symbol) else: logger.info("Alpaca served %s from '%s' feed", symbol, feed) break out = [ { "date": b["t"][:10], "open": float(b["o"]), "high": float(b["h"]), "low": float(b["l"]), "close": float(b["c"]), "volume": int(b["v"]), } for b in bars ] if len(out) < 30: # not enough data to compute indicators return self._fallback.get_history(symbol, days) return out[-days:] except Exception as exc: # noqa: BLE001 — any vendor failure → mock logger.warning("Alpaca history failed for %s: %s — using mock", symbol, exc) return self._fallback.get_history(symbol, days) def get_quote(self, symbol: str) -> dict: hist = self.get_history(symbol) last, prev = hist[-1], hist[-2] price = last["close"] # Real-time upgrade: pull the latest trade (IEX feed is real-time on # the free tier). Falls back to the last daily close on any failure. # Crypto quotes use the last daily bar (sufficient for daily swings). if not is_crypto(symbol) and self.alpaca_key and self.alpaca_secret: try: r = httpx.get( f"{ALPACA_DATA_URL}/v2/stocks/{symbol}/trades/latest", params={"feed": "iex"}, headers=self._alpaca_headers(), timeout=5, ) if r.status_code == 200: p = (r.json().get("trade") or {}).get("p") if p: price = float(p) except Exception as exc: # noqa: BLE001 logger.debug("Latest-trade fetch failed for %s: %s", symbol, exc) avg_vol = sum(b["volume"] for b in hist[-20:]) / min(20, len(hist)) return { "symbol": symbol, "price": round(price, 2), "change_pct": round((price / prev["close"] - 1) * 100, 2), "volume": last["volume"], "relative_volume": round(last["volume"] / avg_vol, 2) if avg_vol else 1.0, } def get_options_chain(self, symbol: str) -> list[dict]: if not self.tradier_token: return self._fallback.get_options_chain(symbol) try: # Nearest expiration ~30 days out r = httpx.get( f"{TRADIER_URL}/v1/markets/options/expirations", params={"symbol": symbol}, headers=self._tradier_headers(), timeout=TIMEOUT, ) r.raise_for_status() expirations = (r.json().get("expirations") or {}).get("date") or [] if isinstance(expirations, str): expirations = [expirations] if not expirations: return self._fallback.get_options_chain(symbol) target = date.today() + timedelta(days=30) expiry = min(expirations, key=lambda d: abs(date.fromisoformat(d) - target)) r = httpx.get( f"{TRADIER_URL}/v1/markets/options/chains", params={"symbol": symbol, "expiration": expiry, "greeks": "true"}, headers=self._tradier_headers(), timeout=TIMEOUT, ) r.raise_for_status() options = (r.json().get("options") or {}).get("option") or [] chain = [] for o in options: greeks = o.get("greeks") or {} chain.append({ "symbol": symbol, "type": o["option_type"], # "call" | "put" "strike": float(o["strike"]), "expiry": o["expiration_date"], "bid": float(o.get("bid") or 0), "ask": float(o.get("ask") or 0), "iv": float(greeks.get("mid_iv") or 0), "delta": float(greeks.get("delta") or 0), "gamma": float(greeks.get("gamma") or 0), "theta": float(greeks.get("theta") or 0), "vega": float(greeks.get("vega") or 0), "open_interest": int(o.get("open_interest") or 0), "volume": int(o.get("volume") or 0), }) return chain or self._fallback.get_options_chain(symbol) except Exception as exc: # noqa: BLE001 logger.warning("Tradier chain failed for %s: %s — using mock", symbol, exc) return self._fallback.get_options_chain(symbol)