diff --git a/backend/CLAUDE.md b/backend/CLAUDE.md
index 612ff18f5..8df8b871e 100644
--- a/backend/CLAUDE.md
+++ b/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
 
diff --git a/backend/README.md b/backend/README.md
index 7cdd84757..5fe0d4cb9 100644
--- a/backend/README.md
+++ b/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
diff --git a/backend/app/market/__init__.py b/backend/app/market/__init__.py
index 57ad0a121..c6d9c93e8 100644
--- a/backend/app/market/__init__.py
+++ b/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",
diff --git a/backend/app/market/anchored.py b/backend/app/market/anchored.py
new file mode 100644
index 000000000..1c76f745d
--- /dev/null
+++ b/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")
diff --git a/backend/app/market/cache.py b/backend/app/market/cache.py
index 4d0215778..d60354184 100644
--- a/backend/app/market/cache.py
+++ b/backend/app/market/cache.py
@@ -11,8 +11,8 @@
 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:
@@ -20,21 +20,39 @@ def __init__(self) -> None:
         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
diff --git a/backend/app/market/capabilities.py b/backend/app/market/capabilities.py
new file mode 100644
index 000000000..b54f143cb
--- /dev/null
+++ b/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}")
diff --git a/backend/app/market/factory.py b/backend/app/market/factory.py
index 00360e94f..b6bfcd773 100644
--- a/backend/app/market/factory.py
+++ b/backend/app/market/factory.py
@@ -2,10 +2,13 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import os
 
+from .anchored import AnchoredSimulatorDataSource
 from .cache import PriceCache
+from .capabilities import probe_capabilities
 from .interface import MarketDataSource
 from .massive_client import MassiveDataSource
 from .simulator import SimulatorDataSource
@@ -13,19 +16,40 @@
 logger = logging.getLogger(__name__)
 
 
-def create_market_data_source(price_cache: PriceCache) -> MarketDataSource:
-    """Create the appropriate market data source based on environment variables.
+async def create_market_data_source(price_cache: PriceCache) -> MarketDataSource:
+    """Select a market data source. Never raises; always returns a working source.
 
-    - MASSIVE_API_KEY set and non-empty → MassiveDataSource (real market data)
-    - Otherwise → SimulatorDataSource (GBM simulation)
+    Async because, when MASSIVE_API_KEY is set, this makes real network calls to
+    probe entitlement. Call once, from the FastAPI lifespan handler, before
+    start() is invoked.
 
-    Returns an unstarted source. Caller must await source.start(tickers).
+    A Massive key can be valid but unable to return live prices (Basic/free
+    tier is end-of-day only). Selection is therefore driven by a one-time
+    capability probe rather than by key presence alone:
+
+    - No key                    -> SimulatorDataSource (synthetic seeds)
+    - Key, real-time entitled   -> MassiveDataSource (live snapshots)
+    - Key, end-of-day only      -> AnchoredSimulatorDataSource (real closes, synthetic motion)
+    - Key, invalid/rejected     -> SimulatorDataSource (never boot into a broken state)
     """
     api_key = os.environ.get("MASSIVE_API_KEY", "").strip()
 
-    if api_key:
-        logger.info("Market data source: Massive API (real data)")
-        return MassiveDataSource(api_key=api_key, price_cache=price_cache)
-    else:
-        logger.info("Market data source: GBM Simulator")
-        return SimulatorDataSource(price_cache=price_cache)
+    if not api_key:
+        logger.info("No MASSIVE_API_KEY — using GBM simulator")
+        return SimulatorDataSource(price_cache)
+
+    caps = await asyncio.to_thread(probe_capabilities, api_key)
+
+    if caps.realtime:
+        logger.info("Massive: real-time entitled — using live snapshots")
+        return MassiveDataSource(api_key, price_cache, poll_interval=5.0)
+
+    if caps.end_of_day:
+        logger.warning(
+            "Massive key is end-of-day only (Basic tier). Anchoring the simulator "
+            "to real closing prices — displayed prices are simulated, not live."
+        )
+        return AnchoredSimulatorDataSource(api_key, price_cache)
+
+    logger.error("MASSIVE_API_KEY rejected (%s) — falling back to simulator", caps.detail)
+    return SimulatorDataSource(price_cache, status_detail=f"key rejected: {caps.detail}")
diff --git a/backend/app/market/interface.py b/backend/app/market/interface.py
index 0f3b7d8c9..911eb4c6b 100644
--- a/backend/app/market/interface.py
+++ b/backend/app/market/interface.py
@@ -4,6 +4,8 @@
 
 from abc import ABC, abstractmethod
 
+from .models import PricePoint, SourceStatus
+
 
 class MarketDataSource(ABC):
     """Contract for market data providers.
@@ -13,7 +15,7 @@ class MarketDataSource(ABC):
     it reads from the cache.
 
     Lifecycle:
-        source = create_market_data_source(cache)
+        source = await create_market_data_source(cache)
         await source.start(["AAPL", "GOOGL", ...])
         # ... app runs ...
         await source.add_ticker("TSLA")
@@ -55,3 +57,15 @@ async def remove_ticker(self, ticker: str) -> None:
     @abstractmethod
     def get_tickers(self) -> list[str]:
         """Return the current list of actively tracked tickers."""
+
+    @abstractmethod
+    def describe(self) -> SourceStatus:
+        """Introspection for GET /api/health. Must never raise."""
+
+    async def get_history(self, ticker: str, points: int = 120) -> list[PricePoint]:
+        """Historical series for the chart's first paint.
+
+        Default implementation returns [] — the frontend then accumulates from
+        SSE. Sources with real or synthesized history override this.
+        """
+        return []
diff --git a/backend/app/market/massive_client.py b/backend/app/market/massive_client.py
index 00bc7b2aa..5be85d98d 100644
--- a/backend/app/market/massive_client.py
+++ b/backend/app/market/massive_client.py
@@ -1,18 +1,30 @@
-"""Massive (Polygon.io) API client for real market data."""
+"""Massive (Polygon.io) API client for real market data.
+
+Only usable by keys entitled to real-time snapshots (Advanced tier or above,
+$199/mo). Lower tiers authenticate but get NOT_AUTHORIZED on every snapshot
+call — factory.py probes entitlement once at startup and routes those keys to
+AnchoredSimulatorDataSource instead. See planning/MASSIVE_API.md.
+"""
 
 from __future__ import annotations
 
 import asyncio
+import datetime
 import logging
 
+import urllib3.exceptions
 from massive import RESTClient
+from massive.exceptions import AuthError, BadResponse
 from massive.rest.models import SnapshotMarketType
 
 from .cache import PriceCache
 from .interface import MarketDataSource
+from .models import PricePoint, SourceStatus, epoch_to_iso
 
 logger = logging.getLogger(__name__)
 
+NANOSECONDS_PER_SECOND = 1e9
+
 
 class MassiveDataSource(MarketDataSource):
     """MarketDataSource backed by the Massive (Polygon.io) REST API.
@@ -20,27 +32,31 @@ class MassiveDataSource(MarketDataSource):
     Polls GET /v2/snapshot/locale/us/markets/stocks/tickers for all watched
     tickers in a single API call, then writes results to the PriceCache.
 
-    Rate limits:
-      - Free tier: 5 req/min → poll every 15s (default)
-      - Paid tiers: higher limits → poll every 2-5s
+    `retries=0` on the client deliberately: the SDK's default retry/backoff is
+    entirely inside the same rate-limit window, so a retry on 429 is guaranteed
+    to fail too — it just burns budget. Poll-level backoff (the interval itself)
+    is what matters.
     """
 
     def __init__(
         self,
         api_key: str,
         price_cache: PriceCache,
-        poll_interval: float = 15.0,
+        poll_interval: float = 5.0,
     ) -> None:
         self._api_key = api_key
         self._cache = price_cache
         self._interval = poll_interval
         self._tickers: list[str] = []
+        self._unknown_tickers: set[str] = set()
         self._task: asyncio.Task | None = None
         self._client: RESTClient | None = None
+        self._live = False
+        self._last_error: str | None = None
 
     async def start(self, tickers: list[str]) -> None:
-        self._client = RESTClient(api_key=self._api_key)
-        self._tickers = list(tickers)
+        self._client = RESTClient(api_key=self._api_key, retries=0)
+        self._tickers = [t.upper().strip() for t in tickers]
 
         # Do an immediate first poll so the cache has data right away
         await self._poll_once()
@@ -72,12 +88,42 @@ async def add_ticker(self, ticker: str) -> None:
     async def remove_ticker(self, ticker: str) -> None:
         ticker = ticker.upper().strip()
         self._tickers = [t for t in self._tickers if t != ticker]
+        self._unknown_tickers.discard(ticker)
         self._cache.remove(ticker)
         logger.info("Massive: removed ticker %s", ticker)
 
     def get_tickers(self) -> list[str]:
         return list(self._tickers)
 
+    def describe(self) -> SourceStatus:
+        detail = self._last_error or "real-time snapshots"
+        if self._unknown_tickers:
+            detail += f"; not returned by last poll: {', '.join(sorted(self._unknown_tickers))}"
+        return SourceStatus(
+            name="massive",
+            live=self._live,
+            detail=detail,
+            tickers=len(self._tickers),
+            cache_populated=len(self._cache) > 0,
+        )
+
+    async def get_history(self, ticker: str, points: int = 120) -> list[PricePoint]:
+        """Real intraday minute bars for the current session (paid-tier only)."""
+        if not self._client:
+            return []
+        try:
+            bars = await asyncio.to_thread(
+                self._fetch_today_minute_bars, ticker, points
+            )
+        except Exception as e:
+            logger.warning("Massive get_history failed for %s: %s", ticker, e)
+            return []
+        return [PricePoint(epoch_to_iso(bar.timestamp / 1000), bar.close) for bar in bars[-points:]]
+
+    def _fetch_today_minute_bars(self, ticker: str, points: int) -> list:
+        today = datetime.date.today().isoformat()
+        return self._client.get_aggs(ticker, 1, "minute", today, today, limit=max(points, 1))
+
     # --- Internal ---
 
     async def _poll_loop(self) -> None:
@@ -95,30 +141,65 @@ 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 AuthError as e:
+            self._live = False
+            self._last_error = f"auth error: {e}"
+            logger.error("Massive poll failed (auth): %s", e)
+            return
+        except urllib3.exceptions.MaxRetryError as e:
+            # Rate limited or unreachable. Do not retry within this cycle — the
+            # next scheduled poll is the backoff.
+            self._live = False
+            self._last_error = "rate limited or unreachable"
+            logger.warning("Massive poll rate-limited/unreachable: %s", e)
+            return
+        except BadResponse as e:
+            self._live = False
+            self._last_error = str(e)
+            logger.error("Massive poll failed (bad response): %s", e)
+            return
         except Exception as e:
+            self._live = False
+            self._last_error = str(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.
+            return
+
+        processed = 0
+        seen: set[str] = set()
+        for snap in snapshots:
+            ticker = getattr(snap, "ticker", None)
+            if ticker:
+                seen.add(ticker)
+            try:
+                last_trade = snap.last_trade
+                if last_trade is None:
+                    raise AttributeError("snapshot has no last_trade")
+                price = last_trade.price
+                # Massive snapshot timestamps are Unix NANOSECONDS on sip_timestamp
+                # (not `timestamp`, and not milliseconds — see planning/MASSIVE_API.md §6).
+                timestamp = last_trade.sip_timestamp / NANOSECONDS_PER_SECOND
+                open_price = snap.prev_day.close if snap.prev_day else None
+                self._cache.update(
+                    ticker=ticker,
+                    price=price,
+                    timestamp=timestamp,
+                    open_price=open_price,
+                )
+                processed += 1
+            except (AttributeError, TypeError) as e:
+                logger.warning(
+                    "Skipping snapshot for %s: %s",
+                    ticker or "???",
+                    e,
+                )
+
+        # Tickers we asked for but that never appeared in the response —
+        # surfaced via describe() rather than vanishing silently.
+        self._unknown_tickers = {t for t in self._tickers if t not in seen}
+        self._live = processed > 0
+        if processed:
+            self._last_error = None
+        logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers))
 
     def _fetch_snapshots(self) -> list:
         """Synchronous call to the Massive REST API. Runs in a thread."""
diff --git a/backend/app/market/models.py b/backend/app/market/models.py
index de81b1dbc..1f370cd7f 100644
--- a/backend/app/market/models.py
+++ b/backend/app/market/models.py
@@ -4,46 +4,111 @@
 
 import time
 from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+
