Close the four outstanding market data gaps from MARKET_DATA_DESIGN.md

backend/CLAUDE.md

@@ -24,7 +24,8 @@ from app.market import PriceCache, PriceUpdate, MarketDataSource, create_market_
   - `get(ticker) -> PriceUpdate | None`
   - `get_price(ticker) -> float | None`
   - `get_all() -> dict[str, PriceUpdate]`
-  - `remove(ticker)`
+  - `get_history(ticker, limit=HISTORY_MAXLEN) -> list[tuple[float, float]]` — rolling in-memory `(timestamp, price)` points, oldest-first, capped at `HISTORY_MAXLEN` (600, ~5 min at the simulator's 500ms cadence). Empty list for an untracked ticker.
+  - `remove(ticker)` — also clears that ticker's history
   - `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()`.
@@ -40,6 +41,26 @@ router = create_stream_router(price_cache)  # Returns FastAPI APIRouter
 # Endpoint: GET /api/stream/prices (text/event-stream)
 ```
 
+Pushes the entire price cache as one JSON object (map keyed by ticker) whenever
+`PriceCache.version` changes, polled every 500ms. A connecting client always gets
+a full snapshot immediately, including after a reconnect. When the version hasn't
+changed for 15s (`KEEPALIVE_SECONDS`), an SSE comment line (`: ping`) is sent so
+proxies and the frontend's connection indicator don't mistake a quiet market for
+a dead connection.
+
+### Price History
+
+```python
+from app.market import create_history_router
+
+router = create_history_router(price_cache)  # Returns FastAPI APIRouter
+# Endpoint: GET /api/prices/{ticker}/history?limit=600
+```
+
+Backs the main chart's initial backfill from `PriceCache`'s rolling in-memory
+history. Not persisted — a restart clears it, matching the simulator's own
+reset-to-seed behavior.
+
 ### Seed Data
 
 Default tickers: AAPL, GOOGL, MSFT, AMZN, TSLA, NVDA, META, JPM, V, NFLX. Seed prices and per-ticker volatility/drift params are in `app/market/seed_prices.py`.

backend/app/market/__init__.py

@@ -2,22 +2,25 @@
 
 Public API:
     PriceUpdate         - Immutable price snapshot dataclass
-    PriceCache          - Thread-safe in-memory price store
+    PriceCache          - Thread-safe in-memory price store (latest price + rolling history)
     MarketDataSource    - Abstract interface for data providers
     create_market_data_source - Factory that selects simulator or Massive
-    create_stream_router - FastAPI router factory for SSE endpoint
+    create_stream_router - FastAPI router factory for the SSE endpoint
+    create_history_router - FastAPI router factory for the price history endpoint
 """
 
-from .cache import PriceCache
+from .cache import HISTORY_MAXLEN, PriceCache
 from .factory import create_market_data_source
 from .interface import MarketDataSource
 from .models import PriceUpdate
-from .stream import create_stream_router
+from .stream import create_history_router, create_stream_router
 
 __all__ = [
     "PriceUpdate",
     "PriceCache",
+    "HISTORY_MAXLEN",
     "MarketDataSource",
     "create_market_data_source",
     "create_stream_router",
+    "create_history_router",
 ]

backend/app/market/cache.py

@@ -3,20 +3,29 @@
 from __future__ import annotations
 
 import time
+from collections import deque
 from threading import Lock
 
 from .models import PriceUpdate
 
+# ~5 minutes of history at the 500ms simulator cadence. Under Massive's 15s
+# poll interval this fills much more slowly (one point per poll), which is
+# correct: the chart is sparse but accurate rather than padded with guesses.
+HISTORY_MAXLEN = 600
+
 
 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.
+    Readers: SSE streaming endpoint, portfolio valuation, trade execution,
+    the price history endpoint.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, history_maxlen: int = HISTORY_MAXLEN) -> None:
         self._prices: dict[str, PriceUpdate] = {}
+        self._history: dict[str, deque[tuple[float, float]]] = {}
+        self._history_maxlen = history_maxlen
         self._lock = Lock()
         self._version: int = 0  # Monotonically increasing; bumped on every update
 
@@ -25,6 +34,7 @@ def update(self, ticker: str, price: float, timestamp: float | None = None) -> P
 
         Automatically computes direction and change from the previous price.
         If this is the first update for the ticker, previous_price == price (direction='flat').
+        Also appends to the ticker's rolling history (see get_history()).
         """
         with self._lock:
             ts = timestamp or time.time()
@@ -38,6 +48,13 @@ def update(self, ticker: str, price: float, timestamp: float | None = None) -> P
                 timestamp=ts,
             )
             self._prices[ticker] = update
+
+            history = self._history.get(ticker)
+            if history is None:
+                history = deque(maxlen=self._history_maxlen)
+                self._history[ticker] = history
+            history.append((ts, update.price))
+
             self._version += 1
             return update
 
@@ -60,6 +77,19 @@ def remove(self, ticker: str) -> None:
         """Remove a ticker from the cache (e.g., when removed from watchlist)."""
         with self._lock:
             self._prices.pop(ticker, None)
+            self._history.pop(ticker, None)
+
+    def get_history(self, ticker: str, limit: int = HISTORY_MAXLEN) -> list[tuple[float, float]]:
+        """Oldest-first (timestamp, price) points for a ticker.
+
+        Returns an empty list for an untracked ticker rather than raising, so
+        callers (e.g. the chart) can draw nothing instead of erroring.
+        """
+        with self._lock:
+            points = self._history.get(ticker)
+            if not points:
+                return []
+            return list(points)[-limit:]
 
     @property
     def version(self) -> int:

backend/app/market/massive_client.py

@@ -4,15 +4,20 @@
 
 import asyncio
 import logging
+import time
 
 from massive import RESTClient
+from massive.exceptions import AuthError, BadResponse
 from massive.rest.models import SnapshotMarketType
 
 from .cache import PriceCache
 from .interface import MarketDataSource
 
 logger = logging.getLogger(__name__)
 
+# Snapshot last_trade.sip_timestamp is Unix nanoseconds; PriceCache wants seconds.
+NANOS_PER_SECOND = 1_000_000_000
+
 
 class MassiveDataSource(MarketDataSource):
     """MarketDataSource backed by the Massive (Polygon.io) REST API.
@@ -95,30 +100,42 @@ async def _poll_once(self) -> None:
             # The Massive RESTClient is synchronous — run in a thread to
             # avoid blocking the event loop.
             snapshots = await asyncio.to_thread(self._fetch_snapshots)
-            processed = 0
-            for snap in snapshots:
-                try:
-                    price = snap.last_trade.price
-                    # Massive timestamps are Unix milliseconds → convert to seconds
-                    timestamp = snap.last_trade.timestamp / 1000.0
-                    self._cache.update(
-                        ticker=snap.ticker,
-                        price=price,
-                        timestamp=timestamp,
-                    )
-                    processed += 1
-                except (AttributeError, TypeError) as e:
-                    logger.warning(
-                        "Skipping snapshot for %s: %s",
-                        getattr(snap, "ticker", "???"),
-                        e,
-                    )
-            logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers))
-
-        except Exception as e:
-            logger.error("Massive poll failed: %s", e)
-            # Don't re-raise — the loop will retry on the next interval.
-            # Common failures: 401 (bad key), 429 (rate limit), network errors.
+        except AuthError:
+            logger.error("Massive API key rejected — the source does not fall back automatically")
+            raise  # unrecoverable: do not retry on a loop
+        except BadResponse as e:
+            logger.warning("Massive returned an error response: %s", e)
+            return  # transient: retry next interval
+        except Exception:
+            logger.exception("Massive poll failed")
+            return
+
+        processed = self._apply_snapshots(snapshots)
+        logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers))
+
+    def _apply_snapshots(self, snapshots: list) -> int:
+        """Write snapshot data into the cache. Returns the number of tickers updated.
+
+        Extracted from _poll_once so it can be tested directly against real
+        `TickerSnapshot` objects (built via `TickerSnapshot.from_dict(...)`)
+        instead of mocks that would silently accept a misspelled attribute.
+        """
+        processed = 0
+        for snap in snapshots:
+            trade = snap.last_trade
+            if trade is None or trade.price is None:
+                # No trade yet today (pre-market, or an unrecognized symbol
+                # that still made it into the response) — leave it as "—".
+                continue
+            self._cache.update(
+                ticker=snap.ticker,
+                price=trade.price,
+                timestamp=(
+                    trade.sip_timestamp / NANOS_PER_SECOND if trade.sip_timestamp else time.time()
+                ),
+            )
+            processed += 1
+        return processed
 
     def _fetch_snapshots(self) -> list:
         """Synchronous call to the Massive REST API. Runs in a thread."""

backend/app/market/stream.py

@@ -5,16 +5,22 @@
 import asyncio
 import json
 import logging
+import time
 from collections.abc import AsyncGenerator
 
 from fastapi import APIRouter, Request
 from fastapi.responses import StreamingResponse
 
-from .cache import PriceCache
+from .cache import HISTORY_MAXLEN, PriceCache
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/api/stream", tags=["streaming"])
+history_router = APIRouter(prefix="/api/prices", tags=["prices"])
+
+# How long the cache version can go unchanged before we send an SSE comment
+# line to keep the connection (and proxies in between) from timing it out.
+KEEPALIVE_SECONDS = 15.0
 
 
 def create_stream_router(price_cache: PriceCache) -> APIRouter:
@@ -48,20 +54,50 @@ async def stream_prices(request: Request) -> StreamingResponse:
     return router
 
 
+def create_history_router(price_cache: PriceCache) -> APIRouter:
+    """Create the router serving rolling in-memory price history.
+
+    Factory pattern mirrors create_stream_router so the PriceCache is
+    injected without module-level globals.
+    """
+
+    @history_router.get("/{ticker}/history")
+    async def get_price_history(ticker: str, limit: int = HISTORY_MAXLEN) -> dict:
+        """Rolling in-memory price history for the main chart.
+
+        Returns an empty `points` list for an untracked ticker — not a 404 —
+        so the chart draws nothing rather than erroring. Oldest-first,
+        matching what a left-to-right time axis wants.
+        """
+        ticker = ticker.strip().upper()
+        limit = max(1, min(limit, HISTORY_MAXLEN))
+        points = price_cache.get_history(ticker, limit=limit)
+        return {
+            "ticker": ticker,
+            "points": [{"timestamp": ts, "price": price} for ts, price in points],
+        }
+
+    return history_router
+
+
 async def _generate_events(
     price_cache: PriceCache,
     request: Request,
     interval: float = 0.5,
 ) -> AsyncGenerator[str, None]:
     """Async generator that yields SSE-formatted price events.
 
-    Sends all prices every `interval` seconds. Stops when the client
-    disconnects (detected via request.is_disconnected()).
+    Sends all prices every `interval` seconds whenever the cache version has
+    changed. When the version is unchanged for KEEPALIVE_SECONDS, sends an
+    SSE comment line instead — EventSource ignores it, but it keeps the
+    connection (and any proxy in between) from treating a quiet market as a
+    dead connection. Stops when the client disconnects.
     """
     # Tell the client to retry after 1 second if the connection drops
     yield "retry: 1000\n\n"
 
     last_version = -1
+    last_sent = time.monotonic()
     client_ip = request.client.host if request.client else "unknown"
     logger.info("SSE client connected: %s", client_ip)
 
@@ -81,6 +117,10 @@ async def _generate_events(
                     data = {ticker: update.to_dict() for ticker, update in prices.items()}
                     payload = json.dumps(data)
                     yield f"data: {payload}\n\n"
+                    last_sent = time.monotonic()
+            elif time.monotonic() - last_sent >= KEEPALIVE_SECONDS:
+                yield ": ping\n\n"
+                last_sent = time.monotonic()
 
             await asyncio.sleep(interval)
     except asyncio.CancelledError:

backend/scripts/verify_massive.py

@@ -0,0 +1,43 @@
+"""Smoke-test the Massive REST API against a live key.
+
+Run once a real MASSIVE_API_KEY exists — it confirms auth, the multi-ticker
+snapshot request, and the nanosecond-to-second timestamp conversion in one
+pass. Timestamps far in the future mean the divisor regressed; an
+AttributeError means the attribute name regressed (see massive_client.py).
+
+    cd backend && uv run python scripts/verify_massive.py
+"""
+
+import os
+from datetime import UTC, datetime
+
+from massive import RESTClient
+from massive.rest.models import SnapshotMarketType
+
+NANOS_PER_SECOND = 1_000_000_000
+TICKERS = ["AAPL", "GOOGL", "MSFT", "NVDA", "TSLA"]
+
+
+def main() -> None:
+    client = RESTClient(api_key=os.environ["MASSIVE_API_KEY"])
+
+    print(f"market: {client.get_market_status().market}")
+
+    snapshots = client.get_snapshot_all(SnapshotMarketType.STOCKS, TICKERS)
+    print(f"requested {len(TICKERS)}, received {len(snapshots)}")
+
+    for snap in snapshots:
+        trade = snap.last_trade
+        if trade is None or trade.price is None:
+            print(f"{snap.ticker}: no trade data")
+            continue
+        when = datetime.fromtimestamp(trade.sip_timestamp / NANOS_PER_SECOND, UTC)
+        print(f"{snap.ticker}: ${trade.price:.2f} at {when:%Y-%m-%d %H:%M:%S} UTC")
+
+    missing = set(TICKERS) - {s.ticker for s in snapshots}
+    if missing:
+        print(f"absent from response (unknown or untraded): {sorted(missing)}")
+
+
+if __name__ == "__main__":
+    main()