Complete the market data backend per planning docs

backend/CLAUDE.md

@@ -12,24 +12,24 @@ uv sync --extra dev   # Install all dependencies including test/lint tools
 The market data subsystem lives in `app/market/`. Use these imports:
 
 ```python
-from app.market import PriceCache, PriceUpdate, MarketDataSource, create_market_data_source
+from app.market import PriceCache, PriceUpdate, PricePoint, SourceStatus, MarketDataSource, create_market_data_source
 ```
 
 ### Core Types
 
-- **`PriceUpdate`** — Immutable dataclass: `ticker`, `price`, `previous_price`, `timestamp`, plus properties `change`, `change_percent`, `direction` ("up"/"down"/"flat"), and `to_dict()` for JSON serialization.
+- **`PriceUpdate`** — Immutable dataclass: `ticker`, `price`, `previous_price` (previous tick), `open_price` (session baseline), `timestamp`, plus properties `change`/`change_percent` (tick-to-tick), `tick_direction` ("up"/"down"/"flat", drives the flash animation), `change_today`/`change_percent_today` (vs. `open_price`, drives the daily % column), and `to_dict()` for JSON serialization (timestamp is ISO 8601 UTC on the wire).
 
 - **`PriceCache`** — Thread-safe in-memory store. Key methods:
-  - `update(ticker, price, timestamp=None) -> PriceUpdate`
+  - `update(ticker, price, timestamp=None, open_price=None) -> PriceUpdate` — `open_price` sticks for the session once set; omit it on subsequent calls to keep the existing baseline.
   - `get(ticker) -> PriceUpdate | None`
   - `get_price(ticker) -> float | None`
   - `get_all() -> dict[str, PriceUpdate]`
   - `remove(ticker)`
   - `version` property — monotonic counter, increments on every update (for SSE change detection)
 
-- **`MarketDataSource`** — Abstract interface implemented by `SimulatorDataSource` and `MassiveDataSource`. Lifecycle: `start(tickers)` -> `add_ticker()` / `remove_ticker()` -> `stop()`.
+- **`MarketDataSource`** — Abstract interface implemented by `SimulatorDataSource`, `AnchoredSimulatorDataSource`, and `MassiveDataSource`. Lifecycle: `start(tickers)` -> `add_ticker()` / `remove_ticker()` -> `stop()`. Also: `describe() -> SourceStatus` (for `GET /api/health`) and `get_history(ticker, points=120) -> list[PricePoint]` (for chart first-paint; default `[]`).
 
-- **`create_market_data_source(cache)`** — Factory. Returns `MassiveDataSource` if `MASSIVE_API_KEY` is set, otherwise `SimulatorDataSource`.
+- **`create_market_data_source(cache)`** — **Async** factory; must be awaited from the FastAPI lifespan handler before `start()`. Selection is driven by a one-time Massive entitlement probe (`capabilities.py`), not just key presence — a key that authenticates but isn't entitled to live prices (free/Basic tier) routes to `AnchoredSimulatorDataSource` (real closing prices, synthetic motion) rather than a `MassiveDataSource` that would poll forever and never populate the cache. See `planning/MARKET_INTERFACE.md` for the full decision table.
 
 ### SSE Streaming
 

backend/README.md

@@ -6,13 +6,15 @@ FastAPI backend for the FinAlly AI Trading Workstation.
 
 - `app/` - Application code
   - `market/` - Market data subsystem
-    - `models.py` - PriceUpdate dataclass
+    - `models.py` - PriceUpdate/PricePoint/SourceStatus dataclasses
     - `cache.py` - Thread-safe price cache
     - `interface.py` - MarketDataSource abstract interface
+    - `capabilities.py` - Massive API entitlement probe
     - `simulator.py` - GBM-based market simulator
-    - `massive_client.py` - Massive/Polygon.io API client
-    - `factory.py` - Data source factory
-    - `stream.py` - SSE streaming endpoint
+    - `anchored.py` - Simulator seeded from real Massive closing prices (free-tier keys)
+    - `massive_client.py` - Massive/Polygon.io API client (real-time tiers)
+    - `factory.py` - Async, capability-driven data source factory
+    - `stream.py` - SSE streaming endpoint (with heartbeat)
     - `seed_prices.py` - Default ticker prices and parameters
 
 - `tests/` - Unit and integration tests