+def epoch_to_iso(timestamp: float) -> str:
+    """Convert a Unix epoch (seconds) to an ISO 8601 UTC string, e.g. '...T17:42:11.413700Z'.
+
+    Always includes microseconds (unlike `datetime.isoformat()`, which omits them
+    when exactly zero) so every timestamp has a fixed-width format — plain string
+    comparison then agrees with chronological order.
+    """
+    dt = datetime.fromtimestamp(timestamp, tz=timezone.utc)
+    return dt.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z"
 
 
 @dataclass(frozen=True, slots=True)
 class PriceUpdate:
-    """Immutable snapshot of a single ticker's price at a point in time."""
+    """Immutable snapshot of a single ticker's price at a point in time.
+
+    `previous_price` is the previous *tick* (drives the flash animation only).
+    `open_price` is fixed for the session (drives the daily change column) —
+    the seed price for the simulator, the anchor close for the anchored
+    simulator, or the previous day's close for Massive.
+    """
 
     ticker: str
     price: float
     previous_price: float
+    open_price: float
     timestamp: float = field(default_factory=time.time)  # Unix seconds
 
     @property
     def change(self) -> float:
-        """Absolute price change from previous update."""
+        """Absolute price change from the previous tick."""
         return round(self.price - self.previous_price, 4)
 
     @property
     def change_percent(self) -> float:
-        """Percentage change from previous update."""
+        """Percentage change from the previous tick."""
         if self.previous_price == 0:
             return 0.0
         return round((self.price - self.previous_price) / self.previous_price * 100, 4)
 
     @property
-    def direction(self) -> str:
-        """'up', 'down', or 'flat'."""
+    def tick_direction(self) -> str:
+        """'up', 'down', or 'flat' — drives the CSS flash class."""
         if self.price > self.previous_price:
             return "up"
         elif self.price < self.previous_price:
             return "down"
         return "flat"
 
+    @property
+    def change_today(self) -> float:
+        """Absolute change since the session open/anchor."""
+        return round(self.price - self.open_price, 4)
+
+    @property
+    def change_percent_today(self) -> float:
+        """Percentage change since the session open/anchor — drives the daily % column."""
+        if self.open_price == 0:
+            return 0.0
+        return round((self.price - self.open_price) / self.open_price * 100, 4)
+
     def to_dict(self) -> dict:
-        """Serialize for JSON / SSE transmission."""
+        """Serialize for JSON / SSE transmission. Timestamp is ISO 8601 UTC on the wire."""
         return {
             "ticker": self.ticker,
             "price": self.price,
             "previous_price": self.previous_price,
-            "timestamp": self.timestamp,
+            "open_price": self.open_price,
+            "timestamp": epoch_to_iso(self.timestamp),
+            "tick_direction": self.tick_direction,
             "change": self.change,
             "change_percent": self.change_percent,
-            "direction": self.direction,
+            "change_today": self.change_today,
+            "change_percent_today": self.change_percent_today,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class PricePoint:
+    """A single point in a historical price series."""
+
+    timestamp: str  # ISO 8601 UTC
+    price: float
+
+    def to_dict(self) -> dict:
+        return {"timestamp": self.timestamp, "price": self.price}
+
+
+@dataclass(frozen=True, slots=True)
+class SourceStatus:
+    """Introspection for a MarketDataSource, surfaced via GET /api/health."""
+
+    name: str  # "simulator" | "massive" | "anchored-simulator"
+    live: bool  # True only when prices reflect the real current market
+    detail: str  # human-readable, surfaced verbatim in /api/health
+    tickers: int
+    cache_populated: bool
+
+    def to_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "live": self.live,
+            "detail": self.detail,
+            "tickers": self.tickers,
+            "cache_populated": self.cache_populated,
         }
diff --git a/backend/app/market/seed_prices.py b/backend/app/market/seed_prices.py
index 69586df03..9d70a19ff 100644
--- a/backend/app/market/seed_prices.py
+++ b/backend/app/market/seed_prices.py
@@ -1,17 +1,24 @@
-"""Seed prices and per-ticker parameters for the market simulator."""
+"""Seed prices and per-ticker parameters for the market simulator.
 
-# Realistic starting prices for the default watchlist (as of project creation)
+These are the no-key fallback only. When a Massive API key is available,
+AnchoredSimulatorDataSource replaces these levels with real closing prices —
+see planning/MARKET_INTERFACE.md. Refreshed to real 2026-09-02 closes so the
+no-key demo does not show implausible levels (e.g. NVDA and NFLX are badly
+stale after their 2024 splits).
+"""
+
+# Starting prices for the default watchlist, sourced from real 2026-09-02 closes.
 SEED_PRICES: dict[str, float] = {
-    "AAPL": 190.00,
-    "GOOGL": 175.00,
-    "MSFT": 420.00,
-    "AMZN": 185.00,
-    "TSLA": 250.00,
-    "NVDA": 800.00,
-    "META": 500.00,
-    "JPM": 195.00,
-    "V": 280.00,
-    "NFLX": 600.00,
+    "AAPL": 324.96,
+    "GOOGL": 337.12,
+    "MSFT": 496.82,
+    "AMZN": 254.98,
+    "TSLA": 357.01,
+    "NVDA": 224.41,
+    "META": 592.85,
+    "JPM": 356.22,
+    "V": 378.40,
+    "NFLX": 82.73,
 }
 
 # Per-ticker GBM parameters
diff --git a/backend/app/market/simulator.py b/backend/app/market/simulator.py
index b6803f592..2b0946706 100644
--- a/backend/app/market/simulator.py
+++ b/backend/app/market/simulator.py
@@ -3,14 +3,19 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import logging
 import math
 import random
+import time
+from collections import deque
+from dataclasses import dataclass
 
 import numpy as np
 
 from .cache import PriceCache
 from .interface import MarketDataSource
