diff --git a/apps/orb_trader/engine.py b/apps/orb_trader/engine.py index 9f4413d..81f02bf 100644 --- a/apps/orb_trader/engine.py +++ b/apps/orb_trader/engine.py @@ -109,12 +109,6 @@ class ORBTradingEngine: raw_bars.update(self._broker.get_bars(chunk, start, today)) except Exception as e: self._log(f" WARNING: daily bars chunk {i//chunk_size+1} failed ({e}) — skipping") - # Retry chunk symbol-by-symbol to isolate the bad ticker(s) - for sym in chunk: - try: - raw_bars.update(self._broker.get_bars([sym], start, today)) - except Exception: - self._log(f" Skipping invalid symbol: {sym}") daily_bars_dict = bars_to_enrichment_format(raw_bars) diff --git a/apps/paper_trader/alpaca_broker.py b/apps/paper_trader/alpaca_broker.py index abcf61c..773b3c0 100644 --- a/apps/paper_trader/alpaca_broker.py +++ b/apps/paper_trader/alpaca_broker.py @@ -225,35 +225,33 @@ class AlpacaBroker: start: dt.date, end: dt.date, ) -> dict[str, list[Bar]]: - """Fetch daily OHLCV bars for a list of symbols in [start, end].""" + """Fetch daily OHLCV bars for a list of symbols via Oracle API. + + Oracle normalises problematic symbols (e.g. BF-B → BF.B) and maps + responses back to the original symbol names. + """ if not symbols: return {} - from alpaca.data.requests import StockBarsRequest - from alpaca.data.timeframe import TimeFrame + from libs.oracle_client.alpaca import get_multi_daily_bars - req = StockBarsRequest( - symbol_or_symbols=symbols, - timeframe=TimeFrame.Day, - start=dt.datetime.combine(start, dt.time.min), - end=dt.datetime.combine(end, dt.time.max), - feed="iex", + raw = get_multi_daily_bars( + tickers=symbols, + start_date=start.isoformat(), + end_date=end.isoformat(), ) - response = self._data.get_stock_bars(req) + result: dict[str, list[Bar]] = {} for sym in symbols: - try: - bars_data = response[sym] - except (KeyError, TypeError): - bars_data = [] + bars_data = raw.get(sym, []) result[sym] = [ Bar( - date=b.timestamp.date().isoformat() if hasattr(b.timestamp, "date") else str(b.timestamp)[:10], - open=float(b.open), - high=float(b.high), - low=float(b.low), - close=float(b.close), - volume=float(b.volume), + date=b["date"], + open=float(b["open"]), + high=float(b["high"]), + low=float(b["low"]), + close=float(b["close"]), + volume=float(b["volume"]), ) for b in bars_data ] @@ -283,6 +281,35 @@ class AlpacaBroker: result[sym] = date_map return result + def get_intraday_bars( + self, + symbols: list[str], + start: dt.datetime, + end: dt.datetime, + timeframe_minutes: int = 5, + ) -> dict[str, list[dict]]: + """Fetch intraday OHLCV bars for a list of symbols via Oracle API. + + Returns {symbol: [{timestamp: ISO8601, open, high, low, close, volume}, ...]}. + """ + if not symbols: + return {} + + from libs.oracle_client.alpaca import get_multi_intraday_bars + + interval = f"{timeframe_minutes}min" + raw = get_multi_intraday_bars( + tickers=symbols, + start_date=start.date().isoformat(), + end_date=end.date().isoformat(), + interval=interval, + ) + + result: dict[str, list[dict]] = {sym: [] for sym in symbols} + for sym in symbols: + result[sym] = raw.get(sym, []) + return result + def get_latest_bars(self, symbols: list[str]) -> dict[str, Bar]: """Fetch the latest bar for each symbol.""" if not symbols: diff --git a/libs/oracle_client/alpaca.py b/libs/oracle_client/alpaca.py new file mode 100644 index 0000000..5454ffd --- /dev/null +++ b/libs/oracle_client/alpaca.py @@ -0,0 +1,197 @@ +"""Alpaca real-time snapshot service via Stock Oracle API. + +Provides synchronous helpers for real-time price snapshots, suitable for use +in threaded engine code (which already runs in asyncio.to_thread). +""" +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass +from typing import Any + +log = logging.getLogger(__name__) + + +@dataclass +class AlpacaSnapshot: + """Real-time Alpaca snapshot for a single ticker.""" + + ticker: str + price: float | None = None # last trade price + bid: float | None = None + ask: float | None = None + bid_size: int | None = None + ask_size: int | None = None + open: float | None = None + high: float | None = None + low: float | None = None + volume: int | None = None + vwap: float | None = None + prev_close: float | None = None + change: float | None = None + change_pct: float | None = None + timestamp: str | None = None + + @property + def mid(self) -> float | None: + """Midpoint between bid and ask, or last trade if unavailable.""" + if self.bid is not None and self.ask is not None: + return (self.bid + self.ask) / 2 + return self.price + + +def _base_url() -> str: + return os.environ.get("ORACLE_URL", "http://localhost:18001").rstrip("/") + + +def _parse_snapshot(data: dict[str, Any]) -> AlpacaSnapshot: + return AlpacaSnapshot( + ticker=data["ticker"], + price=data.get("price"), + bid=data.get("bid"), + ask=data.get("ask"), + bid_size=data.get("bid_size"), + ask_size=data.get("ask_size"), + open=data.get("open"), + high=data.get("high"), + low=data.get("low"), + volume=data.get("volume"), + vwap=data.get("vwap"), + prev_close=data.get("prev_close"), + change=data.get("change"), + change_pct=data.get("change_pct"), + timestamp=data.get("timestamp"), + ) + + +def get_multi_daily_bars( + tickers: list[str], + start_date: str, + end_date: str, + base_url: str | None = None, +) -> dict[str, list[dict]]: + """Fetch daily OHLCV bars for multiple tickers via Oracle API. + + Oracle handles symbol normalisation (e.g. BF-B → BF.B) and Alpaca batching + internally. Returns {ticker: [{date, open, high, low, close, volume}, ...]}. + Missing/errored tickers are omitted. + """ + import httpx + + if not tickers: + return {} + + result: dict[str, list[dict]] = {} + url = (base_url or _base_url()) + "/api/v1/price/data" + + # Oracle handles its own batching, but we chunk here as a safety net for + # very large URL query strings. + chunk_size = 300 + for i in range(0, len(tickers), chunk_size): + chunk = tickers[i : i + chunk_size] + try: + resp = httpx.get( + url, + params={"tickers": ",".join(chunk), "start_date": start_date, "end_date": end_date}, + timeout=90.0, + ) + resp.raise_for_status() + for ticker, bars in resp.json().get("bars", {}).items(): + result[ticker] = bars + except Exception as exc: + log.warning("Oracle multi_daily_bars chunk %d failed: %s", i // chunk_size, exc) + + return result + + +def get_multi_intraday_bars( + tickers: list[str], + start_date: str, + end_date: str, + interval: str = "5min", + base_url: str | None = None, +) -> dict[str, list[dict]]: + """Fetch intraday bars for multiple tickers via Oracle API. + + Returns {ticker: [{timestamp (ISO8601), open, high, low, close, volume}, ...]}. + """ + import httpx + + if not tickers: + return {} + + result: dict[str, list[dict]] = {} + url = (base_url or _base_url()) + "/api/v1/alpaca/intraday" + + chunk_size = 300 + for i in range(0, len(tickers), chunk_size): + chunk = tickers[i : i + chunk_size] + try: + resp = httpx.get( + url, + params={ + "tickers": ",".join(chunk), + "interval": interval, + "start_date": start_date, + "end_date": end_date, + }, + timeout=90.0, + ) + resp.raise_for_status() + for ticker, bars in resp.json().get("bars", {}).items(): + result[ticker] = bars + except Exception as exc: + log.warning("Oracle multi_intraday_bars chunk %d failed: %s", i // chunk_size, exc) + + return result + + +def get_snapshot(ticker: str, base_url: str | None = None) -> AlpacaSnapshot | None: + """Fetch real-time snapshot for a single ticker (synchronous). + + Returns None on any error. + """ + import httpx + + url = (base_url or _base_url()) + f"/api/v1/alpaca/snapshot/{ticker}" + try: + resp = httpx.get(url, timeout=10.0) + resp.raise_for_status() + return _parse_snapshot(resp.json()) + except Exception as exc: + log.warning("Oracle snapshot(%s) failed: %s", ticker, exc) + return None + + +def get_snapshots( + tickers: list[str], + base_url: str | None = None, + batch_size: int = 200, +) -> dict[str, AlpacaSnapshot]: + """Fetch real-time snapshots for multiple tickers (synchronous). + + Batches requests and returns {ticker: AlpacaSnapshot}. + Missing/errored tickers are omitted. + """ + import httpx + + if not tickers: + return {} + + result: dict[str, AlpacaSnapshot] = {} + url = (base_url or _base_url()) + "/api/v1/alpaca/snapshot" + + for i in range(0, len(tickers), batch_size): + chunk = tickers[i : i + batch_size] + try: + resp = httpx.get(url, params={"tickers": ",".join(chunk)}, timeout=15.0) + resp.raise_for_status() + data = resp.json() + for snap_data in data.get("snapshots", []): + snap = _parse_snapshot(snap_data) + result[snap.ticker] = snap + except Exception as exc: + log.warning("Oracle snapshots(batch %d) failed: %s", i // batch_size, exc) + + return result