backend/app/market/__init__.py

@@ -2,20 +2,25 @@
 
 Public API:
     PriceUpdate         - Immutable price snapshot dataclass
+    PricePoint          - Single point in a historical price series
+    SourceStatus        - Introspection payload for GET /api/health
     PriceCache          - Thread-safe in-memory price store
     MarketDataSource    - Abstract interface for data providers
-    create_market_data_source - Factory that selects simulator or Massive
+    create_market_data_source - Async factory; probes Massive entitlement and
+                                 selects simulator / anchored-simulator / massive
     create_stream_router - FastAPI router factory for SSE endpoint
 """
 
 from .cache import PriceCache
 from .factory import create_market_data_source
 from .interface import MarketDataSource
-from .models import PriceUpdate
+from .models import PricePoint, PriceUpdate, SourceStatus
 from .stream import create_stream_router
 
 __all__ = [
     "PriceUpdate",
+    "PricePoint",
+    "SourceStatus",
     "PriceCache",
     "MarketDataSource",
     "create_market_data_source",

backend/app/market/anchored.py

@@ -0,0 +1,173 @@
+"""GBM simulation seeded from real Massive closing prices.
+
+Bridges the gap for Basic-tier (free) Massive keys, which authenticate fine but
+are not entitled to any live price. One free-tier API call
+(`get_grouped_daily_aggs`) prices the whole market at once, so the simulator
+starts from real closing levels — e.g. AAPL at its genuine close instead of a
+hard-coded, rapidly stale seed — while GBM supplies the tick-to-tick motion so
+the terminal still looks alive. See planning/MARKET_INTERFACE.md §6.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import date, timedelta
+
+from massive import RESTClient
+
+from .cache import PriceCache
+from .interface import MarketDataSource
+from .models import PricePoint, SourceStatus, epoch_to_iso
+from .simulator import SimulatorDataSource
+
+logger = logging.getLogger(__name__)
+
+MAX_ANCHOR_LOOKBACK_DAYS = 7  # walk back at most a week to skip weekends/holidays
+DEFAULT_REANCHOR_INTERVAL_SECONDS = 3600.0  # re-anchor hourly so a long-running container
+# tracks the next session's close instead of drifting from reality
+
+
+class AnchoredSimulatorDataSource(MarketDataSource):
+    """GBM simulation seeded from real Massive closing prices.
+
+    Bridges the gap for Basic-tier keys: real price *levels* from one
+    free-tier API call, plus synthetic price *motion* so the terminal is alive.
+    Displayed prices are simulated, not live — describe().live is always False.
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        price_cache: PriceCache,
+        update_interval: float = 0.5,
+        reanchor_interval: float = DEFAULT_REANCHOR_INTERVAL_SECONDS,
+    ) -> None:
+        self._client = RESTClient(api_key=api_key, retries=0)
+        self._cache = price_cache
+        self._update_interval = update_interval
+        self._reanchor_interval = reanchor_interval
+        self._sim: SimulatorDataSource | None = None
+        self._anchors: dict[str, float] = {}
+        self._anchor_date: str | None = None
+        self._reanchor_task: asyncio.Task | None = None
+
+    async def start(self, tickers: list[str]) -> None:
+        self._anchors, self._anchor_date = await asyncio.to_thread(
+            self._fetch_anchors, tickers
+        )
+        # Real closes where we have them; the static seed table covers the rest.
+        self._sim = SimulatorDataSource(
+            self._cache,
+            update_interval=self._update_interval,
+            seed_overrides=self._anchors,
+            status_detail=self._status_detail(),
+        )
+        await self._sim.start(tickers)
+
+        if self._reanchor_interval > 0:
+            self._reanchor_task = asyncio.create_task(
+                self._reanchor_loop(), name="anchored-reanchor"
+            )
+
+    async def stop(self) -> None:
+        if self._reanchor_task and not self._reanchor_task.done():
+            self._reanchor_task.cancel()
+            try:
+                await self._reanchor_task
+            except asyncio.CancelledError:
+                pass
+        self._reanchor_task = None
+        if self._sim:
+            await self._sim.stop()
+
+    async def add_ticker(self, ticker: str) -> None:
+        # Costs nothing: the underlying simulator synthesizes a seed for any
+        # symbol not in the anchor set, so the demo never dead-ends on an
+        # unknown ticker.
+        if self._sim:
+            await self._sim.add_ticker(ticker)
+
+    async def remove_ticker(self, ticker: str) -> None:
+        if self._sim:
+            await self._sim.remove_ticker(ticker)
+
+    def get_tickers(self) -> list[str]:
+        return self._sim.get_tickers() if self._sim else []
+
+    def describe(self) -> SourceStatus:
+        return SourceStatus(
+            name="anchored-simulator",
+            live=False,
+            detail=self._status_detail(),
+            tickers=len(self.get_tickers()),
+            cache_populated=len(self._cache) > 0,
+        )
+
+    async def get_history(self, ticker: str, points: int = 120) -> list[PricePoint]:
+        """Real minute bars from the last completed session (free tier allows this)."""
+        if not self._anchor_date:
+            return []
+        try:
+            bars = await asyncio.to_thread(
+                self._client.get_aggs,
+                ticker,
+                1,
+                "minute",
+                self._anchor_date,
+                self._anchor_date,
+                limit=50_000,
+            )
+        except Exception as e:
+            logger.warning("Anchored get_history failed for %s: %s", ticker, e)
+            return []
+        return [PricePoint(epoch_to_iso(bar.timestamp / 1000), bar.close) for bar in bars[-points:]]
+
+    # --- Internal ---
+
+    def _fetch_anchors(self, tickers: list[str]) -> tuple[dict[str, float], str | None]:
+        """ONE API call prices every ticker. Walks back over weekends/holidays."""
+        wanted = {t.upper() for t in tickers}
+        day = date.today()
+        for _ in range(MAX_ANCHOR_LOOKBACK_DAYS):
+            day -= timedelta(days=1)
+            iso = day.isoformat()
+            try:
+                bars = self._client.get_grouped_daily_aggs(iso, adjusted=True)
+            except Exception as e:
+                logger.warning("Anchor fetch failed for %s: %s", iso, e)
+                continue
+            if not bars:
+                continue  # weekend or holiday
+            found = {bar.ticker: bar.close for bar in bars if bar.ticker in wanted}
+            logger.info(
+                "Anchored %d/%d tickers to %s closes", len(found), len(wanted), iso
+            )
+            return found, iso
+        logger.warning("No anchors available — falling back to the static seed table")
+        return {}, None
+
+    def _status_detail(self) -> str:
+        if self._anchor_date:
+            return (
+                f"simulated from real {self._anchor_date} closes "
+                f"({len(self._anchors)} anchored)"
+            )
+        return "simulated from static seed prices (anchor fetch failed)"
+
+    async def _reanchor_loop(self) -> None:
+        """Periodically re-fetch anchors so a long-running container tracks the
+        next session's close instead of drifting from reality."""
+        while True:
+            await asyncio.sleep(self._reanchor_interval)
+            try:
+                tickers = self.get_tickers()
+                anchors, anchor_date = await asyncio.to_thread(
+                    self._fetch_anchors, tickers
+                )
+                if anchors:
+                    self._anchors, self._anchor_date = anchors, anchor_date
+                    if self._sim:
+                        self._sim.set_status_detail(self._status_detail())
+            except Exception:
+                logger.exception("Re-anchor failed")