+from .models import PricePoint, SourceStatus, epoch_to_iso
 from .seed_prices import (
     CORRELATION_GROUPS,
     CROSS_GROUP_CORR,
@@ -24,6 +29,33 @@
 
 logger = logging.getLogger(__name__)
 
+# Deterministic synthetic seed range for unknown tickers, e.g. $20.00-$500.00.
+SYNTHETIC_SEED_MIN = 20.0
+SYNTHETIC_SEED_SPAN_CENTS = 48_000  # (SEED_MAX - SEED_MIN) * 100
+
+
+def synthesize_seed(ticker: str) -> float:
+    """Stable pseudo-price for an unknown symbol. Same ticker, same price, always.
+
+    Unlike a random draw, this survives container restarts — a held position's
+    cost basis and the P&L chart stay consistent instead of jumping around.
+    """
+    digest = int(hashlib.sha256(ticker.encode()).hexdigest()[:8], 16)
+    return round(SYNTHETIC_SEED_MIN + (digest % SYNTHETIC_SEED_SPAN_CENTS) / 100.0, 2)
+
+
+@dataclass
+class _Shock:
+    """A transient, decaying price overlay simulating a sudden intraday move.
+
+    Modeled as an additive overlay on top of the GBM path (not a permanent
+    level shift), so a shock is visible as a spike that bleeds off over
+    roughly a minute without permanently distorting the calibrated volatility.
+    """
+
+    magnitude: float  # signed fraction, e.g. -0.03
+    decay: float = 0.985  # per-tick multiplier; ~half-life 46 ticks (~23s at 500ms)
+
 
 class GBMSimulator:
     """Geometric Brownian Motion simulator for correlated stock prices.
@@ -47,19 +79,27 @@ class GBMSimulator:
     TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600  # 5,896,800
     DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR  # ~8.48e-8
 
+    # Default event probability, calibrated so shocks add drama without
+    # swamping the calibrated sigma: ~1.2 events per 10-minute demo across a
+    # 10-ticker watchlist. See planning/MARKET_SIMULATOR.md §6.
+    DEFAULT_EVENT_PROBABILITY = 1e-4
+
     def __init__(
         self,
         tickers: list[str],
         dt: float = DEFAULT_DT,
-        event_probability: float = 0.001,
+        event_probability: float = DEFAULT_EVENT_PROBABILITY,
+        seed_overrides: dict[str, float] | None = None,
     ) -> None:
         self._dt = dt
         self._event_prob = event_probability
+        self._seed_overrides = dict(seed_overrides or {})
 
         # Per-ticker state
         self._tickers: list[str] = []
         self._prices: dict[str, float] = {}
         self._params: dict[str, dict[str, float]] = {}
+        self._shocks: dict[str, _Shock] = {}
 
         # Cholesky decomposition of the correlation matrix (for correlated moves)
         self._cholesky: np.ndarray | None = None
@@ -96,24 +136,29 @@ def step(self) -> dict[str, float]:
             sigma = params["sigma"]
 
             # GBM: S(t+dt) = S(t) * exp((mu - 0.5*sigma^2)*dt + sigma*sqrt(dt)*Z)
+            # This is the underlying, correctly-calibrated price. Shocks (below)
+            # are a separate overlay and never touch this value directly.
             drift = (mu - 0.5 * sigma**2) * self._dt
             diffusion = sigma * math.sqrt(self._dt) * z_correlated[i]
             self._prices[ticker] *= math.exp(drift + diffusion)
 
-            # Random event: ~0.1% chance per tick per ticker
-            # With 10 tickers at 2 ticks/sec, expect an event ~every 50 seconds
+            # Random event: a transient, decaying overlay — not a permanent
+            # level shift. See _Shock docstring.
             if random.random() < self._event_prob:
-                shock_magnitude = random.uniform(0.02, 0.05)
-                shock_sign = random.choice([-1, 1])
-                self._prices[ticker] *= 1 + shock_magnitude * shock_sign
-                logger.debug(
-                    "Random event on %s: %.1f%% %s",
-                    ticker,
-                    shock_magnitude * 100,
-                    "up" if shock_sign > 0 else "down",
-                )
+                magnitude = random.uniform(0.015, 0.04) * random.choice([-1, 1])
+                self._shocks[ticker] = _Shock(magnitude=magnitude)
+                logger.debug("Shock event on %s: %.2f%%", ticker, magnitude * 100)
 
-            result[ticker] = round(self._prices[ticker], 2)
+            shock = self._shocks.get(ticker)
+            if shock is not None:
+                displayed = self._prices[ticker] * (1 + shock.magnitude)
+                shock.magnitude *= shock.decay
+                if abs(shock.magnitude) < 1e-4:
+                    del self._shocks[ticker]
+            else:
+                displayed = self._prices[ticker]
+
+            result[ticker] = round(displayed, 2)
 
         return result
 
@@ -131,6 +176,7 @@ def remove_ticker(self, ticker: str) -> None:
         self._tickers.remove(ticker)
         del self._prices[ticker]
         del self._params[ticker]
+        self._shocks.pop(ticker, None)
         self._rebuild_cholesky()
 
     def get_price(self, ticker: str) -> float | None:
@@ -141,6 +187,16 @@ def get_tickers(self) -> list[str]:
         """Return the list of currently tracked tickers."""
         return list(self._tickers)
 
+    def reset_price(self, ticker: str, price: float) -> None:
+        """Reset a ticker's price and clear any active shock, without touching correlation.
+
+        Used to rewind the simulator after running it forward to prefill history,
+        so live trading starts from the true seed rather than the prefill's endpoint.
+        """
+        if ticker in self._prices:
+            self._prices[ticker] = price
+            self._shocks.pop(ticker, None)
+
     # --- Internals ---
 
     def _add_ticker_internal(self, ticker: str) -> None:
@@ -148,13 +204,18 @@ def _add_ticker_internal(self, ticker: str) -> None:
         if ticker in self._prices:
             return
         self._tickers.append(ticker)
-        self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0))
+        seed = self._seed_overrides.get(ticker)
+        if seed is None:
+            seed = SEED_PRICES.get(ticker)
+        if seed is None:
+            seed = synthesize_seed(ticker)
+        self._prices[ticker] = seed
         self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS))
 
     def _rebuild_cholesky(self) -> None:
         """Rebuild the Cholesky decomposition of the ticker correlation matrix.
 
-        Called whenever tickers are added or removed. O(n^2) but n < 50.
+        Called whenever tickers are added or removed. O(n^2) but n < 100.
         """
         n = len(self._tickers)
         if n <= 1:
@@ -169,7 +230,19 @@ def _rebuild_cholesky(self) -> None:
                 corr[i, j] = rho
                 corr[j, i] = rho
 
-        self._cholesky = np.linalg.cholesky(corr)
+        try:
+            self._cholesky = np.linalg.cholesky(corr)
+        except np.linalg.LinAlgError:
+            # The shipped correlation structure is verified positive-definite at
+            # any size (block-equicorrelated, min eigenvalue = 1 - rho_max), but a
+            # future change to the structure could break that. Degrade to
+            # independent moves rather than 500ing a watchlist add.
+            logger.error(
+                "Correlation matrix not positive-definite for %d tickers; "
+                "falling back to uncorrelated moves",
+                n,
+            )
+            self._cholesky = np.eye(n)
 
     @staticmethod
     def _pairwise_correlation(t1: str, t2: str) -> float:
@@ -204,28 +277,47 @@ class SimulatorDataSource(MarketDataSource):
     `update_interval` seconds and writes results to the PriceCache.
     """
 
+    # 600 ticks x 500ms = 5 minutes of history per ticker. ~480KB for 50 tickers.
+    HISTORY_POINTS = 600
+
     def __init__(
         self,
         price_cache: PriceCache,
         update_interval: float = 0.5,
-        event_probability: float = 0.001,
+        event_probability: float = GBMSimulator.DEFAULT_EVENT_PROBABILITY,
+        seed_overrides: dict[str, float] | None = None,
+        status_detail: str | None = None,
     ) -> None:
         self._cache = price_cache
         self._interval = update_interval
         self._event_prob = event_probability
+        self._seed_overrides = dict(seed_overrides or {})
+        self._status_detail = status_detail
         self._sim: GBMSimulator | None = None
         self._task: asyncio.Task | None = None
+        self._history: dict[str, deque[tuple[float, float]]] = {}
+        self._open_prices: dict[str, float] = {}
 
     async def start(self, tickers: list[str]) -> None:
         self._sim = GBMSimulator(
             tickers=tickers,
             event_probability=self._event_prob,
+            seed_overrides=self._seed_overrides,
         )
+        for ticker in tickers:
+            price = self._sim.get_price(ticker)
+            if price is not None:
+                self._open_prices[ticker] = price
+
+        self._prefill_history(tickers)
+
         # Seed the cache with initial prices so SSE has data immediately
         for ticker in tickers:
             price = self._sim.get_price(ticker)
             if price is not None:
-                self._cache.update(ticker=ticker, price=price)
+                self._cache.update(
+                    ticker=ticker, price=price, open_price=self._open_prices.get(ticker, price)
+                )
         self._task = asyncio.create_task(self._run_loop(), name="simulator-loop")
         logger.info("Simulator started with %d tickers", len(tickers))
 
@@ -242,29 +334,86 @@ async def stop(self) -> None:
     async def add_ticker(self, ticker: str) -> None:
         if self._sim:
             self._sim.add_ticker(ticker)
-            # Seed cache immediately so the ticker has a price right away
             price = self._sim.get_price(ticker)
             if price is not None:
-                self._cache.update(ticker=ticker, price=price)
+                self._open_prices.setdefault(ticker, price)
+                self._history.setdefault(ticker, deque(maxlen=self.HISTORY_POINTS))
+                self._cache.update(
+                    ticker=ticker, price=price, open_price=self._open_prices[ticker]
+                )
             logger.info("Simulator: added ticker %s", ticker)
 
     async def remove_ticker(self, ticker: str) -> None:
         if self._sim:
             self._sim.remove_ticker(ticker)
         self._cache.remove(ticker)
+        self._history.pop(ticker, None)
+        self._open_prices.pop(ticker, None)
         logger.info("Simulator: removed ticker %s", ticker)
 
     def get_tickers(self) -> list[str]:
         return self._sim.get_tickers() if self._sim else []
 
+    def describe(self) -> SourceStatus:
+        return SourceStatus(
+            name="simulator",
+            live=False,
+            detail=self._status_detail or "synthetic GBM simulation",
+            tickers=len(self.get_tickers()),
+            cache_populated=len(self._cache) > 0,
+        )
+
+    def set_status_detail(self, detail: str | None) -> None:
+        """Allow a wrapping source (e.g. AnchoredSimulatorDataSource) to update the
+        health detail after re-anchoring, without exposing the private attribute."""
+        self._status_detail = detail
+
+    async def get_history(self, ticker: str, points: int = 120) -> list[PricePoint]:
+        series = self._history.get(ticker, ())
+        return [PricePoint(epoch_to_iso(t), p) for t, p in list(series)[-points:]]
+
     async def _run_loop(self) -> None:
         """Core loop: step the simulation, write to cache, sleep."""
         while True:
             try:
                 if self._sim:
                     prices = self._sim.step()
+                    now = time.time()
                     for ticker, price in prices.items():
-                        self._cache.update(ticker=ticker, price=price)
+                        self._cache.update(
+                            ticker=ticker,
+                            price=price,
+                            open_price=self._open_prices.get(ticker, price),
+                        )
+                        hist = self._history.setdefault(
+                            ticker, deque(maxlen=self.HISTORY_POINTS)
+                        )
+                        hist.append((now, price))
             except Exception:
-                logger.exception("Simulator step failed")
+                logger.exception("Simulator step failed")  # never let one bad tick kill the loop
             await asyncio.sleep(self._interval)
+
+    def _prefill_history(self, tickers: list[str]) -> None:
+        """Run the simulator forward HISTORY_POINTS steps with no sleeping, recording
+        the path, then reset every price to its seed. So the very first chart paint has
+        five minutes of plausible history instead of a single dot.
+        """
+        if self._sim is None or not tickers:
+            return
+
+        now = time.time()
+        paths: dict[str, deque[tuple[float, float]]] = {
+            t: deque(maxlen=self.HISTORY_POINTS) for t in tickers
+        }
+        for i in range(self.HISTORY_POINTS):
+            prices = self._sim.step()
+            ts = now - (self.HISTORY_POINTS - i) * self._interval
+            for ticker in tickers:
+                if ticker in prices:
+                    paths[ticker].append((ts, prices[ticker]))
+
+        for ticker in tickers:
+            self._history[ticker] = paths[ticker]
+            seed = self._open_prices.get(ticker)
+            if seed is not None:
+                self._sim.reset_price(ticker, seed)
diff --git a/backend/app/market/stream.py b/backend/app/market/stream.py
index 7fd974b7c..14b96c3e2 100644
--- a/backend/app/market/stream.py
+++ b/backend/app/market/stream.py
@@ -5,6 +5,7 @@
 import asyncio
 import json
 import logging
+import time
 from collections.abc import AsyncGenerator
 
 from fastapi import APIRouter, Request
@@ -16,6 +17,12 @@
 
 router = APIRouter(prefix="/api/stream", tags=["streaming"])
 
+# How long the stream may stay silent before sending an SSE comment ping.
+# Matters most in Massive mode: a 15s poll interval (free tier) or a closed
+# market can otherwise leave the connection silent long enough for a proxy or
+# a laptop sleep to drop it without either side noticing.
+HEARTBEAT_INTERVAL_SECONDS = 10.0
+
 
 def create_stream_router(price_cache: PriceCache) -> APIRouter:
     """Create the SSE streaming router with a reference to the price cache.
@@ -52,16 +59,22 @@ async def _generate_events(
     price_cache: PriceCache,
     request: Request,
     interval: float = 0.5,
+    heartbeat_interval: float = HEARTBEAT_INTERVAL_SECONDS,
 ) -> 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 the full price map whenever the cache version changes (deliberately
+    not a delta protocol — at 10-50 tickers the payload is a couple of KB, and
+    a delta would need reconnection-resync logic for no measurable gain).
+    Sends a ": ping" comment if nothing has changed for `heartbeat_interval`
+    seconds, so proxies and idle connections don't silently die. Stops when
+    the client disconnects (detected via request.is_disconnected()).
     """
     # 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 +94,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 > heartbeat_interval:
+                yield ": ping\n\n"  # SSE comment — ignored by EventSource, keeps the pipe alive
+                last_sent = time.monotonic()
 
             await asyncio.sleep(interval)
     except asyncio.CancelledError:
diff --git a/backend/market_data_demo.py b/backend/market_data_demo.py
index 7414416c4..5a617311d 100644
--- a/backend/market_data_demo.py
+++ b/backend/market_data_demo.py
@@ -79,10 +79,10 @@ def build_table(
             continue
 
         # Direction styling
-        if update.direction == "up":
+        if update.tick_direction == "up":
             color = "green"
             arrow = "[bold green]\u25b2[/]"
-        elif update.direction == "down":
+        elif update.tick_direction == "down":
             color = "red"
             arrow = "[bold red]\u25bc[/]"
         else:
@@ -248,8 +248,8 @@ async def run() -> None:
 
                     # Log notable moves
                     if abs(update.change_percent) > 1.0:
-                        direction = "\u25b2" if update.direction == "up" else "\u25bc"
-                        color = "green" if update.direction == "up" else "red"
+                        direction = "\u25b2" if update.tick_direction == "up" else "\u25bc"
+                        color = "green" if update.tick_direction == "up" else "red"
                         timestamp = time.strftime("%H:%M:%S")
                         events.appendleft(
                             f"[bright_black]{timestamp}[/]  "
diff --git a/backend/tests/market/test_anchored.py b/backend/tests/market/test_anchored.py
new file mode 100644
index 000000000..e7b1160d3
--- /dev/null
+++ b/backend/tests/market/test_anchored.py
@@ -0,0 +1,235 @@
+"""Tests for AnchoredSimulatorDataSource. No test hits the network."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.market.anchored import AnchoredSimulatorDataSource
+from app.market.cache import PriceCache
+from app.market.models import SourceStatus
+
+
+def _bar(ticker: str, close: float) -> MagicMock:
+    bar = MagicMock()
+    bar.ticker = ticker
+    bar.close = close
+    return bar
+
+
+@pytest.mark.asyncio
+class TestAnchoredSimulatorDataSource:
+    async def test_start_anchors_to_grouped_daily_close(self):
+        """A single get_grouped_daily_aggs call should price the whole watchlist."""
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.return_value = [
+            _bar("AAPL", 324.96),
+            _bar("GOOGL", 337.12),
+            _bar("OTHERTICKER", 12.34),
+        ]
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0
+            )
+            await source.start(["AAPL", "GOOGL"])
+
+        assert source._anchors == {"AAPL": 324.96, "GOOGL": 337.12}
+        assert mock_client.get_grouped_daily_aggs.call_count == 1
+
+        await source.stop()
+
+    async def test_anchored_prices_seed_the_cache(self):
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)]
+        cache = PriceCache()
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=cache, reanchor_interval=0
+            )
+            await source.start(["AAPL"])
+
+        update = cache.get("AAPL")
+        assert update.price == 324.96
+        assert update.open_price == 324.96
+
+        await source.stop()
+
+    async def test_fetch_anchors_walks_back_over_a_weekend(self):
+        """Saturday and Sunday return empty results; Friday's close should be used."""
+        mock_client = MagicMock()
+        # today() - 1 day = Saturday (empty), -2 = Friday (populated)
+        mock_client.get_grouped_daily_aggs.side_effect = [
+            [],  # Saturday
+            [_bar("AAPL", 324.96)],  # Friday
+        ]
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache()
+            )
+            anchors, anchor_date = source._fetch_anchors(["AAPL"])
+
+        assert anchors == {"AAPL": 324.96}
+        assert anchor_date is not None
+        assert mock_client.get_grouped_daily_aggs.call_count == 2
+
+    async def test_fetch_anchors_falls_back_after_repeated_failures(self):
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.side_effect = Exception("network error")
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache()
+            )
+            anchors, anchor_date = source._fetch_anchors(["AAPL"])
+
+        assert anchors == {}
+        assert anchor_date is None
+        assert mock_client.get_grouped_daily_aggs.call_count == 7  # MAX_ANCHOR_LOOKBACK_DAYS
+
+    async def test_start_falls_back_to_static_seeds_when_anchor_fetch_fails(self):
+        """If Massive is unreachable at startup, the simulator must still start
+        (with the static seed table) rather than dead-ending the app."""
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.side_effect = Exception("network error")
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0
+            )
+            await source.start(["AAPL"])
+
+        assert "AAPL" in source.get_tickers()
+        assert source.describe().detail == "simulated from static seed prices (anchor fetch failed)"
+
+        await source.stop()
+
+    async def test_describe_reports_anchor_date_and_count(self):
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)]
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0
+            )
+            await source.start(["AAPL"])
+
+        status = source.describe()
+        assert isinstance(status, SourceStatus)
+        assert status.name == "anchored-simulator"
+        assert status.live is False  # simulated, never presented as live
+        assert "1 anchored" in status.detail
+
+        await source.stop()
+
+    async def test_add_ticker_delegates_to_underlying_simulator(self):
+        """Adding an unanchored symbol must never dead-end — the underlying
+        simulator synthesizes a seed for it."""
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)]
+        cache = PriceCache()
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=cache, reanchor_interval=0
+            )
+            await source.start(["AAPL"])
+            await source.add_ticker("ZZZZ")
+
+        assert "ZZZZ" in source.get_tickers()
+        assert cache.get("ZZZZ") is not None
+
+        await source.stop()
+
+    async def test_remove_ticker_delegates_to_underlying_simulator(self):
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)]
+        cache = PriceCache()
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=cache, reanchor_interval=0
+            )
+            await source.start(["AAPL"])
+            await source.remove_ticker("AAPL")
+
+        assert "AAPL" not in source.get_tickers()
+        assert cache.get("AAPL") is None
+
+        await source.stop()
+
+    async def test_get_history_uses_real_minute_bars(self):
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)]
+        minute_bar = MagicMock(timestamp=1757000000000, close=325.10)
+        mock_client.get_aggs.return_value = [minute_bar]
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0
+            )
+            await source.start(["AAPL"])
+            history = await source.get_history("AAPL")
+
+        assert len(history) == 1
+        assert history[0].price == 325.10
+
+        await source.stop()
+
+    async def test_get_history_empty_when_no_anchor_date(self):
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.side_effect = Exception("network error")
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0
+            )
+            await source.start(["AAPL"])
+            history = await source.get_history("AAPL")
+
+        assert history == []
+
+        await source.stop()
+
+    async def test_stop_is_idempotent(self):
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)]
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0
+            )
+            await source.start(["AAPL"])
+            await source.stop()
+            await source.stop()  # should not raise
+
+    async def test_reanchor_task_started_when_interval_positive(self):
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)]
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache(), reanchor_interval=3600.0
+            )
+            await source.start(["AAPL"])
+
+        assert source._reanchor_task is not None
+        assert not source._reanchor_task.done()
+
+        await source.stop()
+        assert source._reanchor_task is None
+
+    async def test_no_reanchor_task_when_interval_zero(self):
+        mock_client = MagicMock()
+        mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)]
+
+        with patch("app.market.anchored.RESTClient", return_value=mock_client):
+            source = AnchoredSimulatorDataSource(
+                api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0
+            )
+            await source.start(["AAPL"])
+
+        assert source._reanchor_task is None
+
+        await source.stop()
diff --git a/backend/tests/market/test_cache.py b/backend/tests/market/test_cache.py
index b5ab3d55d..81b537a91 100644
--- a/backend/tests/market/test_cache.py
+++ b/backend/tests/market/test_cache.py
@@ -15,26 +15,32 @@ def test_update_and_get(self):
         assert cache.get("AAPL") == update
 
     def test_first_update_is_flat(self):
