From 6eaeff33537673b9bd222a15abec8bb0909bf2a3 Mon Sep 17 00:00:00 2001 From: Jackson Atassi Date: Thu, 11 Jun 2026 19:14:01 -0600 Subject: [PATCH 1/3] fix(data_sources): merge Alpaca indicative NBBO + IEX spot into the options collector (ALP-949) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Polygon's options snapshots on the current plan carry greeks/OI but withhold last_quote/last_trade/underlying_asset.price, so every options_contract_snapshots row since the table's first batch has NULL bid/ask/last_price/underlying_price — no IVr context line renders, the chain tool returns no contracts, premium validation rejects everything, and option marks have no price. Merge at collection time (no schema/schedule change): - New vendor module src/alphamind/data_sources/alpaca/options.py: fetch_chain_quotes (OptionHistoricalDataClient, indicative feed — _OPTIONS_FEED is the single flip point for a future OPRA upgrade) and fetch_underlying_trades (one multi-symbol IEX StockLatestTradeRequest). - collect_options_chains fetches the spot map once per run and the Alpaca chain per underlying (with_retries important); any Alpaca failure degrades that scope to Polygon-only rows — never aborts. - _build_rows fills bid/ask/last_price/underlying_price where Polygon's value is None (Polygon wins per field under a future plan upgrade); Alpaca rows matched via bare OCC symbol (O: prefix stripped); source is polygon+alpaca when the chain fetch succeeded, else polygon. - config/data_sources.yaml already carries the alpaca provider entry (Q1 failover) — no config change needed. - docs/runbooks/services.md documents the dependency + degraded mode. Tests: fakes in tests/data_sources/_fakes/alpaca.py (sanctioned broker-API boundary), vendor-module tests, merge/precedence/isolation/ spot-failure tests; existing options tests inject empty Alpaca fakes for hermeticity. Two `# type: ignore[no-untyped-call]` on APIError construction in tests, mirroring tests/execution/broker_adapter. Co-Authored-By: Claude Fable 5 --- docs/runbooks/services.md | 14 ++ src/alphamind/data_sources/alpaca/__init__.py | 0 src/alphamind/data_sources/alpaca/options.py | 116 +++++++++ src/alphamind/data_sources/polygon/options.py | 115 ++++++++- tests/data_sources/_fakes/alpaca.py | 75 ++++++ tests/data_sources/_fakes/polygon.py | 3 +- tests/data_sources/alpaca/__init__.py | 0 tests/data_sources/alpaca/test_options.py | 97 +++++++ tests/data_sources/polygon/test_polygon.py | 236 ++++++++++++++++++ 9 files changed, 648 insertions(+), 8 deletions(-) create mode 100644 src/alphamind/data_sources/alpaca/__init__.py create mode 100644 src/alphamind/data_sources/alpaca/options.py create mode 100644 tests/data_sources/_fakes/alpaca.py create mode 100644 tests/data_sources/alpaca/__init__.py create mode 100644 tests/data_sources/alpaca/test_options.py diff --git a/docs/runbooks/services.md b/docs/runbooks/services.md index f3e032625..63ed4129e 100644 --- a/docs/runbooks/services.md +++ b/docs/runbooks/services.md @@ -50,6 +50,20 @@ failure. A frozen loop also self-captures its blocking frame: the faulthandler deadman dumps the frozen stack to `monitor_faulthandler.log` before the restart lands. The same-account rule applies to this watchdog too. +**Collector's Alpaca market-data dependency (ALP-949).** The +`alphamind-collector`'s `polygon.options` task merges Alpaca market data into +every chain snapshot at collection time: per-underlying option NBBO quotes +from Alpaca's free **indicative** options feed (bid/ask/last) plus one +multi-symbol **IEX** latest-trade call for the underlying spot — the Polygon +plan withholds those fields (greeks/open interest stay Polygon-sourced). +Credentials are `ALPACA_PAPER_KEY` / `ALPACA_PAPER_SECRET` from `.env`. +**Degraded mode:** any Alpaca failure (missing credentials, API or transport +error) never aborts the task — it still writes that scope's rows +Polygon-only (`source` `polygon`, quote columns NULL), logged as warnings in +`collector.log`. A per-underlying chain-fetch failure degrades only that +underlying; a failed batch spot fetch leaves `underlying_price` NULL for the +run. Rows written with `source` `polygon+alpaca` confirm the merge is live. + Logs land under `%USERPROFILE%\AlphaMind\logs\`. DB is `%USERPROFILE%\AlphaMind\data\alphamind.db`. Invocation archives land under `%USERPROFILE%\AlphaMind\archive\\\`. diff --git a/src/alphamind/data_sources/alpaca/__init__.py b/src/alphamind/data_sources/alpaca/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/alphamind/data_sources/alpaca/options.py b/src/alphamind/data_sources/alpaca/options.py new file mode 100644 index 000000000..df05bc589 --- /dev/null +++ b/src/alphamind/data_sources/alpaca/options.py @@ -0,0 +1,116 @@ +"""Alpaca market-data fetches merged into the options collector (ALP-949). + +Polygon's options snapshots on the current plan carry greeks / open +interest but no NBBO quotes and no underlying price; Alpaca's free plan +carries indicative real-time option quotes and IEX stock trades but no +open interest. The two complement exactly, so +:func:`alphamind.data_sources.polygon.options.collect_options_chains` +merges this module's quotes and spot prices into the Polygon-sourced +snapshot rows at collection time. + +Exports +------- +OptionQuote +fetch_chain_quotes(underlying, ...) — full chain, keyed by bare OCC symbol +fetch_underlying_trades(tickers, ...) — latest IEX trade price per ticker +ALPACA_FETCH_ERRORS — exception classes callers degrade on +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any + +import requests.exceptions +from alpaca.common.exceptions import APIError +from alpaca.data.enums import DataFeed, OptionsFeed +from alpaca.data.historical import StockHistoricalDataClient +from alpaca.data.historical.option import OptionHistoricalDataClient +from alpaca.data.requests import OptionChainRequest, StockLatestTradeRequest + +# The single flip point if the operator subscribes to Alpaca's Algo Trader +# Plus market-data plan (paid; unlocks the real-time OPRA feed). On the +# Basic (free) plan, ``indicative`` is the only available options feed. +_OPTIONS_FEED = OptionsFeed.INDICATIVE + +# Exceptions a chain/spot fetch raises on vendor or transport failure — +# callers degrade to Polygon-only rows on these rather than aborting. +ALPACA_FETCH_ERRORS: tuple[type[Exception], ...] = ( + APIError, + requests.exceptions.RequestException, +) + + +@dataclass(frozen=True) +class OptionQuote: + """NBBO + last trade for one contract, any field absent as ``None``.""" + + bid: float | None + ask: float | None + last_price: float | None + + +def make_option_client() -> OptionHistoricalDataClient: + """Build the option-snapshot client from ``ALPACA_PAPER_*`` env vars. + + The market-data plane accepts paper keys; reading the env directly + mirrors the :class:`~alphamind.data_sources.polygon.client.PolygonClient` + convention. Raises ``ValueError`` (from the SDK) when the keys are unset. + """ + return OptionHistoricalDataClient( + api_key=os.environ.get("ALPACA_PAPER_KEY"), + secret_key=os.environ.get("ALPACA_PAPER_SECRET"), + ) + + +def make_stock_client() -> StockHistoricalDataClient: + """Build the stock-trade client from ``ALPACA_PAPER_*`` env vars.""" + return StockHistoricalDataClient( + api_key=os.environ.get("ALPACA_PAPER_KEY"), + secret_key=os.environ.get("ALPACA_PAPER_SECRET"), + ) + + +def fetch_chain_quotes(underlying: str, *, _client: Any = None) -> dict[str, OptionQuote]: + """Fetch the full unfiltered chain for *underlying*, keyed by bare OCC symbol. + + Keys are Alpaca's bare OCC symbols (e.g. ``MU260612P01062500`` — no + Polygon ``O:`` prefix). Bid/ask come from ``latest_quote``, last price + from ``latest_trade``; each is ``None`` when the sub-object is absent. + """ + client = _client if _client is not None else make_option_client() + chain = client.get_option_chain( + OptionChainRequest(underlying_symbol=underlying, feed=_OPTIONS_FEED) + ) + quotes: dict[str, OptionQuote] = {} + for symbol, snap in chain.items(): + lq = snap.latest_quote + lt = snap.latest_trade + quotes[symbol] = OptionQuote( + bid=float(lq.bid_price) if lq is not None and lq.bid_price is not None else None, + ask=float(lq.ask_price) if lq is not None and lq.ask_price is not None else None, + last_price=float(lt.price) if lt is not None and lt.price is not None else None, + ) + return quotes + + +def fetch_underlying_trades(tickers: list[str], *, _client: Any = None) -> dict[str, float]: + """Latest IEX trade price per ticker via one multi-symbol request. + + Tickers without a positive-price trade in the response are omitted, so + callers fall back to ``None`` for them. The IEX feed is the same free + surface the continuous monitor's safety-core price feed polls. + """ + if not tickers: + return {} + client = _client if _client is not None else make_stock_client() + response = client.get_stock_latest_trade( + StockLatestTradeRequest(symbol_or_symbols=tickers, feed=DataFeed.IEX) + ) + out: dict[str, float] = {} + for ticker in tickers: + price = getattr(response.get(ticker), "price", None) + if price is not None and float(price) > 0: + out[ticker] = float(price) + return out diff --git a/src/alphamind/data_sources/polygon/options.py b/src/alphamind/data_sources/polygon/options.py index 4853f1f1b..21309ba39 100644 --- a/src/alphamind/data_sources/polygon/options.py +++ b/src/alphamind/data_sources/polygon/options.py @@ -10,6 +10,7 @@ from __future__ import annotations import logging +from collections.abc import Mapping from datetime import UTC, datetime from typing import Any @@ -24,6 +25,14 @@ track_run, with_retries, ) +from alphamind.data_sources.alpaca.options import ( + ALPACA_FETCH_ERRORS, + OptionQuote, + fetch_chain_quotes, + fetch_underlying_trades, + make_option_client, + make_stock_client, +) from alphamind.data_sources.polygon.client import PolygonClient from alphamind.persistence.models import OptionsContracts, OptionsContractSnapshots @@ -53,6 +62,60 @@ ) +def _alpaca_clients(option_client: Any, stock_client: Any) -> tuple[Any, Any]: + """Build any missing Alpaca data client, degrading to ``(None, None)`` + when the ``ALPACA_PAPER_*`` credentials are unset so the run proceeds + Polygon-only instead of aborting.""" + try: + if option_client is None: + option_client = make_option_client() + if stock_client is None: + stock_client = make_stock_client() + except ValueError as exc: + logger.warning("alpaca data clients unavailable; writing polygon-only rows", exc_info=exc) + return None, None + return option_client, stock_client + + +@with_retries(RetryShape.important, _sleep=lambda _: None) +def _alpaca_chain_with_retries(underlying: str, client: Any) -> dict[str, OptionQuote]: + return fetch_chain_quotes(underlying, _client=client) + + +@with_retries(RetryShape.important, _sleep=lambda _: None) +def _alpaca_spots_with_retries(tickers: list[str], client: Any) -> dict[str, float]: + return fetch_underlying_trades(tickers, _client=client) + + +def _fetch_alpaca_quotes(underlying: str, option_client: Any) -> dict[str, OptionQuote] | None: + """Alpaca chain quotes for *underlying*; ``None`` on failure (degraded).""" + if option_client is None: + return None + try: + return _alpaca_chain_with_retries(underlying, option_client) + except ALPACA_FETCH_ERRORS as exc: + logger.warning( + "alpaca option chain fetch failed for %s; writing polygon-only rows", + underlying, + exc_info=exc, + ) + return None + + +def _fetch_spot_map(tickers: list[str], stock_client: Any) -> dict[str, float]: + """Batch IEX spot map; ``{}`` on failure (``underlying_price`` stays None).""" + if stock_client is None: + return {} + try: + return _alpaca_spots_with_retries(tickers, stock_client) + except ALPACA_FETCH_ERRORS as exc: + logger.warning( + "alpaca underlying spot fetch failed; underlying_price stays None for this batch", + exc_info=exc, + ) + return {} + + def collect_options_chains( ticker_scope: list[str] | None = None, since: datetime | None = None, @@ -61,12 +124,19 @@ def collect_options_chains( _session_factory: Any = None, _repo: Any = None, _snapshot_ts: str | None = None, + _alpaca_option_client: Any = None, + _alpaca_stock_client: Any = None, ) -> None: """ Pull full chain snapshot per universe ticker and write to: - ``options_contracts`` (UPSERT: insert new, update ``last_seen_at`` on known) - ``options_contract_snapshots`` (one row per contract per snapshot timestamp) + Polygon's plan withholds NBBO quotes and the underlying price, so each + underlying's rows are merged with Alpaca indicative chain quotes and one + IEX spot fetch per run (ALP-949). Any Alpaca failure degrades that scope + to Polygon-only rows — it never aborts the collection. + Parameters ---------- ticker_scope: @@ -81,6 +151,9 @@ def collect_options_chains( del since # accepted for runner-contract compatibility; ignored by snapshot semantics if _client is None: _client = PolygonClient() + _alpaca_option_client, _alpaca_stock_client = _alpaca_clients( + _alpaca_option_client, _alpaca_stock_client + ) _session_factory = _session_factory or default_session_factory() if ticker_scope is None: ticker_scope = active_universe_tickers( @@ -94,6 +167,8 @@ def collect_options_chains( def _fetch_chain(underlying: str) -> list[Any]: return _client.list_snapshot_options_chain(underlying) # type: ignore[no-any-return] + spot_by_ticker = _fetch_spot_map(ticker_scope, _alpaca_stock_client) + with track_run("polygon.options", _repo=_repo) as run: rows_written = 0 single = len(ticker_scope) == 1 @@ -112,7 +187,12 @@ def _fetch_chain(underlying: str) -> list[Any]: continue contract_rows, snapshot_rows = _build_rows( - snapshots, underlying, snapshot_ts, ingested_at + snapshots, + underlying, + snapshot_ts, + ingested_at, + alpaca_quotes=_fetch_alpaca_quotes(underlying, _alpaca_option_client), + underlying_price=spot_by_ticker.get(underlying), ) _upsert_contracts(_session_factory, contract_rows, snapshot_ts) _upsert_snapshots(_session_factory, snapshot_rows) @@ -126,8 +206,22 @@ def _build_rows( underlying: str, snapshot_ts: str, ingested_at: str, + *, + alpaca_quotes: Mapping[str, OptionQuote] | None = None, + underlying_price: float | None = None, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: - """Convert a Polygon snapshot list to row-dicts for both options tables.""" + """Convert a Polygon snapshot list to row-dicts for both options tables. + + ``alpaca_quotes`` (keyed by bare OCC symbol — Polygon's ``O:`` prefix + stripped for the lookup) and ``underlying_price`` fill ``bid`` / ``ask`` + / ``last_price`` / ``underlying_price`` where the Polygon-parsed value + is None; a Polygon-delivered value wins per field (the current plan + never delivers them, so Alpaca always fills; precedence stays correct + under a future plan upgrade). ``alpaca_quotes=None`` means the Alpaca + chain fetch failed — the snapshot ``source`` then stays ``polygon`` + instead of ``polygon+alpaca``. + """ + snapshot_source = "polygon" if alpaca_quotes is None else "polygon+alpaca" contract_rows: list[dict[str, Any]] = [] snapshot_rows: list[dict[str, Any]] = [] for snap in snapshots: @@ -137,6 +231,7 @@ def _build_rows( lq = snap.last_quote lt = snap.last_trade ua = snap.underlying_asset + aq = (alpaca_quotes or {}).get(details.ticker.removeprefix("O:")) contract_rows.append( { @@ -160,9 +255,13 @@ def _build_rows( int(snap.open_interest) if snap.open_interest is not None else None ), "volume_today": int(day.volume) if day and day.volume is not None else None, - "last_price": float(lt.price) if lt and lt.price is not None else None, - "bid": float(lq.bid) if lq and lq.bid is not None else None, - "ask": float(lq.ask) if lq and lq.ask is not None else None, + "last_price": ( + float(lt.price) + if lt and lt.price is not None + else (aq.last_price if aq else None) + ), + "bid": float(lq.bid) if lq and lq.bid is not None else (aq.bid if aq else None), + "ask": float(lq.ask) if lq and lq.ask is not None else (aq.ask if aq else None), "implied_volatility": float(snap.implied_volatility) if snap.implied_volatility is not None else None, @@ -171,8 +270,10 @@ def _build_rows( "theta": float(greeks.theta) if greeks and greeks.theta is not None else None, "vega": float(greeks.vega) if greeks and greeks.vega is not None else None, "rho": float(rho_val) if rho_val is not None else None, - "underlying_price": float(ua.price) if ua and ua.price is not None else None, - "source": "polygon", + "underlying_price": ( + float(ua.price) if ua and ua.price is not None else underlying_price + ), + "source": snapshot_source, "ingested_at": ingested_at, } ) diff --git a/tests/data_sources/_fakes/alpaca.py b/tests/data_sources/_fakes/alpaca.py new file mode 100644 index 000000000..5e1747c30 --- /dev/null +++ b/tests/data_sources/_fakes/alpaca.py @@ -0,0 +1,75 @@ +"""In-memory fakes for the alpaca-py market-data clients consumed by the +options collector (ALP-949). + +The SDK returns mapping responses keyed by symbol whose values carry +attribute access (``snap.latest_quote.bid_price``), so the fakes here are +:func:`dataclasses.dataclass` records mirroring those shapes without +depending on the real SDK in tests. Request objects are the real SDK +pydantic models (no network is involved in constructing them); the fake +clients read the fields the production code sets. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +# --------------------------------------------------------------------------- +# Record shapes (attribute-bearing fakes of the SDK return types) +# --------------------------------------------------------------------------- + + +@dataclass +class FakeOptionQuoteRecord: + bid_price: float | None = None + ask_price: float | None = None + + +@dataclass +class FakeOptionTradeRecord: + price: float | None = None + + +@dataclass +class FakeOptionSnapshotRecord: + latest_quote: FakeOptionQuoteRecord | None = None + latest_trade: FakeOptionTradeRecord | None = None + + +@dataclass +class FakeStockTradeRecord: + price: float | None = None + + +# --------------------------------------------------------------------------- +# Fake clients +# --------------------------------------------------------------------------- + + +@dataclass +class FakeAlpacaOptionClient: + """Fake of ``OptionHistoricalDataClient`` keyed by underlying symbol.""" + + chains_by_underlying: dict[str, dict[str, FakeOptionSnapshotRecord]] = field( + default_factory=dict + ) + chain_calls: list[Any] = field(default_factory=list) + + def get_option_chain(self, request_params: Any) -> dict[str, FakeOptionSnapshotRecord]: + self.chain_calls.append(request_params) + return dict(self.chains_by_underlying.get(request_params.underlying_symbol, {})) + + +@dataclass +class FakeAlpacaStockClient: + """Fake of ``StockHistoricalDataClient`` serving latest trades.""" + + trades_by_symbol: dict[str, FakeStockTradeRecord] = field(default_factory=dict) + trade_calls: list[Any] = field(default_factory=list) + + def get_stock_latest_trade(self, request_params: Any) -> dict[str, FakeStockTradeRecord]: + self.trade_calls.append(request_params) + symbols = request_params.symbol_or_symbols + if isinstance(symbols, str): + symbols = [symbols] + return {s: self.trades_by_symbol[s] for s in symbols if s in self.trades_by_symbol} diff --git a/tests/data_sources/_fakes/polygon.py b/tests/data_sources/_fakes/polygon.py index 38ec3b2ec..06a3826f2 100644 --- a/tests/data_sources/_fakes/polygon.py +++ b/tests/data_sources/_fakes/polygon.py @@ -230,6 +230,7 @@ def get_ticker_details(self, ticker: str) -> object: "volume": 500, "bid": 5.0, "ask": 5.5, + "last_price": 5.25, "iv": 0.30, "delta": 0.45, "gamma": 0.02, @@ -257,7 +258,7 @@ def make_option_snapshot(**overrides: Any) -> FakeOptionSnapshot: ), day=FakeDay(volume=v["volume"]), last_quote=FakeQuote(bid=v["bid"], ask=v["ask"]), - last_trade=FakeTrade(price=(v["bid"] + v["ask"]) / 2), + last_trade=FakeTrade(price=v["last_price"]), underlying_asset=FakeUnderlyingAsset(ticker=v["underlying"], price=v["underlying_price"]), implied_volatility=v["iv"], open_interest=v["oi"], diff --git a/tests/data_sources/alpaca/__init__.py b/tests/data_sources/alpaca/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/data_sources/alpaca/test_options.py b/tests/data_sources/alpaca/test_options.py new file mode 100644 index 000000000..d21a2c726 --- /dev/null +++ b/tests/data_sources/alpaca/test_options.py @@ -0,0 +1,97 @@ +"""Tests for src/alphamind/data_sources/alpaca/options.py — ALP-949. + +All alpaca-py SDK calls are routed through the fakes in +``tests/data_sources/_fakes/alpaca.py``. No real network traffic. +""" + +from __future__ import annotations + +from tests.data_sources._fakes.alpaca import ( + FakeAlpacaOptionClient, + FakeAlpacaStockClient, + FakeOptionQuoteRecord, + FakeOptionSnapshotRecord, + FakeOptionTradeRecord, + FakeStockTradeRecord, +) + +# --------------------------------------------------------------------------- +# fetch_chain_quotes +# --------------------------------------------------------------------------- + + +class TestFetchChainQuotes: + def test_keys_by_bare_occ_and_maps_quote_fields(self) -> None: + from alphamind.data_sources.alpaca.options import OptionQuote, fetch_chain_quotes + + client = FakeAlpacaOptionClient( + chains_by_underlying={ + "MU": { + "MU260612P01062500": FakeOptionSnapshotRecord( + latest_quote=FakeOptionQuoteRecord(bid_price=1.05, ask_price=1.15), + latest_trade=FakeOptionTradeRecord(price=1.10), + ), + "MU260612C01062500": FakeOptionSnapshotRecord(), + } + } + ) + + quotes = fetch_chain_quotes("MU", _client=client) + + assert quotes == { + "MU260612P01062500": OptionQuote(bid=1.05, ask=1.15, last_price=1.10), + "MU260612C01062500": OptionQuote(bid=None, ask=None, last_price=None), + } + + def test_requests_full_chain_on_the_indicative_feed(self) -> None: + """The request sent to the vendor carries the underlying and the + indicative feed (the only options feed entitled on the free plan) + with no strike/expiration narrowing — the merge needs the full chain.""" + from alpaca.data.enums import OptionsFeed + + from alphamind.data_sources.alpaca.options import fetch_chain_quotes + + client = FakeAlpacaOptionClient() + + fetch_chain_quotes("NVDA", _client=client) + + (request,) = client.chain_calls + assert request.underlying_symbol == "NVDA" + assert request.feed == OptionsFeed.INDICATIVE + assert request.strike_price_gte is None + assert request.strike_price_lte is None + assert request.expiration_date is None + + +# --------------------------------------------------------------------------- +# fetch_underlying_trades +# --------------------------------------------------------------------------- + + +class TestFetchUnderlyingTrades: + def test_one_multi_symbol_iex_request_maps_prices_and_omits_missing(self) -> None: + from alpaca.data.enums import DataFeed + + from alphamind.data_sources.alpaca.options import fetch_underlying_trades + + client = FakeAlpacaStockClient( + trades_by_symbol={ + "MU": FakeStockTradeRecord(price=106.25), + "NVDA": FakeStockTradeRecord(price=None), + } + ) + + spots = fetch_underlying_trades(["MU", "NVDA", "AAPL"], _client=client) + + assert spots == {"MU": 106.25} + (request,) = client.trade_calls + assert request.symbol_or_symbols == ["MU", "NVDA", "AAPL"] + assert request.feed == DataFeed.IEX + + def test_empty_ticker_list_makes_no_request(self) -> None: + from alphamind.data_sources.alpaca.options import fetch_underlying_trades + + client = FakeAlpacaStockClient() + + assert fetch_underlying_trades([], _client=client) == {} + assert client.trade_calls == [] diff --git a/tests/data_sources/polygon/test_polygon.py b/tests/data_sources/polygon/test_polygon.py index e854261dd..ded8edbc1 100644 --- a/tests/data_sources/polygon/test_polygon.py +++ b/tests/data_sources/polygon/test_polygon.py @@ -22,6 +22,14 @@ OptionsContractSnapshots, ) from alphamind.persistence.session import make_engine, make_session_factory +from tests.data_sources._fakes.alpaca import ( + FakeAlpacaOptionClient, + FakeAlpacaStockClient, + FakeOptionQuoteRecord, + FakeOptionSnapshotRecord, + FakeOptionTradeRecord, + FakeStockTradeRecord, +) from tests.data_sources._fakes.polygon import ( FakePolygonAPI, make_agg, @@ -421,6 +429,7 @@ def test_writes_new_contract_and_snapshot(self) -> None: _client=client, _session_factory=sf, _repo=FakeRunRepo(), + **_empty_alpaca_clients(), ) with sf() as sess: @@ -447,6 +456,7 @@ def test_updates_last_seen_at_on_known_contract(self) -> None: _client=client, _session_factory=sf, _repo=FakeRunRepo(), + **_empty_alpaca_clients(), ) with sf() as sess: @@ -458,6 +468,7 @@ def test_updates_last_seen_at_on_known_contract(self) -> None: _client=client, _session_factory=sf, _repo=FakeRunRepo(), + **_empty_alpaca_clients(), ) with sf() as sess: @@ -483,6 +494,7 @@ def test_idempotent_snapshots(self) -> None: _session_factory=sf, _repo=FakeRunRepo(), _snapshot_ts=fixed_ts, + **_empty_alpaca_clients(), ) options.collect_options_chains( ticker_scope=["AAPL"], @@ -490,6 +502,7 @@ def test_idempotent_snapshots(self) -> None: _session_factory=sf, _repo=FakeRunRepo(), _snapshot_ts=fixed_ts, + **_empty_alpaca_clients(), ) with sf() as sess: @@ -516,6 +529,7 @@ def list_snapshot_options_chain(self, underlying: str) -> list[Any]: _client=client, _session_factory=sf, _repo=repo, + **_empty_alpaca_clients(), ) assert repo.failed() @@ -523,6 +537,228 @@ def list_snapshot_options_chain(self, underlying: str) -> list[Any]: assert sess.query(OptionsContracts).count() == 0 +def _quoteless_snapshot(**overrides: Any) -> Any: + """The production Polygon plan shape: greeks/OI present, NBBO/spot withheld.""" + return make_option_snapshot( + bid=None, ask=None, last_price=None, underlying_price=None, **overrides + ) + + +def _empty_alpaca_clients() -> dict[str, Any]: + """Hermetic no-data Alpaca fakes — keeps tests off the real SDK clients + (and the network) even when ``ALPACA_PAPER_*`` env vars are set.""" + return { + "_alpaca_option_client": FakeAlpacaOptionClient(), + "_alpaca_stock_client": FakeAlpacaStockClient(), + } + + +class TestCollectOptionsChainsAlpacaMerge: + """ALP-949 — Alpaca indicative quotes + IEX spot merged into Polygon rows.""" + + def test_alpaca_fills_quote_columns_on_quoteless_polygon_rows(self) -> None: + """Quote-less Polygon snapshots gain bid/ask/last_price from the + Alpaca chain (matched via the bare OCC symbol — Polygon's ``O:`` + prefix stripped) and the IEX spot on every row; a contract absent + from the Alpaca chain keeps None quote fields. Source records the + merge for the whole underlying.""" + from alphamind.data_sources.polygon import options + + sf, _ = _make_db() + _seed_asset_universe(sf, ["AAPL"], []) + + polygon_client = FakePolygonAPI( + snapshots_by_underlying={ + "AAPL": [ + _quoteless_snapshot(), + _quoteless_snapshot( + contract_ticker="O:AAPL260117P00200000", contract_type="put" + ), + ] + } + ) + alpaca_option_client = FakeAlpacaOptionClient( + chains_by_underlying={ + "AAPL": { + "AAPL260117C00200000": FakeOptionSnapshotRecord( + latest_quote=FakeOptionQuoteRecord(bid_price=5.0, ask_price=5.5), + latest_trade=FakeOptionTradeRecord(price=5.2), + ) + } + } + ) + alpaca_stock_client = FakeAlpacaStockClient( + trades_by_symbol={"AAPL": FakeStockTradeRecord(price=195.5)} + ) + + options.collect_options_chains( + ticker_scope=["AAPL"], + _client=polygon_client, + _session_factory=sf, + _repo=FakeRunRepo(), + _alpaca_option_client=alpaca_option_client, + _alpaca_stock_client=alpaca_stock_client, + ) + + with sf() as sess: + rows = {r.contract_ticker: r for r in sess.query(OptionsContractSnapshots).all()} + assert len(rows) == 2 + call_row = rows["O:AAPL260117C00200000"] + assert call_row.bid == pytest.approx(5.0) + assert call_row.ask == pytest.approx(5.5) + assert call_row.last_price == pytest.approx(5.2) + put_row = rows["O:AAPL260117P00200000"] + assert (put_row.bid, put_row.ask, put_row.last_price) == (None, None, None) + for row in rows.values(): + assert row.underlying_price == pytest.approx(195.5) + assert row.source == "polygon+alpaca" + + def test_polygon_delivered_values_win_per_field(self) -> None: + """Under a future Polygon plan upgrade the Polygon-parsed bid/ask/ + last_price/underlying_price survive the merge unchanged.""" + from alphamind.data_sources.polygon import options + + sf, _ = _make_db() + _seed_asset_universe(sf, ["AAPL"], []) + + polygon_client = FakePolygonAPI(snapshots_by_underlying={"AAPL": [make_option_snapshot()]}) + alpaca_option_client = FakeAlpacaOptionClient( + chains_by_underlying={ + "AAPL": { + "AAPL260117C00200000": FakeOptionSnapshotRecord( + latest_quote=FakeOptionQuoteRecord(bid_price=9.0, ask_price=9.9), + latest_trade=FakeOptionTradeRecord(price=9.5), + ) + } + } + ) + alpaca_stock_client = FakeAlpacaStockClient( + trades_by_symbol={"AAPL": FakeStockTradeRecord(price=200.0)} + ) + + options.collect_options_chains( + ticker_scope=["AAPL"], + _client=polygon_client, + _session_factory=sf, + _repo=FakeRunRepo(), + _alpaca_option_client=alpaca_option_client, + _alpaca_stock_client=alpaca_stock_client, + ) + + with sf() as sess: + row = sess.query(OptionsContractSnapshots).one() + assert row.bid == pytest.approx(5.0) + assert row.ask == pytest.approx(5.5) + assert row.last_price == pytest.approx(5.25) + assert row.underlying_price == pytest.approx(195.0) + assert row.source == "polygon+alpaca" + + def test_alpaca_chain_failure_isolated_per_underlying(self) -> None: + """A chain-fetch failure for one underlying writes its quote-less + Polygon rows with ``source`` ``polygon`` and leaves the other + underlyings in the run fully merged.""" + from alpaca.common.exceptions import APIError + + from alphamind.data_sources.polygon import options + + sf, _ = _make_db() + _seed_asset_universe(sf, ["AAPL", "MSFT"], []) + + polygon_client = FakePolygonAPI( + snapshots_by_underlying={ + "AAPL": [_quoteless_snapshot()], + "MSFT": [ + _quoteless_snapshot( + contract_ticker="O:MSFT260117C00400000", underlying="MSFT", strike=400.0 + ) + ], + } + ) + + class _ChainFailsForAAPL(FakeAlpacaOptionClient): + def get_option_chain(self, request_params: Any) -> dict[str, Any]: + if request_params.underlying_symbol == "AAPL": + raise APIError("boom") # type: ignore[no-untyped-call] + return super().get_option_chain(request_params) + + alpaca_option_client = _ChainFailsForAAPL( + chains_by_underlying={ + "MSFT": { + "MSFT260117C00400000": FakeOptionSnapshotRecord( + latest_quote=FakeOptionQuoteRecord(bid_price=12.0, ask_price=12.6), + ) + } + } + ) + alpaca_stock_client = FakeAlpacaStockClient( + trades_by_symbol={ + "AAPL": FakeStockTradeRecord(price=195.5), + "MSFT": FakeStockTradeRecord(price=410.0), + } + ) + + options.collect_options_chains( + ticker_scope=["AAPL", "MSFT"], + _client=polygon_client, + _session_factory=sf, + _repo=FakeRunRepo(), + _alpaca_option_client=alpaca_option_client, + _alpaca_stock_client=alpaca_stock_client, + ) + + with sf() as sess: + rows = {r.contract_ticker: r for r in sess.query(OptionsContractSnapshots).all()} + assert len(rows) == 2 + failed = rows["O:AAPL260117C00200000"] + assert (failed.bid, failed.ask, failed.last_price) == (None, None, None) + assert failed.source == "polygon" + # The batch-level spot still stamps the failed underlying's rows. + assert failed.underlying_price == pytest.approx(195.5) + merged = rows["O:MSFT260117C00400000"] + assert merged.bid == pytest.approx(12.0) + assert merged.ask == pytest.approx(12.6) + assert merged.underlying_price == pytest.approx(410.0) + assert merged.source == "polygon+alpaca" + + def test_spot_fetch_failure_leaves_underlying_price_none_without_aborting(self) -> None: + from alpaca.common.exceptions import APIError + + from alphamind.data_sources.polygon import options + + sf, _ = _make_db() + _seed_asset_universe(sf, ["AAPL"], []) + + polygon_client = FakePolygonAPI(snapshots_by_underlying={"AAPL": [_quoteless_snapshot()]}) + alpaca_option_client = FakeAlpacaOptionClient( + chains_by_underlying={ + "AAPL": { + "AAPL260117C00200000": FakeOptionSnapshotRecord( + latest_quote=FakeOptionQuoteRecord(bid_price=5.0, ask_price=5.5), + ) + } + } + ) + + class _SpotFails(FakeAlpacaStockClient): + def get_stock_latest_trade(self, request_params: Any) -> dict[str, Any]: + raise APIError("spot down") # type: ignore[no-untyped-call] + + options.collect_options_chains( + ticker_scope=["AAPL"], + _client=polygon_client, + _session_factory=sf, + _repo=FakeRunRepo(), + _alpaca_option_client=alpaca_option_client, + _alpaca_stock_client=_SpotFails(), + ) + + with sf() as sess: + row = sess.query(OptionsContractSnapshots).one() + assert row.underlying_price is None + assert row.bid == pytest.approx(5.0) + assert row.source == "polygon+alpaca" + + # --------------------------------------------------------------------------- # corporate_actions.py # --------------------------------------------------------------------------- From 47bcd89bb918354f2310fac6f42e387c7f662cdd Mon Sep 17 00:00:00 2001 From: Jackson Atassi Date: Thu, 11 Jun 2026 19:25:07 -0600 Subject: [PATCH 2/3] fix: address code-review findings on ALP-949 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Move the batch spot fetch inside track_run so an unexpected exception still writes a failed-run record to collection_runs. - Add ValueError to ALPACA_FETCH_ERRORS: alpaca-py raises it on unexpected response shapes (and pydantic ValidationError subclasses it) — a vendor format change must degrade, not abort the run. The spot-failure test now pins this member. - Install a socket-level timeout on both Alpaca data clients (alpaca-py issues blocking requests calls with no timeout); mirrors the execution layer's client factory shim (ALP-841) with a test in the same shape. - Hoist the per-contract `alpaca_quotes or {}` allocation out of the 10k-iteration row loop. Co-Authored-By: Claude Fable 5 --- src/alphamind/data_sources/alpaca/options.py | 34 ++++++++++++++++-- src/alphamind/data_sources/polygon/options.py | 6 ++-- tests/data_sources/alpaca/test_options.py | 35 +++++++++++++++++++ tests/data_sources/polygon/test_polygon.py | 6 ++-- 4 files changed, 73 insertions(+), 8 deletions(-) diff --git a/src/alphamind/data_sources/alpaca/options.py b/src/alphamind/data_sources/alpaca/options.py index df05bc589..85847afe4 100644 --- a/src/alphamind/data_sources/alpaca/options.py +++ b/src/alphamind/data_sources/alpaca/options.py @@ -18,6 +18,7 @@ from __future__ import annotations +import functools import os from dataclasses import dataclass from typing import Any @@ -36,11 +37,36 @@ # Exceptions a chain/spot fetch raises on vendor or transport failure — # callers degrade to Polygon-only rows on these rather than aborting. +# ``ValueError`` covers alpaca-py's unexpected-response-shape error ("The +# data in response does not match any known keys") and pydantic response +# validation (``ValidationError`` subclasses ``ValueError``) — a vendor +# format change must degrade, not abort the collection run. ALPACA_FETCH_ERRORS: tuple[type[Exception], ...] = ( APIError, requests.exceptions.RequestException, + ValueError, ) +# alpaca-py issues blocking ``requests`` calls without a timeout, so a hung +# connection would wedge the collector task forever. Mirrors the execution +# layer's client factory (``broker_adapter/client_factory.py``, ALP-841) — +# duplicated because ``data_sources`` and ``execution`` are independent +# siblings under the import-linter layer contract. +_SOCKET_TIMEOUT_SECONDS = 60.0 + + +def _install_socket_timeout(client: Any) -> None: + """Default a connect/read timeout onto a REST client's ``requests`` session.""" + session: Any = client._session + original_request = session.request + + @functools.wraps(original_request) + def request_with_timeout(*args: Any, **kwargs: Any) -> Any: + kwargs.setdefault("timeout", _SOCKET_TIMEOUT_SECONDS) + return original_request(*args, **kwargs) + + session.request = request_with_timeout + @dataclass(frozen=True) class OptionQuote: @@ -58,18 +84,22 @@ def make_option_client() -> OptionHistoricalDataClient: mirrors the :class:`~alphamind.data_sources.polygon.client.PolygonClient` convention. Raises ``ValueError`` (from the SDK) when the keys are unset. """ - return OptionHistoricalDataClient( + client = OptionHistoricalDataClient( api_key=os.environ.get("ALPACA_PAPER_KEY"), secret_key=os.environ.get("ALPACA_PAPER_SECRET"), ) + _install_socket_timeout(client) + return client def make_stock_client() -> StockHistoricalDataClient: """Build the stock-trade client from ``ALPACA_PAPER_*`` env vars.""" - return StockHistoricalDataClient( + client = StockHistoricalDataClient( api_key=os.environ.get("ALPACA_PAPER_KEY"), secret_key=os.environ.get("ALPACA_PAPER_SECRET"), ) + _install_socket_timeout(client) + return client def fetch_chain_quotes(underlying: str, *, _client: Any = None) -> dict[str, OptionQuote]: diff --git a/src/alphamind/data_sources/polygon/options.py b/src/alphamind/data_sources/polygon/options.py index 21309ba39..e54a69190 100644 --- a/src/alphamind/data_sources/polygon/options.py +++ b/src/alphamind/data_sources/polygon/options.py @@ -167,9 +167,8 @@ def collect_options_chains( def _fetch_chain(underlying: str) -> list[Any]: return _client.list_snapshot_options_chain(underlying) # type: ignore[no-any-return] - spot_by_ticker = _fetch_spot_map(ticker_scope, _alpaca_stock_client) - with track_run("polygon.options", _repo=_repo) as run: + spot_by_ticker = _fetch_spot_map(ticker_scope, _alpaca_stock_client) rows_written = 0 single = len(ticker_scope) == 1 @@ -222,6 +221,7 @@ def _build_rows( instead of ``polygon+alpaca``. """ snapshot_source = "polygon" if alpaca_quotes is None else "polygon+alpaca" + quote_lookup: Mapping[str, OptionQuote] = alpaca_quotes or {} contract_rows: list[dict[str, Any]] = [] snapshot_rows: list[dict[str, Any]] = [] for snap in snapshots: @@ -231,7 +231,7 @@ def _build_rows( lq = snap.last_quote lt = snap.last_trade ua = snap.underlying_asset - aq = (alpaca_quotes or {}).get(details.ticker.removeprefix("O:")) + aq = quote_lookup.get(details.ticker.removeprefix("O:")) contract_rows.append( { diff --git a/tests/data_sources/alpaca/test_options.py b/tests/data_sources/alpaca/test_options.py index d21a2c726..6388f95b9 100644 --- a/tests/data_sources/alpaca/test_options.py +++ b/tests/data_sources/alpaca/test_options.py @@ -68,6 +68,41 @@ def test_requests_full_chain_on_the_indicative_feed(self) -> None: # --------------------------------------------------------------------------- +class TestInstallSocketTimeout: + def test_defaults_a_timeout_on_session_request(self) -> None: + """alpaca-py issues blocking ``requests`` calls with no ``timeout``; + the factories wrap the session's ``request`` so a stalled socket + cannot wedge the collector task forever (mirrors the execution + layer's client factory — ALP-841).""" + from typing import Any, cast + + import requests + + from alphamind.data_sources.alpaca.options import ( + _SOCKET_TIMEOUT_SECONDS, + _install_socket_timeout, + ) + + captured: dict[str, Any] = {} + + class _RecordingSession(requests.Session): + def request(self, *args: Any, **kwargs: Any) -> Any: + captured.update(kwargs) + return "ok" + + client = cast(Any, type("FakeRest", (), {})()) + client._session = _RecordingSession() + + _install_socket_timeout(client) + + assert client._session.request("GET", "https://example/x") == "ok" + assert captured["timeout"] == _SOCKET_TIMEOUT_SECONDS + + captured.clear() + client._session.request("GET", "https://example/x", timeout=5.0) + assert captured["timeout"] == 5.0 + + class TestFetchUnderlyingTrades: def test_one_multi_symbol_iex_request_maps_prices_and_omits_missing(self) -> None: from alpaca.data.enums import DataFeed diff --git a/tests/data_sources/polygon/test_polygon.py b/tests/data_sources/polygon/test_polygon.py index ded8edbc1..4763e68b2 100644 --- a/tests/data_sources/polygon/test_polygon.py +++ b/tests/data_sources/polygon/test_polygon.py @@ -721,8 +721,8 @@ def get_option_chain(self, request_params: Any) -> dict[str, Any]: assert merged.source == "polygon+alpaca" def test_spot_fetch_failure_leaves_underlying_price_none_without_aborting(self) -> None: - from alpaca.common.exceptions import APIError - + """Raises ValueError — alpaca-py's unexpected-response-shape error — + which must degrade like any vendor failure, not abort the run.""" from alphamind.data_sources.polygon import options sf, _ = _make_db() @@ -741,7 +741,7 @@ def test_spot_fetch_failure_leaves_underlying_price_none_without_aborting(self) class _SpotFails(FakeAlpacaStockClient): def get_stock_latest_trade(self, request_params: Any) -> dict[str, Any]: - raise APIError("spot down") # type: ignore[no-untyped-call] + raise ValueError("The data in response does not match any known keys.") options.collect_options_chains( ticker_scope=["AAPL"], From 269456365bbc6ea25db5cc715925814c856a494c Mon Sep 17 00:00:00 2001 From: Jackson Atassi Date: Thu, 11 Jun 2026 19:29:04 -0600 Subject: [PATCH 3/3] fix: address /review findings on ALP-949 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Clarify in the _build_rows docstring that `source` records the merge ran for the underlying, not per-row Alpaca enrichment — a contract absent from the Alpaca chain keeps NULL quote columns under polygon+alpaca. Co-Authored-By: Claude Fable 5 --- src/alphamind/data_sources/polygon/options.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/alphamind/data_sources/polygon/options.py b/src/alphamind/data_sources/polygon/options.py index e54a69190..6cdaee5c5 100644 --- a/src/alphamind/data_sources/polygon/options.py +++ b/src/alphamind/data_sources/polygon/options.py @@ -218,7 +218,10 @@ def _build_rows( never delivers them, so Alpaca always fills; precedence stays correct under a future plan upgrade). ``alpaca_quotes=None`` means the Alpaca chain fetch failed — the snapshot ``source`` then stays ``polygon`` - instead of ``polygon+alpaca``. + instead of ``polygon+alpaca``. ``source`` records that the merge ran + for the underlying, not that a given row carries Alpaca data — a + contract absent from the Alpaca chain keeps NULL quote columns under + ``source`` ``polygon+alpaca``. """ snapshot_source = "polygon" if alpaca_quotes is None else "polygon+alpaca" quote_lookup: Mapping[str, OptionQuote] = alpaca_quotes or {}