backend/app/market/cache.py

@@ -11,30 +11,48 @@
 class PriceCache:
     """Thread-safe in-memory cache of the latest price for each ticker.
 
-    Writers: SimulatorDataSource or MassiveDataSource (one at a time).
-    Readers: SSE streaming endpoint, portfolio valuation, trade execution.
+    Writers: SimulatorDataSource, AnchoredSimulatorDataSource, or MassiveDataSource
+    (one at a time). Readers: SSE streaming endpoint, portfolio valuation, trade execution.
     """
 
     def __init__(self) -> None:
         self._prices: dict[str, PriceUpdate] = {}
         self._lock = Lock()
         self._version: int = 0  # Monotonically increasing; bumped on every update
 
-    def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate:
+    def update(
+        self,
+        ticker: str,
+        price: float,
+        timestamp: float | None = None,
+        open_price: float | None = None,
+    ) -> PriceUpdate:
         """Record a new price for a ticker. Returns the created PriceUpdate.
 
-        Automatically computes direction and change from the previous price.
-        If this is the first update for the ticker, previous_price == price (direction='flat').
+        Automatically computes direction and change from the previous tick.
+        If this is the first update for the ticker, previous_price == price (tick_direction='flat').
+
+        `open_price` is the session baseline used for the daily change column. When
+        omitted, the ticker keeps whatever open_price it already had; on the very
+        first write it defaults to `price`.
         """
         with self._lock:
             ts = timestamp or time.time()
             prev = self._prices.get(ticker)
             previous_price = prev.price if prev else price
 