-        """Test that the first update has flat direction."""
+        """Test that the first update has flat tick_direction."""
         cache = PriceCache()
         update = cache.update("AAPL", 190.50)
-        assert update.direction == "flat"
+        assert update.tick_direction == "flat"
         assert update.previous_price == 190.50
 
+    def test_first_update_open_price_defaults_to_price(self):
+        """On first write, open_price defaults to price when not given."""
+        cache = PriceCache()
+        update = cache.update("AAPL", 190.50)
+        assert update.open_price == 190.50
+
     def test_direction_up(self):
-        """Test price update with upward direction."""
+        """Test price update with upward tick_direction."""
         cache = PriceCache()
         cache.update("AAPL", 190.00)
         update = cache.update("AAPL", 191.00)
-        assert update.direction == "up"
+        assert update.tick_direction == "up"
         assert update.change == 1.00
 
     def test_direction_down(self):
-        """Test price update with downward direction."""
+        """Test price update with downward tick_direction."""
         cache = PriceCache()
         cache.update("AAPL", 190.00)
         update = cache.update("AAPL", 189.00)
-        assert update.direction == "down"
+        assert update.tick_direction == "down"
         assert update.change == -1.00
 
     def test_remove(self):
@@ -101,3 +107,29 @@ def test_price_rounding(self):
         cache = PriceCache()
         update = cache.update("AAPL", 190.12345)
         assert update.price == 190.12
+
+    def test_explicit_open_price(self):
+        """Test that an explicit open_price is stored as given."""
+        cache = PriceCache()
+        update = cache.update("AAPL", 190.50, open_price=180.00)
+        assert update.open_price == 180.00
+
+    def test_open_price_persists_across_updates(self):
+        """open_price should stay fixed for the session unless explicitly changed."""
+        cache = PriceCache()
+        cache.update("AAPL", 190.00, open_price=180.00)
+        update = cache.update("AAPL", 195.00)  # no open_price given
+        assert update.open_price == 180.00
+
+    def test_open_price_can_be_updated_explicitly(self):
+        """A source may explicitly move the baseline (e.g. Massive's prev_day.close)."""
+        cache = PriceCache()
+        cache.update("AAPL", 190.00, open_price=180.00)
+        update = cache.update("AAPL", 195.00, open_price=185.00)
+        assert update.open_price == 185.00
+
+    def test_open_price_rounded(self):
+        """Test that open_price is rounded to 2 decimal places."""
+        cache = PriceCache()
+        update = cache.update("AAPL", 190.00, open_price=180.126)
+        assert update.open_price == 180.13
diff --git a/backend/tests/market/test_capabilities.py b/backend/tests/market/test_capabilities.py
new file mode 100644
index 000000000..de411c4f4
--- /dev/null
+++ b/backend/tests/market/test_capabilities.py
@@ -0,0 +1,111 @@
+"""Tests for probe_capabilities — the Massive entitlement probe.
+
+No test hits the network: the RESTClient constructor and its methods are
+always mocked.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import urllib3.exceptions
+from massive.exceptions import AuthError, BadResponse
+
+from app.market.capabilities import probe_capabilities
+
+
+class TestProbeCapabilities:
+    def test_no_key_is_invalid(self):
+        with patch("app.market.capabilities.RESTClient", side_effect=AuthError("no key")):
+            caps = probe_capabilities("")
+
+        assert caps.valid is False
+        assert caps.realtime is False
+        assert caps.end_of_day is False
+
+    def test_realtime_entitled_key(self):
+        """Advanced tier: snapshot call succeeds outright."""
+        mock_client = MagicMock()
+        mock_client.get_snapshot_all.return_value = []
+
+        with patch("app.market.capabilities.RESTClient", return_value=mock_client):
+            caps = probe_capabilities("advanced-key")
+
+        assert caps.valid is True
+        assert caps.realtime is True
+        assert caps.end_of_day is True
+
+    def test_end_of_day_only_key(self):
+        """Basic/free tier: snapshot NOT_AUTHORIZED, but aggregate call succeeds.
+        This is the tier nearly every student will have — the whole point of the
+        capability probe is to detect it accurately."""
+        mock_client = MagicMock()
+        mock_client.get_snapshot_all.side_effect = BadResponse(
+            "NOT_AUTHORIZED: Please upgrade your plan"
+        )
+        mock_client.get_previous_close_agg.return_value = [MagicMock(close=324.96)]
+
+        with patch("app.market.capabilities.RESTClient", return_value=mock_client):
+            caps = probe_capabilities("basic-key")
+
+        assert caps.valid is True
+        assert caps.realtime is False
+        assert caps.end_of_day is True
+        assert "end-of-day" in caps.detail.lower()
+
+    def test_invalid_key_rejected_by_both_calls(self):
+        mock_client = MagicMock()
+        mock_client.get_snapshot_all.side_effect = BadResponse("NOT_AUTHORIZED")
+        mock_client.get_previous_close_agg.side_effect = AuthError("invalid key")
+
+        with patch("app.market.capabilities.RESTClient", return_value=mock_client):
+            caps = probe_capabilities("bad-key")
+
+        assert caps.valid is False
+        assert caps.realtime is False
+        assert caps.end_of_day is False
+        assert "rejected" in caps.detail.lower()
+
+    def test_snapshot_error_not_not_authorized_is_treated_as_unexpected(self):
+        """A BadResponse that isn't NOT_AUTHORIZED (e.g. a genuine server error)
+        should not be silently reinterpreted as an entitlement gap."""
+        mock_client = MagicMock()
+        mock_client.get_snapshot_all.side_effect = BadResponse("INTERNAL_SERVER_ERROR")
+
+        with patch("app.market.capabilities.RESTClient", return_value=mock_client):
+            caps = probe_capabilities("some-key")
+
+        assert caps.valid is False
+        assert caps.realtime is False
+        assert caps.end_of_day is False
+
+    def test_rate_limited_on_first_call(self):
+        """MaxRetryError (rate limiting), not BadResponse — planning/MASSIVE_API.md §7."""
+        mock_client = MagicMock()
+        error = urllib3.exceptions.MaxRetryError(pool=MagicMock(), url="/x", reason="429")
+        mock_client.get_snapshot_all.side_effect = error
+
+        with patch("app.market.capabilities.RESTClient", return_value=mock_client):
+            caps = probe_capabilities("some-key")
+
+        assert caps.valid is False
+        assert "rate" in caps.detail.lower() or "unreachable" in caps.detail.lower()
+
+    def test_never_raises(self):
+        """probe_capabilities must never raise — the factory awaits it directly
+        with no try/except of its own."""
+        mock_client = MagicMock()
+        mock_client.get_snapshot_all.side_effect = RuntimeError("something exploded")
+
+        with patch("app.market.capabilities.RESTClient", return_value=mock_client):
+            caps = probe_capabilities("some-key")  # must not raise
+
+        assert caps.valid is False
+
+    def test_probe_uses_no_retries(self):
+        mock_client = MagicMock()
+        mock_client.get_snapshot_all.return_value = []
+
+        with patch("app.market.capabilities.RESTClient", return_value=mock_client) as mock_cls:
+            probe_capabilities("some-key")
+
+        _, kwargs = mock_cls.call_args
+        assert kwargs.get("retries") == 0
diff --git a/backend/tests/market/test_factory.py b/backend/tests/market/test_factory.py
index 5ff5dd49e..e04d93039 100644
--- a/backend/tests/market/test_factory.py
+++ b/backend/tests/market/test_factory.py
@@ -1,79 +1,131 @@
-"""Tests for market data source factory."""
+"""Tests for the market data source factory."""
 
 import os
 from unittest.mock import patch
 
+import pytest
+
+from app.market.anchored import AnchoredSimulatorDataSource
 from app.market.cache import PriceCache
+from app.market.capabilities import MassiveCapabilities
 from app.market.factory import create_market_data_source
 from app.market.massive_client import MassiveDataSource
 from app.market.simulator import SimulatorDataSource
 
 
+@pytest.mark.asyncio
 class TestFactory:
-    """Tests for create_market_data_source factory."""
+    """Tests for create_market_data_source. Never hits the network — probe_capabilities
+    is always mocked."""
 
-    def test_creates_simulator_when_no_api_key(self):
-        """Test that simulator is created when MASSIVE_API_KEY is not set."""
+    async def test_creates_simulator_when_no_api_key(self):
+        """No key → simulator, and the (network-calling) probe must not even run."""
         cache = PriceCache()
 
         with patch.dict(os.environ, {}, clear=True):
-            source = create_market_data_source(cache)
+            with patch("app.market.factory.probe_capabilities") as mock_probe:
+                source = await create_market_data_source(cache)
 
         assert isinstance(source, SimulatorDataSource)
+        mock_probe.assert_not_called()
 
-    def test_creates_simulator_when_api_key_empty(self):
-        """Test that simulator is created when MASSIVE_API_KEY is empty."""
+    async def test_creates_simulator_when_api_key_empty(self):
         cache = PriceCache()
 
         with patch.dict(os.environ, {"MASSIVE_API_KEY": ""}, clear=True):
-            source = create_market_data_source(cache)
+            with patch("app.market.factory.probe_capabilities"):
+                source = await create_market_data_source(cache)
 
         assert isinstance(source, SimulatorDataSource)
 
-    def test_creates_simulator_when_api_key_whitespace(self):
-        """Test that simulator is created when MASSIVE_API_KEY is whitespace."""
+    async def test_creates_simulator_when_api_key_whitespace(self):
         cache = PriceCache()
 
         with patch.dict(os.environ, {"MASSIVE_API_KEY": "   "}, clear=True):
-            source = create_market_data_source(cache)
+            with patch("app.market.factory.probe_capabilities"):
+                source = await create_market_data_source(cache)
 
         assert isinstance(source, SimulatorDataSource)
 
-    def test_creates_massive_when_api_key_set(self):
-        """Test that Massive client is created when MASSIVE_API_KEY is set."""
+    async def test_creates_massive_when_realtime_entitled(self):
         cache = PriceCache()
+        caps = MassiveCapabilities(True, True, True, "real-time snapshots entitled")
 
         with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True):
-            source = create_market_data_source(cache)
+            with patch("app.market.factory.probe_capabilities", return_value=caps):
+                source = await create_market_data_source(cache)
 
         assert isinstance(source, MassiveDataSource)
+        assert source._api_key == "test-key"
 
-    def test_massive_receives_api_key(self):
-        """Test that Massive client receives the API key."""
+    async def test_creates_anchored_simulator_when_end_of_day_only(self):
+        """The free-tier case: a valid key that cannot return live prices must
+        route to the anchored simulator, not a MassiveDataSource that would
+        poll forever and never populate the cache."""
         cache = PriceCache()
+        caps = MassiveCapabilities(True, False, True, "end-of-day only (Basic tier)")
 
-        with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key-123"}, clear=True):
-            source = create_market_data_source(cache)
+        with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True):
+            with patch("app.market.factory.probe_capabilities", return_value=caps):
+                with patch("app.market.anchored.RESTClient"):
+                    source = await create_market_data_source(cache)
 
-        assert isinstance(source, MassiveDataSource)
-        assert source._api_key == "test-key-123"
+        assert isinstance(source, AnchoredSimulatorDataSource)
+
+    async def test_falls_back_to_simulator_when_key_rejected(self):
+        """An invalid/revoked key must never boot into a broken state — it
+        degrades to the simulator instead of a source that produces nothing."""
+        cache = PriceCache()
+        caps = MassiveCapabilities(False, False, False, "key rejected: 401")
 
-    def test_simulator_receives_cache(self):
-        """Test that simulator receives the cache reference."""
+        with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True):
+            with patch("app.market.factory.probe_capabilities", return_value=caps):
+                source = await create_market_data_source(cache)
+
+        assert isinstance(source, SimulatorDataSource)
+        assert "key rejected" in source.describe().detail
+
+    async def test_probe_called_with_the_configured_key(self):
+        cache = PriceCache()
+        caps = MassiveCapabilities(True, True, True, "real-time snapshots entitled")
+
+        with patch.dict(os.environ, {"MASSIVE_API_KEY": "specific-key-123"}, clear=True):
+            with patch(
+                "app.market.factory.probe_capabilities", return_value=caps
+            ) as mock_probe:
+                await create_market_data_source(cache)
+
+        mock_probe.assert_called_once_with("specific-key-123")
+
+    async def test_simulator_receives_cache(self):
         cache = PriceCache()
 
         with patch.dict(os.environ, {}, clear=True):
-            source = create_market_data_source(cache)
+            source = await create_market_data_source(cache)
 
         assert isinstance(source, SimulatorDataSource)
         assert source._cache is cache
 
-    def test_massive_receives_cache(self):
-        """Test that Massive client receives the cache reference."""
+    async def test_massive_receives_cache(self):
         cache = PriceCache()
+        caps = MassiveCapabilities(True, True, True, "real-time snapshots entitled")
 
         with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True):
-            source = create_market_data_source(cache)
+            with patch("app.market.factory.probe_capabilities", return_value=caps):
+                source = await create_market_data_source(cache)
 
         assert isinstance(source, MassiveDataSource)
         assert source._cache is cache
+
+    async def test_never_raises_on_probe_exception_paths(self):
+        """probe_capabilities itself never raises (it catches everything), but
+        the factory must still resolve to a usable source for every capability
+        combination it can return."""
+        cache = PriceCache()
+        caps = MassiveCapabilities(False, False, False, "unreachable/rate-limited: timeout")
+
+        with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True):
+            with patch("app.market.factory.probe_capabilities", return_value=caps):
+                source = await create_market_data_source(cache)
+
+        assert isinstance(source, SimulatorDataSource)
diff --git a/backend/tests/market/test_interface.py b/backend/tests/market/test_interface.py
new file mode 100644
index 000000000..cbd698b12
--- /dev/null
+++ b/backend/tests/market/test_interface.py
@@ -0,0 +1,60 @@
+"""Tests for the MarketDataSource abstract contract."""
+
+import pytest
+
+from app.market.interface import MarketDataSource
+
+
+class _MinimalSource(MarketDataSource):
+    """Implements only the abstract members, to exercise the default get_history()."""
+
+    async def start(self, tickers):
+        pass
+
+    async def stop(self):
+        pass
+
+    async def add_ticker(self, ticker):
+        pass
+
+    async def remove_ticker(self, ticker):
+        pass
+
+    def get_tickers(self):
+        return []
+
+    def describe(self):
+        from app.market.models import SourceStatus
+
+        return SourceStatus(name="minimal", live=False, detail="", tickers=0, cache_populated=False)
+
+
+class TestMarketDataSource:
+    def test_cannot_instantiate_directly(self):
+        with pytest.raises(TypeError):
+            MarketDataSource()
+
+    def test_subclass_missing_describe_cannot_instantiate(self):
+        class _Incomplete(MarketDataSource):
+            async def start(self, tickers):
+                pass
+
+            async def stop(self):
+                pass
+
+            async def add_ticker(self, ticker):
+                pass
+
+            async def remove_ticker(self, ticker):
+                pass
+
+            def get_tickers(self):
+                return []
+
+        with pytest.raises(TypeError):
+            _Incomplete()
+
+    @pytest.mark.asyncio
+    async def test_default_get_history_returns_empty_list(self):
+        source = _MinimalSource()
+        assert await source.get_history("AAPL") == []
diff --git a/backend/tests/market/test_massive.py b/backend/tests/market/test_massive.py
index cdd7dbd24..2e7b23f17 100644
--- a/backend/tests/market/test_massive.py
+++ b/backend/tests/market/test_massive.py
@@ -1,20 +1,30 @@
-"""Tests for MassiveDataSource (mocked)."""
+"""Tests for MassiveDataSource (mocked). No test hits the network."""
 
 from unittest.mock import MagicMock, patch
 
 import pytest
+import urllib3.exceptions
+from massive.exceptions import AuthError, BadResponse
 
 from app.market.cache import PriceCache
 from app.market.massive_client import MassiveDataSource
+from app.market.models import SourceStatus
 
 
-def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock:
-    """Create a mock Massive snapshot object."""
+def _make_snapshot(
+    ticker: str, price: float, sip_timestamp_ns: int, prev_close: float | None = None
+) -> MagicMock:
+    """Create a mock Massive snapshot object matching the real SDK's field names."""
     snap = MagicMock()
     snap.ticker = ticker
     snap.last_trade = MagicMock()
     snap.last_trade.price = price
