fix(data_sources): merge Alpaca indicative NBBO + IEX spot into the options collector (ALP-949)

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\<YYYY-MM-DD>\<invocation_id>\`.

src/alphamind/data_sources/alpaca/options.py

@@ -0,0 +1,146 @@
+"""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 functools
+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.
+# ``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:
+    """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.
+    """
+    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."""
+    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]:
+    """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

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(
@@ -95,6 +168,7 @@ def _fetch_chain(underlying: str) -> list[Any]:
         return _client.list_snapshot_options_chain(underlying)  # type: ignore[no-any-return]
 
     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
 
@@ -112,7 +186,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 +205,26 @@ 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``. ``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 {}
     contract_rows: list[dict[str, Any]] = []
     snapshot_rows: list[dict[str, Any]] = []
     for snap in snapshots:
@@ -137,6 +234,7 @@ def _build_rows(
         lq = snap.last_quote
         lt = snap.last_trade
         ua = snap.underlying_asset
+        aq = quote_lookup.get(details.ticker.removeprefix("O:"))
 
         contract_rows.append(
             {
@@ -160,9 +258,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 +273,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,
             }
         )