+            if open_price is not None:
+                resolved_open = open_price
+            elif prev is not None:
+                resolved_open = prev.open_price
+            else:
+                resolved_open = price
+
             update = PriceUpdate(
                 ticker=ticker,
                 price=round(price, 2),
                 previous_price=round(previous_price, 2),
+                open_price=round(resolved_open, 2),
                 timestamp=ts,
             )
             self._prices[ticker] = update

backend/app/market/capabilities.py

@@ -0,0 +1,60 @@
+"""Entitlement probing for the Massive API.
+
+A Massive API key can be valid but still unable to return live prices — a free
+Basic-tier key authenticates successfully and then returns NOT_AUTHORIZED for
+every snapshot/last-trade endpoint, while end-of-day aggregate endpoints work
+fine. This module answers "what can this key actually do?" once at startup so
+the factory can route to a source that will actually produce prices, instead
+of a source that quietly fails on every poll.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+import urllib3.exceptions
+from massive import RESTClient
+from massive.exceptions import AuthError, BadResponse
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class MassiveCapabilities:
+    """What a given API key is actually allowed to do."""
+
+    valid: bool  # key authenticates at all
+    realtime: bool  # snapshot / last-trade endpoints entitled
+    end_of_day: bool  # aggregate endpoints entitled
+    detail: str
+
+
+def probe_capabilities(api_key: str) -> MassiveCapabilities:
+    """Two cheap calls, run once at startup. Costs 2 of the free tier's 5/min budget.
+
+    Never raises — every failure mode is captured in the returned MassiveCapabilities.
+    """
+    try:
+        client = RESTClient(api_key=api_key, retries=0, read_timeout=5.0)
+    except AuthError:
+        return MassiveCapabilities(False, False, False, "no API key configured")
+
+    # 1. Cheapest possible entitlement test for real-time.
+    try:
+        client.get_snapshot_all(market_type="stocks", tickers=["AAPL"])
+        return MassiveCapabilities(True, True, True, "real-time snapshots entitled")
+    except BadResponse as e:
+        if "NOT_AUTHORIZED" not in str(e):
+            return MassiveCapabilities(False, False, False, f"unexpected: {e}")
+    except urllib3.exceptions.MaxRetryError as e:
+        return MassiveCapabilities(False, False, False, f"unreachable/rate-limited: {e}")
+    except Exception as e:  # pragma: no cover - defensive catch-all, never raise
+        return MassiveCapabilities(False, False, False, f"unexpected: {e}")
+
+    # 2. Snapshots refused — is this a valid key on a lower plan, or a bad key?
+    try:
+        client.get_previous_close_agg("AAPL")
+        return MassiveCapabilities(True, False, True, "end-of-day only (Basic tier)")
+    except Exception as e:
+        return MassiveCapabilities(False, False, False, f"key rejected: {e}")