-    snap.last_trade.timestamp = timestamp_ms
+    snap.last_trade.sip_timestamp = sip_timestamp_ns
+    if prev_close is not None:
+        snap.prev_day = MagicMock()
+        snap.prev_day.close = prev_close
+    else:
+        snap.prev_day = None
     return snap
 
 
@@ -34,8 +44,8 @@ async def test_poll_updates_cache(self):
         source._client = MagicMock()  # Satisfy the _poll_once guard
 
         mock_snapshots = [
-            _make_snapshot("AAPL", 190.50, 1707580800000),
-            _make_snapshot("GOOGL", 175.25, 1707580800000),
+            _make_snapshot("AAPL", 190.50, 1707580800_000000000),
+            _make_snapshot("GOOGL", 175.25, 1707580800_000000000),
         ]
 
         with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):
@@ -44,6 +54,57 @@ async def test_poll_updates_cache(self):
         assert cache.get_price("AAPL") == 190.50
         assert cache.get_price("GOOGL") == 175.25
 
+    async def test_nanosecond_timestamp_conversion(self):
+        """Massive snapshot timestamps are sip_timestamp in NANOSECONDS, not
+        `timestamp` in milliseconds. Regression guard for planning/MASSIVE_API.md
+        §6: a naive /1000.0 on the wrong field is wrong by a factor of a million."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()
+
+        mock_snapshots = [_make_snapshot("AAPL", 190.50, 1605192894630916600)]
+
+        with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):
+            await source._poll_once()
+
+        update = cache.get("AAPL")
+        assert update.timestamp == pytest.approx(1605192894.63, abs=0.01)
+
+    async def test_open_price_captured_from_prev_day_close(self):
+        """Regression guard: the daily change column has nowhere to get its
+        baseline from unless prev_day.close is captured as open_price."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()
+
+        mock_snapshots = [
+            _make_snapshot("AAPL", 325.41, 1707580800_000000000, prev_close=324.96)
+        ]
+
+        with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):
+            await source._poll_once()
+
+        update = cache.get("AAPL")
+        assert update.open_price == 324.96
+
+    async def test_missing_prev_day_leaves_open_price_defaulted(self):
+        """When prev_day is unavailable, open_price should fall back to the
+        cache's own default (price on first write) rather than raising."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()
+
+        mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800_000000000)]
+
+        with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):
+            await source._poll_once()
+
+        update = cache.get("AAPL")
+        assert update.open_price == 190.50
+
     async def test_malformed_snapshot_skipped(self):
         """Test that malformed snapshots are skipped gracefully."""
         cache = PriceCache()
@@ -55,7 +116,7 @@ async def test_malformed_snapshot_skipped(self):
         source._tickers = ["AAPL", "BAD"]
         source._client = MagicMock()  # Satisfy the _poll_once guard
 
-        good_snap = _make_snapshot("AAPL", 190.50, 1707580800000)
+        good_snap = _make_snapshot("AAPL", 190.50, 1707580800_000000000)
         bad_snap = MagicMock()
         bad_snap.ticker = "BAD"
         bad_snap.last_trade = None  # Will cause AttributeError
@@ -67,8 +128,51 @@ async def test_malformed_snapshot_skipped(self):
         assert cache.get_price("AAPL") == 190.50
         assert cache.get_price("BAD") is None
 
+    async def test_auth_error_marks_not_live(self):
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="bad-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()
+
+        with patch.object(source, "_fetch_snapshots", side_effect=AuthError("invalid key")):
+            await source._poll_once()
+
+        assert source.describe().live is False
+        assert "auth" in source.describe().detail.lower()
+
+    async def test_rate_limit_error_classified_distinctly(self):
+        """Rate limiting arrives as urllib3.MaxRetryError, not BadResponse —
+        planning/MASSIVE_API.md §7. A handler written as `except BadResponse`
+        alone would miss it entirely."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()
+
+        error = urllib3.exceptions.MaxRetryError(pool=MagicMock(), url="/x", reason="429")
+        with patch.object(source, "_fetch_snapshots", side_effect=error):
+            await source._poll_once()  # must not raise
+
+        status = source.describe()
+        assert status.live is False
+        assert "rate limit" in status.detail.lower() or "unreachable" in status.detail.lower()
+
+    async def test_bad_response_error_does_not_crash(self):
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()
+
+        with patch.object(
+            source, "_fetch_snapshots", side_effect=BadResponse("NOT_AUTHORIZED")
+        ):
+            await source._poll_once()  # Should not raise
+
+        assert cache.get_price("AAPL") is None
+        assert source.describe().live is False
+
     async def test_api_error_does_not_crash(self):
-        """Test that API errors don't crash the poller."""
+        """Test that unexpected errors don't crash the poller."""
         cache = PriceCache()
         source = MassiveDataSource(
             api_key="test-key",
@@ -83,25 +187,33 @@ async def test_api_error_does_not_crash(self):
 
         assert cache.get_price("AAPL") is None  # No update happened
 
-    async def test_timestamp_conversion(self):
-        """Test that timestamps are converted from milliseconds to seconds."""
+    async def test_successful_poll_marks_live(self):
         cache = PriceCache()
-        source = MassiveDataSource(
-            api_key="test-key",
-            price_cache=cache,
-            poll_interval=60.0,
-        )
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
         source._tickers = ["AAPL"]
-        source._client = MagicMock()  # Satisfy the _poll_once guard
+        source._client = MagicMock()
 
-        mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)]
+        with patch.object(
+            source, "_fetch_snapshots", return_value=[_make_snapshot("AAPL", 190.50, 1)]
+        ):
+            await source._poll_once()
 
-        with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):
+        assert source.describe().live is True
+
+    async def test_unknown_tickers_tracked(self):
+        """Tickers requested but absent from the response should be surfaced,
+        not silently dropped — planning/MARKET_INTERFACE.md §7."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL", "ZZZZ"]
+        source._client = MagicMock()
+
+        with patch.object(
+            source, "_fetch_snapshots", return_value=[_make_snapshot("AAPL", 190.50, 1)]
+        ):
             await source._poll_once()
 
-        update = cache.get("AAPL")
-        assert update is not None
-        assert update.timestamp == 1707580800.0  # Converted to seconds
+        assert "ZZZZ" in source.describe().detail
 
     async def test_add_ticker(self):
         """Test adding a ticker."""
@@ -189,7 +301,7 @@ async def test_start_immediate_poll(self):
         cache = PriceCache()
         source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
 
-        mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)]
+        mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800_000000000)]
 
         with patch("app.market.massive_client.RESTClient"):
             with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):
@@ -199,3 +311,27 @@ async def test_start_immediate_poll(self):
         assert cache.get_price("AAPL") == 190.50
 
         await source.stop()
+
+    async def test_start_constructs_client_with_no_retries(self):
+        """retries=0 is deliberate: the SDK's default backoff is inside the
+        same rate-limit window, so a retry on 429 is guaranteed to fail too —
+        it only burns budget. planning/MASSIVE_API.md §7."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+
+        with patch("app.market.massive_client.RESTClient") as mock_client_cls:
+            with patch.object(source, "_fetch_snapshots", return_value=[]):
+                await source.start(["AAPL"])
+
+        _, kwargs = mock_client_cls.call_args
+        assert kwargs.get("retries") == 0
+
+        await source.stop()
+
+    async def test_describe_before_start(self):
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache)
+        status = source.describe()
+        assert isinstance(status, SourceStatus)
+        assert status.name == "massive"
+        assert status.live is False
diff --git a/backend/tests/market/test_models.py b/backend/tests/market/test_models.py
index 21600dfd6..dafa749af 100644
--- a/backend/tests/market/test_models.py
+++ b/backend/tests/market/test_models.py
@@ -1,8 +1,8 @@
-"""Tests for PriceUpdate dataclass."""
+"""Tests for market data models: PriceUpdate, PricePoint, SourceStatus."""
 
 import pytest
 
-from app.market.models import PriceUpdate
+from app.market.models import PricePoint, PriceUpdate, SourceStatus, epoch_to_iso
 
 
 class TestPriceUpdate:
@@ -10,68 +10,207 @@ class TestPriceUpdate:
 
     def test_price_update_creation(self):
         """Test basic PriceUpdate creation."""
-        update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0)
+        update = PriceUpdate(
+            ticker="AAPL",
+            price=190.50,
+            previous_price=190.00,
+            open_price=189.00,
+            timestamp=1234567890.0,
+        )
         assert update.ticker == "AAPL"
         assert update.price == 190.50
         assert update.previous_price == 190.00
+        assert update.open_price == 189.00
         assert update.timestamp == 1234567890.0
 
     def test_change_calculation(self):
-        """Test price change calculation."""
-        update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0)
+        """Test tick-to-tick price change calculation."""
+        update = PriceUpdate(
+            ticker="AAPL", price=190.50, previous_price=190.00, open_price=190.00
+        )
         assert update.change == 0.50
 
     def test_change_negative(self):
-        """Test negative price change."""
-        update = PriceUpdate(ticker="AAPL", price=189.50, previous_price=190.00, timestamp=1234567890.0)
+        """Test negative tick-to-tick price change."""
+        update = PriceUpdate(
+            ticker="AAPL", price=189.50, previous_price=190.00, open_price=190.00
+        )
         assert update.change == -0.50
 
     def test_change_percent_up(self):
-        """Test percentage change calculation (up)."""
-        update = PriceUpdate(ticker="AAPL", price=190.00, previous_price=100.00, timestamp=1234567890.0)
+        """Test tick-to-tick percentage change calculation (up)."""
+        update = PriceUpdate(
+            ticker="AAPL", price=190.00, previous_price=100.00, open_price=100.00
+        )
         assert update.change_percent == 90.0
 
     def test_change_percent_down(self):
-        """Test percentage change calculation (down)."""
-        update = PriceUpdate(ticker="AAPL", price=100.00, previous_price=200.00, timestamp=1234567890.0)
+        """Test tick-to-tick percentage change calculation (down)."""
+        update = PriceUpdate(
+            ticker="AAPL", price=100.00, previous_price=200.00, open_price=200.00
+        )
         assert update.change_percent == -50.0
 
     def test_change_percent_zero_previous(self):
         """Test percentage change with zero previous price."""
-        update = PriceUpdate(ticker="AAPL", price=100.00, previous_price=0.00, timestamp=1234567890.0)
+        update = PriceUpdate(
+            ticker="AAPL", price=100.00, previous_price=0.00, open_price=100.00
+        )
         assert update.change_percent == 0.0
 
-    def test_direction_up(self):
-        """Test direction calculation (up)."""
-        update = PriceUpdate(ticker="AAPL", price=191.00, previous_price=190.00, timestamp=1234567890.0)
-        assert update.direction == "up"
-
-    def test_direction_down(self):
-        """Test direction calculation (down)."""
-        update = PriceUpdate(ticker="AAPL", price=189.00, previous_price=190.00, timestamp=1234567890.0)
-        assert update.direction == "down"
-
-    def test_direction_flat(self):
-        """Test direction calculation (flat)."""
-        update = PriceUpdate(ticker="AAPL", price=190.00, previous_price=190.00, timestamp=1234567890.0)
-        assert update.direction == "flat"
+    def test_tick_direction_up(self):
+        """Test tick_direction calculation (up)."""
+        update = PriceUpdate(
+            ticker="AAPL", price=191.00, previous_price=190.00, open_price=190.00
+        )
+        assert update.tick_direction == "up"
+
+    def test_tick_direction_down(self):
+        """Test tick_direction calculation (down)."""
+        update = PriceUpdate(
+            ticker="AAPL", price=189.00, previous_price=190.00, open_price=190.00
+        )
+        assert update.tick_direction == "down"
+
+    def test_tick_direction_flat(self):
+        """Test tick_direction calculation (flat)."""
+        update = PriceUpdate(
+            ticker="AAPL", price=190.00, previous_price=190.00, open_price=190.00
+        )
+        assert update.tick_direction == "flat"
+
+    def test_change_today(self):
+        """Test the daily change is measured against open_price, not previous tick."""
+        update = PriceUpdate(
+            ticker="AAPL", price=195.00, previous_price=194.99, open_price=190.00
+        )
+        assert update.change_today == 5.00
+
+    def test_change_percent_today(self):
+        """Test the daily percentage change is measured against open_price."""
+        update = PriceUpdate(
+            ticker="AAPL", price=209.00, previous_price=208.99, open_price=190.00
+        )
+        assert update.change_percent_today == pytest.approx(10.0, abs=0.001)
+
+    def test_change_percent_today_zero_open(self):
+        """Test daily percentage change with a zero open_price (guard against ZeroDivisionError)."""
+        update = PriceUpdate(
+            ticker="AAPL", price=100.00, previous_price=100.00, open_price=0.00
+        )
+        assert update.change_percent_today == 0.0
+
+    def test_change_today_independent_of_tick_change(self):
+        """A ticker can be flat tick-to-tick but still up/down on the day."""
+        update = PriceUpdate(
+            ticker="AAPL", price=200.00, previous_price=200.00, open_price=190.00
+        )
+        assert update.tick_direction == "flat"
+        assert update.change_today == 10.00
+        assert update.change_percent_today == pytest.approx(5.263, abs=0.001)
 
     def test_to_dict(self):
         """Test serialization to dictionary."""
-        update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0)
+        update = PriceUpdate(
+            ticker="AAPL",
+            price=190.50,
+            previous_price=190.00,
+            open_price=180.00,
+            timestamp=1234567890.0,
+        )
         result = update.to_dict()
 
         assert result["ticker"] == "AAPL"
         assert result["price"] == 190.50
         assert result["previous_price"] == 190.00
-        assert result["timestamp"] == 1234567890.0
+        assert result["open_price"] == 180.00
+        assert result["timestamp"] == epoch_to_iso(1234567890.0)
         assert result["change"] == 0.50
         assert result["change_percent"] == 0.2632  # (0.50 / 190.00) * 100
-        assert result["direction"] == "up"
+        assert result["tick_direction"] == "up"
+        assert result["change_today"] == 10.50
+        assert "change_percent_today" in result
+
+    def test_to_dict_timestamp_is_iso_string(self):
+        """Wire format is ISO 8601 UTC, not a raw float, per PLAN.md §13.1 item 6."""
+        update = PriceUpdate(
+            ticker="AAPL",
+            price=190.50,
+            previous_price=190.00,
+            open_price=190.00,
+            timestamp=1234567890.0,
+        )
+        result = update.to_dict()
+        assert isinstance(result["timestamp"], str)
+        assert result["timestamp"].endswith("Z")
 
     def test_immutability(self):
         """Test that PriceUpdate is immutable."""
-        update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0)
+        update = PriceUpdate(
+            ticker="AAPL", price=190.50, previous_price=190.00, open_price=190.00
+        )
 
         with pytest.raises(AttributeError):
             update.price = 200.00  # Should raise error
+
+
+class TestEpochToIso:
+    """Unit tests for the epoch_to_iso wire-format helper."""
+
+    def test_returns_utc_z_suffix(self):
+        assert epoch_to_iso(0).startswith("1970-01-01T00:00:00")
+        assert epoch_to_iso(0).endswith("Z")
+
+    def test_preserves_subsecond_precision(self):
+        result = epoch_to_iso(1234567890.4137)
+        assert "413700" in result or "413" in result
+
+
+class TestPricePoint:
+    """Unit tests for the PricePoint model."""
+
+    def test_creation(self):
+        point = PricePoint(timestamp="2026-09-03T17:42:11Z", price=325.41)
+        assert point.timestamp == "2026-09-03T17:42:11Z"
+        assert point.price == 325.41
+
+    def test_to_dict(self):
+        point = PricePoint(timestamp="2026-09-03T17:42:11Z", price=325.41)
+        assert point.to_dict() == {"timestamp": "2026-09-03T17:42:11Z", "price": 325.41}
+
+    def test_immutability(self):
+        point = PricePoint(timestamp="2026-09-03T17:42:11Z", price=325.41)
+        with pytest.raises(AttributeError):
+            point.price = 100.0
+
+
+class TestSourceStatus:
+    """Unit tests for the SourceStatus model."""
+
+    def test_creation(self):
+        status = SourceStatus(
+            name="simulator", live=False, detail="synthetic", tickers=10, cache_populated=True
+        )
+        assert status.name == "simulator"
+        assert status.live is False
+        assert status.tickers == 10
+        assert status.cache_populated is True
+
+    def test_to_dict(self):
+        status = SourceStatus(
+            name="massive", live=True, detail="real-time", tickers=5, cache_populated=True
+        )
+        assert status.to_dict() == {
+            "name": "massive",
+            "live": True,
+            "detail": "real-time",
+            "tickers": 5,
+            "cache_populated": True,
+        }
+
+    def test_immutability(self):
+        status = SourceStatus(
+            name="simulator", live=False, detail="x", tickers=0, cache_populated=False
+        )
+        with pytest.raises(AttributeError):
+            status.live = True
diff --git a/backend/tests/market/test_simulator.py b/backend/tests/market/test_simulator.py
index 1845ec16b..2b3492d9f 100644
--- a/backend/tests/market/test_simulator.py
+++ b/backend/tests/market/test_simulator.py
@@ -1,7 +1,13 @@
 """Tests for GBMSimulator."""
 
-from app.market.seed_prices import SEED_PRICES
-from app.market.simulator import GBMSimulator
+import math
+import random
+
+import numpy as np
+import pytest
+
+from app.market.seed_prices import SEED_PRICES, TICKER_PARAMS
+from app.market.simulator import GBMSimulator, synthesize_seed
 
 
 class TestGBMSimulator:
@@ -52,13 +58,6 @@ def test_remove_nonexistent_is_noop(self):
         sim = GBMSimulator(tickers=["AAPL"])
         sim.remove_ticker("NOPE")  # Should not raise
 
-    def test_unknown_ticker_gets_random_seed_price(self):
-        """Test that unknown tickers get random seed prices."""
-        sim = GBMSimulator(tickers=["ZZZZ"])
-        price = sim.get_price("ZZZZ")
-        assert price is not None
-        assert 50.0 <= price <= 300.0
-
     def test_empty_step(self):
         """Test stepping with no tickers."""
         sim = GBMSimulator(tickers=[])
@@ -89,6 +88,34 @@ def test_cholesky_none_with_one_ticker(self):
         sim = GBMSimulator(tickers=["AAPL"])
         assert sim._cholesky is None
 
+    def test_cholesky_survives_many_unknown_tickers(self):
+        """The correlation matrix must stay positive-definite as tickers are added
+        one at a time (the incremental path, not the batch constructor) — a failure
+        here would 500 a watchlist add."""
+        sim = GBMSimulator(tickers=["AAPL"])
+        for i in range(100):
+            sim.add_ticker(f"ZZ{i:03d}")
+        assert sim._cholesky is not None
+        result = sim.step()
+        assert len(result) == 101
+
+    def test_cholesky_linalg_error_falls_back_to_identity(self, monkeypatch):
+        """A non-positive-definite matrix must degrade to uncorrelated moves,
+        not raise and crash the caller (e.g. a watchlist add)."""
+
+        def raise_linalg_error(_matrix):
+            raise np.linalg.LinAlgError("not positive definite")
+
+        sim = GBMSimulator(tickers=["AAPL"])
+        monkeypatch.setattr(np.linalg, "cholesky", raise_linalg_error)
+        sim.add_ticker("GOOGL")  # triggers _rebuild_cholesky with 2 tickers
+
+        assert sim._cholesky is not None
+        np.testing.assert_array_equal(sim._cholesky, np.eye(2))
+        # Simulator must still be usable.
+        result = sim.step()
+        assert set(result.keys()) == {"AAPL", "GOOGL"}
+
     def test_get_price_returns_none_for_unknown(self):
         """Test that get_price returns None for unknown ticker."""
         sim = GBMSimulator(tickers=["AAPL"])
@@ -120,12 +147,261 @@ def test_default_dt_is_reasonable(self):
         """Test that default dt is a reasonable small value."""
         assert 0 < GBMSimulator.DEFAULT_DT < 0.0001
 
+    def test_default_event_probability_is_calibrated(self):
+        """Regression guard: event_probability must stay at the calibrated 1e-4,
+        not the old 0.001 which inflated realized volatility ~20x
+        (planning/MARKET_SIMULATOR.md §6)."""
+        assert GBMSimulator.DEFAULT_EVENT_PROBABILITY == 1e-4
+
     def test_prices_rounded_to_two_decimals(self):
         """Test that prices are rounded to 2 decimal places."""
         sim = GBMSimulator(tickers=["AAPL"])
         result = sim.step()
         price_str = str(result["AAPL"])
         # Check that we have at most 2 decimal places
-        if '.' in price_str:
-            decimal_part = price_str.split('.')[1]
+        if "." in price_str:
+            decimal_part = price_str.split(".")[1]
             assert len(decimal_part) <= 2
+
+    def test_reset_price(self):
+        """reset_price rewinds the price and clears any active shock."""
+        sim = GBMSimulator(tickers=["AAPL"])
+        sim.step()
+        sim.reset_price("AAPL", 999.99)
+        assert sim.get_price("AAPL") == 999.99
+        assert "AAPL" not in sim._shocks
+
+    def test_reset_price_unknown_ticker_is_noop(self):
+        sim = GBMSimulator(tickers=["AAPL"])
+        sim.reset_price("NOPE", 1.0)  # should not raise
+
+
+class TestSynthesizeSeed:
+    """Unit tests for deterministic unknown-ticker seed synthesis.
+
+    Regression guard for planning/MARKET_SIMULATOR.md §3: unknown tickers used
+    to get `random.uniform(50, 300)`, a different price every restart, which
+    made a held position's cost basis jump and lied to the P&L chart.
+    """
+
+    def test_deterministic_across_calls(self):
+        assert synthesize_seed("ZZZZ") == synthesize_seed("ZZZZ")
+
+    def test_deterministic_across_fresh_instances(self):
+        """Same ticker, same price, across two independently constructed simulators."""
+        sim1 = GBMSimulator(tickers=["ZZZZ"])
+        sim2 = GBMSimulator(tickers=["ZZZZ"])
+        assert sim1.get_price("ZZZZ") == sim2.get_price("ZZZZ")
+
+    def test_different_tickers_differ(self):
+        # Not a hard guarantee for arbitrary hashes, but true for these symbols
+        # and a good smoke test that the hash actually depends on the ticker.
+        assert synthesize_seed("ZZZZ") != synthesize_seed("QQQQ")
+
+    def test_in_expected_range(self):
+        for ticker in ["ZZZZ", "QQQQ", "FOOBAR", "X"]:
+            price = synthesize_seed(ticker)
+            assert 20.0 <= price <= 500.0
+
+    def test_unknown_ticker_uses_synthesized_seed(self):
+        sim = GBMSimulator(tickers=["ZZZZ"])
+        assert sim.get_price("ZZZZ") == synthesize_seed("ZZZZ")
+
+    def test_unknown_ticker_uses_default_params(self):
+        sim = GBMSimulator(tickers=["ZZZZ"])
+        assert sim._params["ZZZZ"]["sigma"] == 0.25
+        assert sim._params["ZZZZ"]["mu"] == 0.05
+
+
+class TestSeedOverrides:
+    """Unit tests for anchoring the simulator to real (Massive-sourced) prices."""
+
+    def test_seed_override_takes_priority_over_static_table(self):
+        sim = GBMSimulator(tickers=["AAPL"], seed_overrides={"AAPL": 999.99})
+        assert sim.get_price("AAPL") == 999.99
+
+    def test_seed_override_used_for_unknown_ticker(self):
+        sim = GBMSimulator(tickers=["ZZZZ"], seed_overrides={"ZZZZ": 42.42})
+        assert sim.get_price("ZZZZ") == 42.42
+
+    def test_missing_override_falls_back_to_static_table(self):
+        sim = GBMSimulator(tickers=["AAPL", "GOOGL"], seed_overrides={"AAPL": 999.99})
+        assert sim.get_price("GOOGL") == SEED_PRICES["GOOGL"]
+
+    def test_missing_override_and_missing_table_entry_synthesizes(self):
+        sim = GBMSimulator(tickers=["ZZZZ"], seed_overrides={"AAPL": 999.99})
+        assert sim.get_price("ZZZZ") == synthesize_seed("ZZZZ")
+
+
+class TestShockDynamics:
+    """Unit tests for the decaying shock overlay.
+
+    Regression guard for the historical bug: shocks used to be a *permanent*
+    multiplicative level shift, which — at the old event_probability — inflated
+    realized volatility roughly 20x and made every per-ticker sigma decorative
+    (planning/MARKET_SIMULATOR.md §6). Shocks must now decay back toward the
+    underlying (still-calibrated) GBM path.
+    """
+
+    def test_shock_is_applied_on_trigger(self, monkeypatch):
+        monkeypatch.setattr(random, "random", lambda: 0.0)  # always triggers
+        monkeypatch.setattr(random, "uniform", lambda a, b: 0.04)
+        monkeypatch.setattr(random, "choice", lambda seq: 1)
+
+        sim = GBMSimulator(tickers=["AAPL"], event_probability=1.0)
+        seed = sim.get_price("AAPL")
+        result = sim.step()
+
+        assert "AAPL" in sim._shocks
+        # Displayed price reflects the +4% shock (approximately, modulo the
+        # underlying GBM drift/diffusion for this one tick).
+        assert result["AAPL"] > seed
+
+    def test_underlying_price_is_never_permanently_shifted_by_a_shock(self, monkeypatch):
+        """The GBM path (`_prices`) must stay independent of the shock overlay —
+        only the *displayed* price reflects the shock."""
+        monkeypatch.setattr(random, "random", lambda: 0.0)
+        monkeypatch.setattr(random, "uniform", lambda a, b: 0.04)
+        monkeypatch.setattr(random, "choice", lambda seq: 1)
+
+        sim = GBMSimulator(tickers=["AAPL"], event_probability=1.0)
+        sim.step()
+        underlying = sim._prices["AAPL"]
+        displayed = sim.step()["AAPL"]
+
+        # The shock inflates the displayed price above the underlying GBM price.
+        assert displayed != round(underlying, 2)
+
+    def test_shock_decays_and_does_not_grow(self, monkeypatch):
+        monkeypatch.setattr(random, "random", lambda: 0.0)
+        monkeypatch.setattr(random, "uniform", lambda a, b: 0.04)
+        monkeypatch.setattr(random, "choice", lambda seq: 1)
+
+        sim = GBMSimulator(tickers=["AAPL"], event_probability=1.0)
+        sim.step()
+        initial_magnitude = abs(sim._shocks["AAPL"].magnitude)
+
+        # Stop triggering new shocks; let the existing one decay.
+        sim._event_prob = 0.0
+        for _ in range(50):
+            sim.step()
+
+        remaining = sim._shocks.get("AAPL")
+        if remaining is not None:
+            assert abs(remaining.magnitude) < initial_magnitude
+
+    def test_shock_eventually_expires(self, monkeypatch):
+        monkeypatch.setattr(random, "random", lambda: 0.0)
+        monkeypatch.setattr(random, "uniform", lambda a, b: 0.03)
+        monkeypatch.setattr(random, "choice", lambda seq: 1)
+
+        sim = GBMSimulator(tickers=["AAPL"], event_probability=1.0)
+        sim.step()
+        sim._event_prob = 0.0  # no further triggers
+        for _ in range(500):
+            sim.step()
+
+        assert "AAPL" not in sim._shocks
+
+
+class TestStatisticalCalibration:
+    """Lighter-weight statistical regression tests.
+
+    These are smoke tests, not the full empirical study in
+    planning/MARKET_SIMULATOR.md §2/§6 (200 one-day runs is too slow for a
+    unit test) — but they use generous, one-sided bounds specifically chosen
+    to catch a regression back to the old, uncalibrated event process
+    (0.001 probability, permanent shifts), which produced roughly a
+    10%-per-hour realized sd against a ~0.5% theoretical target.
+    """
+
+    def test_pure_gbm_daily_volatility_is_within_theory(self):
+        """With shocks disabled, realized log-return sd over many short runs
+        should track sigma * sqrt(steps * dt)."""
+        np.random.seed(42)
+        sigma = TICKER_PARAMS["AAPL"]["sigma"]
+        steps = 2000
+        runs = 60
+
+        log_returns = []
+        for _ in range(runs):
+            sim = GBMSimulator(tickers=["AAPL"], event_probability=0.0)
+            start = sim.get_price("AAPL")
+            for _ in range(steps):
+                sim.step()
+            end = sim.get_price("AAPL")
+            log_returns.append(math.log(end / start))
+
+        realized_sd = float(np.std(log_returns))
+        theoretical_sd = sigma * math.sqrt(steps * GBMSimulator.DEFAULT_DT)
+
+        # Generous tolerance band (statistical test, not an exact bound).
+        assert theoretical_sd * 0.4 < realized_sd < theoretical_sd * 2.5
+
+    def test_default_event_probability_does_not_blow_up_volatility(self):
+        """The historical bug inflated realized volatility ~20x. At the
+        calibrated default, realized sd should stay within a small multiple of
+        the pure-GBM theoretical value — nowhere near the old ~20x blowup."""
+        np.random.seed(7)
+        random.seed(7)
+        sigma = TICKER_PARAMS["AAPL"]["sigma"]
+        steps = 3600  # ~30 minutes of trading time
+        runs = 40
+
+        log_returns = []
+        for _ in range(runs):
+            sim = GBMSimulator(tickers=["AAPL"])  # production default event_probability
+            start = sim.get_price("AAPL")
+            for _ in range(steps):
+                sim.step()
+            end = sim.get_price("AAPL")
+            log_returns.append(math.log(end / start))
+
+        realized_sd = float(np.std(log_returns))
+        theoretical_sd = sigma * math.sqrt(steps * GBMSimulator.DEFAULT_DT)
+
+        # The old bug produced ~20x theory; the fix should stay under ~6x even
+        # with sampling noise from a modest run count.
+        assert realized_sd < theoretical_sd * 6
+
+    def test_correlation_is_applied_not_bypassed(self):
+        """Realized correlation between two tech tickers should be
+        substantially higher than between a tech and a finance ticker —
+        confirms the Cholesky decomposition is actually wired in."""
+        np.random.seed(11)
+        steps = 3000
+
+        sim = GBMSimulator(tickers=["AAPL", "GOOGL", "JPM"], event_probability=0.0)
+        paths = {"AAPL": [], "GOOGL": [], "JPM": []}
+        for _ in range(steps):
+            prices = sim.step()
+            for t in paths:
+                paths[t].append(prices[t])
+
+        def log_returns(series):
+            arr = np.array(series, dtype=float)
+            return np.diff(np.log(arr))
+
+        tech_corr = float(np.corrcoef(log_returns(paths["AAPL"]), log_returns(paths["GOOGL"]))[0, 1])
+        cross_corr = float(np.corrcoef(log_returns(paths["AAPL"]), log_returns(paths["JPM"]))[0, 1])
+
+        assert tech_corr > cross_corr
+        assert tech_corr > 0.3
+
+
+class TestVisibleTickRate:
+    """Regression guard for planning/MARKET_SIMULATOR.md §2: at real anchored
+    price levels, the 2-decimal displayed price must actually change on most
+    ticks, or the flash animation and sparklines look dead."""
+
+    @pytest.mark.parametrize("ticker", list(SEED_PRICES.keys()))
+    def test_visible_tick_rate_exceeds_fifty_percent(self, ticker):
+        np.random.seed(3)
+        sim = GBMSimulator(tickers=[ticker], event_probability=0.0)
+        prices = [sim.get_price(ticker)]
+        for _ in range(1500):
+            prices.append(sim.step()[ticker])
+
+        visible = sum(1 for a, b in zip(prices, prices[1:]) if a != b)
+        rate = visible / (len(prices) - 1)
+        assert rate > 0.5, f"{ticker} visible-tick rate {rate:.2%} too low"
diff --git a/backend/tests/market/test_simulator_source.py b/backend/tests/market/test_simulator_source.py
index 515ce7290..6940da371 100644
--- a/backend/tests/market/test_simulator_source.py
+++ b/backend/tests/market/test_simulator_source.py
@@ -5,6 +5,7 @@
 import pytest
 
 from app.market.cache import PriceCache
+from app.market.models import SourceStatus
 from app.market.simulator import SimulatorDataSource
 
 
@@ -136,3 +137,149 @@ async def test_custom_event_probability(self):
         # Just verify it starts and stops cleanly
         await asyncio.sleep(0.2)
         await source.stop()
+
+    async def test_open_price_seeded_from_start_price(self):
+        """The first cache write's open_price should be the session seed."""
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["AAPL"])
+
+        update = cache.get("AAPL")
+        assert update.open_price == update.price  # nothing has moved yet
+
+        await source.stop()
+
+    async def test_open_price_stable_across_ticks(self):
+        """open_price must stay fixed for the session even as price ticks."""
+        cache = PriceCache()
+        source = SimulatorDataSource(
+            price_cache=cache, update_interval=0.02, event_probability=0.0
+        )
+        await source.start(["AAPL"])
+        first_open = cache.get("AAPL").open_price
+
+        await asyncio.sleep(0.15)
+
+        assert cache.get("AAPL").open_price == first_open
+        await source.stop()
+
+    async def test_seed_overrides_used_for_open_price(self):
+        """Anchored (real-close) seeds should be used verbatim as the session open."""
+        cache = PriceCache()
+        source = SimulatorDataSource(
+            price_cache=cache, update_interval=0.1, seed_overrides={"AAPL": 500.00}
+        )
+        await source.start(["AAPL"])
+
+        update = cache.get("AAPL")
+        assert update.price == 500.00
+        assert update.open_price == 500.00
+
+        await source.stop()
+
+    async def test_describe_returns_source_status(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["AAPL", "GOOGL"])
+
+        status = source.describe()
+        assert isinstance(status, SourceStatus)
+        assert status.name == "simulator"
+        assert status.live is False
+        assert status.tickers == 2
+        assert status.cache_populated is True
+
+        await source.stop()
+
+    async def test_describe_before_start_reports_empty(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache)
+        status = source.describe()
+        assert status.tickers == 0
+        assert status.cache_populated is False
+
+    async def test_describe_uses_custom_status_detail(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, status_detail="key rejected: bad key")
+        status = source.describe()
+        assert status.detail == "key rejected: bad key"
+
+    async def test_set_status_detail_updates_describe(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache)
+        source.set_status_detail("re-anchored")
+        assert source.describe().detail == "re-anchored"
+
+    async def test_get_history_empty_before_start(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache)
+        history = await source.get_history("AAPL")
+        assert history == []
+
+    async def test_get_history_prefilled_on_start(self):
+        """The chart must not be empty on first paint — start() prefills a
+        ring buffer of history rather than waiting for live ticks."""
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["AAPL"])
+
+        history = await source.get_history("AAPL")
+        assert len(history) > 1
+        # Timestamps should be non-decreasing and end at/near "now".
+        assert all(a.timestamp <= b.timestamp for a, b in zip(history, history[1:]))
+
+        await source.stop()
+
+    async def test_get_history_respects_points_limit(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["AAPL"])
+
+        history = await source.get_history("AAPL", points=5)
+        assert len(history) == 5
+
+        await source.stop()
+
+    async def test_prefill_resets_live_price_to_seed(self):
+        """The prefill run must not leave the simulator's live price at wherever
+        the prefill's random walk happened to end up."""
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["AAPL"])
+
+        update = cache.get("AAPL")
+        assert update.price == update.open_price
+
+        await source.stop()
+
+    async def test_get_history_unknown_ticker_is_empty(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["AAPL"])
+
+        assert await source.get_history("NOPE") == []
+
+        await source.stop()
+
+    async def test_add_ticker_history_starts_tracking(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.05)
+        await source.start(["AAPL"])
+
+        await source.add_ticker("TSLA")
+        await asyncio.sleep(0.2)
+
+        history = await source.get_history("TSLA")
+        assert len(history) >= 1
+
+        await source.stop()
+
+    async def test_remove_ticker_clears_history(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["AAPL"])
+        await source.remove_ticker("AAPL")
+
+        assert await source.get_history("AAPL") == []
+
+        await source.stop()
diff --git a/backend/tests/market/test_stream.py b/backend/tests/market/test_stream.py
new file mode 100644
index 000000000..5b1be6065
--- /dev/null
+++ b/backend/tests/market/test_stream.py
@@ -0,0 +1,121 @@
+"""Tests for the SSE streaming generator.
+
+planning/MARKET_INTERFACE.md §8 (PLAN.md §13.4 item 31) drops the Playwright
+"disconnect and verify reconnection" E2E case in favor of covering the server
+side here — an EventSource's auto-reconnect is a browser behavior we can't
+usefully re-test, but the heartbeat and disconnect-detection logic in
+`_generate_events` are ours to verify directly.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.market.cache import PriceCache
+from app.market.stream import _generate_events
+
+
+def _make_request(disconnected_after: int | None = None) -> MagicMock:
+    """A fake Request whose is_disconnected() returns True after N calls."""
+    request = MagicMock()
+    request.client = MagicMock(host="127.0.0.1")
+
+    calls = {"n": 0}
+
+    async def is_disconnected():
+        calls["n"] += 1
+        if disconnected_after is None:
+            return False
+        return calls["n"] > disconnected_after
+
+    request.is_disconnected = is_disconnected
+    return request
+
+
+@pytest.mark.asyncio
+class TestGenerateEvents:
+    async def test_first_event_is_retry_directive(self):
+        cache = PriceCache()
+        request = _make_request(disconnected_after=0)
+
+        events = [e async for e in _generate_events(cache, request, interval=0.01)]
+        assert events[0] == "retry: 1000\n\n"
+
+    async def test_sends_data_when_cache_populated(self):
+        cache = PriceCache()
+        cache.update("AAPL", 190.50)
+        request = _make_request(disconnected_after=1)
+
+        events = [e async for e in _generate_events(cache, request, interval=0.01)]
+        data_events = [e for e in events if e.startswith("data:")]
+        assert len(data_events) == 1
+        assert "AAPL" in data_events[0]
+
+    async def test_no_data_event_when_cache_empty(self):
+        cache = PriceCache()
+        request = _make_request(disconnected_after=1)
+
+        events = [e async for e in _generate_events(cache, request, interval=0.01)]
+        data_events = [e for e in events if e.startswith("data:")]
+        assert data_events == []
+
+    async def test_stops_on_disconnect(self):
+        cache = PriceCache()
+        request = _make_request(disconnected_after=0)
+
+        events = [e async for e in _generate_events(cache, request, interval=0.01)]
+        # Only the initial retry directive — the loop must exit before sleeping/looping.
+        assert events == ["retry: 1000\n\n"]
+
+    async def test_only_sends_once_per_version_change(self):
+        """Re-fetching an unchanged cache must not re-emit the same data event."""
+        cache = PriceCache()
+        cache.update("AAPL", 190.50)
+        request = _make_request(disconnected_after=3)
+
+        events = [e async for e in _generate_events(cache, request, interval=0.01)]
+        data_events = [e for e in events if e.startswith("data:")]
+        assert len(data_events) == 1
+
+    async def test_heartbeat_sent_after_silence(self):
+        """No price change for longer than heartbeat_interval → an SSE comment ping,
+        so a proxy or idle Massive poll doesn't silently drop the connection."""
+        cache = PriceCache()
+        request = _make_request(disconnected_after=2)
+
+        events = [
+            e
+            async for e in _generate_events(
+                cache, request, interval=0.01, heartbeat_interval=0.0
+            )
+        ]
+        assert any(e == ": ping\n\n" for e in events)
+
+    async def test_no_heartbeat_before_interval_elapses(self):
+        cache = PriceCache()
+        request = _make_request(disconnected_after=2)
+
+        events = [
+            e
+            async for e in _generate_events(
+                cache, request, interval=0.01, heartbeat_interval=1000.0
+            )
+        ]
+        assert not any(e == ": ping\n\n" for e in events)
+
+    async def test_new_data_resets_heartbeat_clock(self):
+        """A version bump should count as activity — no ping should immediately
+        follow a fresh data event."""
+        cache = PriceCache()
+        cache.update("AAPL", 190.50)
+        request = _make_request(disconnected_after=1)
+
+        events = [
+            e
+            async for e in _generate_events(
+                cache, request, interval=0.01, heartbeat_interval=0.0
+            )
+        ]
+        # First event after retry is the data event, not a ping, even though
+        # heartbeat_interval is 0 — a version change takes priority.
+        assert events[1].startswith("data:")
diff --git a/planning/MARKET_DATA_SUMMARY.md b/planning/MARKET_DATA_SUMMARY.md
new file mode 100644
index 000000000..346e9b83c
--- /dev/null
+++ b/planning/MARKET_DATA_SUMMARY.md
@@ -0,0 +1,53 @@
+# Market Data — Summary
+
+**Status:** Complete. Implements the design in `MARKET_INTERFACE.md`, `MARKET_SIMULATOR.md`,
+and `MASSIVE_API.md`. Lives entirely in `backend/app/market/`, with tests in
+`backend/tests/market/`.
+
+## What's There
+
+- **`PriceCache`** (`cache.py`) — thread-safe in-memory store, one `PriceUpdate` per ticker,
+  with a monotonic `version` counter for SSE change detection.
+- **`PriceUpdate` / `PricePoint` / `SourceStatus`** (`models.py`) — `PriceUpdate` carries both
+  a tick-to-tick delta (`previous_price`, `change`, `change_percent`, `tick_direction` — drives
+  the flash animation) and a session-baseline delta (`open_price`, `change_today`,
+  `change_percent_today` — drives the daily % column). Wire format is ISO 8601 UTC.
+- **`MarketDataSource`** (`interface.py`) — the shared abstract contract: `start`/`stop`/
+  `add_ticker`/`remove_ticker`/`get_tickers`, plus `describe()` (health introspection) and
+  `get_history()` (chart first-paint backfill).
+- **Three implementations**, selected by `create_market_data_source()` (`factory.py`, async)
+  based on a one-time Massive entitlement probe (`capabilities.py`) rather than key presence
+  alone — a free/Basic-tier key authenticates but can't return a live price, so it is routed
+  away from `MassiveDataSource` instead of silently producing an empty watchlist:
+  - **`SimulatorDataSource`** (`simulator.py`) — no key. GBM price motion, correlated across a
+    sector structure, with a deterministic (hash-based) seed for unknown tickers and a
+    decaying (not permanent) shock overlay for occasional drama. Keeps a 5-minute ring buffer
+    per ticker, prefilled at startup so charts aren't empty on first paint.
+  - **`AnchoredSimulatorDataSource`** (`anchored.py`) — key present but end-of-day only (the
+    free tier). One API call (`get_grouped_daily_aggs`) anchors the whole watchlist to real
+    closing prices; the GBM simulator then supplies tick-to-tick motion on top. Re-anchors
+    hourly. Badged as simulated (`SourceStatus.live == False`).
+  - **`MassiveDataSource`** (`massive_client.py`) — key entitled to real-time snapshots (paid
+    tiers). Polls `get_snapshot_all` on an interval; corrected nanosecond `sip_timestamp`
+    handling, captures `prev_day.close` as `open_price`, classifies rate-limit errors
+    distinctly from bad responses, and surfaces tickers missing from a poll via `describe()`.
+  - Any key state that can't be classified falls back to `SimulatorDataSource` — the app never
+    boots into a broken, empty-watchlist state.
+- **SSE streaming** (`stream.py`) — `GET /api/stream/prices`, full price map on every cache
+  version change, with an SSE comment heartbeat (`: ping`) when the stream would otherwise sit
+  silent (matters most against a 15s Massive poll interval or a closed market).
+- **`seed_prices.py`** — static fallback seeds/volatility/correlation params for the no-key
+  case, kept current against real closing prices.
+
+## Testing
+
+`backend/tests/market/` covers all of the above with mocked Massive calls (no test hits the
+network) plus statistical smoke tests for the GBM calibration, the shock decay behavior, and
+correlation. Run with `uv run --extra dev pytest -v` from `backend/`.
+
+## Details
+
+See `MARKET_INTERFACE.md` (source selection, `PriceUpdate`/`SourceStatus` design, SSE wire
+format), `MARKET_SIMULATOR.md` (GBM math, calibration, shock design, seeding), and
+`MASSIVE_API.md` (entitlement research, endpoint reference, rate-limit gotchas) for the full
+design rationale.