Complete the market data backend: session baseline, backfill, heartbeat

backend/app/market/__init__.py

@@ -3,21 +3,28 @@
 Public API:
     PriceUpdate         - Immutable price snapshot dataclass
     PriceCache          - Thread-safe in-memory price store
+    wait_for_price       - Poll the cache for a first tick on a new ticker
     MarketDataSource    - Abstract interface for data providers
     create_market_data_source - Factory that selects simulator or Massive
     create_stream_router - FastAPI router factory for SSE endpoint
+    TICKER_PATTERN       - Shared ticker validation regex
+    normalize_ticker      - Uppercase, strip, and validate a ticker symbol
 """
 
-from .cache import PriceCache
+from .cache import PriceCache, wait_for_price
 from .factory import create_market_data_source
 from .interface import MarketDataSource
 from .models import PriceUpdate
 from .stream import create_stream_router
+from .tickers import TICKER_PATTERN, normalize_ticker
 
 __all__ = [
-    "PriceUpdate",
-    "PriceCache",
     "MarketDataSource",
+    "PriceCache",
+    "PriceUpdate",
+    "TICKER_PATTERN",
     "create_market_data_source",
     "create_stream_router",
+    "normalize_ticker",
+    "wait_for_price",
 ]

backend/app/market/cache.py

@@ -2,68 +2,123 @@
 
 from __future__ import annotations
 
+import asyncio
 import time
+from collections import deque
 from threading import Lock
 
 from .models import PriceUpdate
 
+HISTORY_POINTS = 60  # Points retained per ticker for sparklines
+HISTORY_INTERVAL_SECONDS = 60.0  # Minimum spacing between recorded points
+
 
 class PriceCache:
-    """Thread-safe in-memory cache of the latest price for each ticker.
+    """Latest price, session baseline and recent history for each tracked ticker.
 
-    Writers: SimulatorDataSource or MassiveDataSource (one at a time).
-    Readers: SSE streaming endpoint, portfolio valuation, trade execution.
+    Writers: exactly one MarketDataSource (simulator or Massive poller).
+    Readers: SSE stream, portfolio valuation, trade execution, watchlist.
     """
 
-    def __init__(self) -> None:
+    def __init__(
+        self,
+        history_points: int = HISTORY_POINTS,
+        history_interval: float = HISTORY_INTERVAL_SECONDS,
+    ) -> None:
         self._prices: dict[str, PriceUpdate] = {}
+        self._history: dict[str, deque[float]] = {}
+        self._history_at: dict[str, float] = {}
+        self._history_points = history_points
+        self._history_interval = history_interval
         self._lock = Lock()
-        self._version: int = 0  # Monotonically increasing; bumped on every update
+        self._version = 0
+
+    # --- Writing ---
 
     def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate:
-        """Record a new price for a ticker. Returns the created PriceUpdate.
+        """Record a new price. Returns the stored PriceUpdate.
 
-        Automatically computes direction and change from the previous price.
-        If this is the first update for the ticker, previous_price == price (direction='flat').
+        Derives previous_price and open_price from the entry being replaced, so
+        sources stay dumb: they produce a number, the cache supplies the meaning.
+        The first update for a ticker pins its session baseline —
+        previous_price == open_price == price, direction 'flat', both changes 0.
         """
         with self._lock:
-            ts = timestamp or time.time()
-            prev = self._prices.get(ticker)
-            previous_price = prev.price if prev else price
+            ts = time.time() if timestamp is None else timestamp
+            price = round(price, 2)
+            previous = self._prices.get(ticker)
 
             update = PriceUpdate(
                 ticker=ticker,
-                price=round(price, 2),
-                previous_price=round(previous_price, 2),
+                price=price,
+                previous_price=previous.price if previous else price,
+                open_price=previous.open_price if previous else price,
                 timestamp=ts,
             )
             self._prices[ticker] = update
-            self._version += 1
+
+            # Version tracks *visible* change. A repeated price refreshes the
+            # timestamp (so /api/health still sees a live feed) without waking
+            # every SSE client to re-send an identical payload.
+            if previous is None or previous.price != price:
+                self._version += 1
+
+            self._record_history(ticker, update)
             return update
 
+    def seed_history(
+        self, ticker: str, prices: list[float], timestamp: float | None = None
+    ) -> None:
+        """Install backfilled history for a ticker, replacing anything present.
+
+        Called by a source at startup and when a ticker is added, so sparklines
+        are populated on first paint rather than filling in over 30 seconds.
+        """
+        with self._lock:
+            self._history[ticker] = deque(
+                (round(p, 2) for p in prices[-self._history_points :]),
+                maxlen=self._history_points,
+            )
+            self._history_at[ticker] = time.time() if timestamp is None else timestamp
+
+    def remove(self, ticker: str) -> None:
+        """Forget a ticker entirely — price, baseline and history."""
+        with self._lock:
+            self._prices.pop(ticker, None)
+            self._history.pop(ticker, None)
+            self._history_at.pop(ticker, None)
+
+    # --- Reading ---
+
     def get(self, ticker: str) -> PriceUpdate | None:
-        """Get the latest price for a single ticker, or None if unknown."""
         with self._lock:
             return self._prices.get(ticker)
 
+    def get_price(self, ticker: str) -> float | None:
+        update = self.get(ticker)
+        return update.price if update else None
+
     def get_all(self) -> dict[str, PriceUpdate]:
-        """Snapshot of all current prices. Returns a shallow copy."""
+        """Shallow copy of every current price. Safe to iterate without the lock."""
         with self._lock:
             return dict(self._prices)
 
-    def get_price(self, ticker: str) -> float | None:
-        """Convenience: get just the price float, or None."""
-        update = self.get(ticker)
-        return update.price if update else None
+    def get_history(self, ticker: str) -> list[float]:
+        """Recent prices, oldest first, up to history_points. Empty if unknown."""
+        with self._lock:
+            history = self._history.get(ticker)
+            return list(history) if history else []
 
-    def remove(self, ticker: str) -> None:
-        """Remove a ticker from the cache (e.g., when removed from watchlist)."""
+    def newest_timestamp(self) -> float | None:
+        """Timestamp of the most recently written price, for /api/health."""
         with self._lock:
-            self._prices.pop(ticker, None)
+            if not self._prices:
+                return None
+            return max(update.timestamp for update in self._prices.values())
 
     @property
     def version(self) -> int:
-        """Current version counter. Useful for SSE change detection."""
+        """Monotonic counter, bumped whenever a price actually changes."""
         return self._version
 
     def __len__(self) -> int:
@@ -73,3 +128,31 @@ def __len__(self) -> int:
     def __contains__(self, ticker: str) -> bool:
         with self._lock:
             return ticker in self._prices
+
+    # --- Internal (callers already hold the lock) ---
+
+    def _record_history(self, ticker: str, update: PriceUpdate) -> None:
+        history = self._history.get(ticker)
+        if history is None:
+            history = self._history[ticker] = deque(maxlen=self._history_points)
+
+        last_at = self._history_at.get(ticker)
+        if last_at is None or update.timestamp - last_at >= self._history_interval:
+            history.append(update.price)
+            self._history_at[ticker] = update.timestamp
+
+
+async def wait_for_price(cache: PriceCache, ticker: str, timeout: float = 2.0) -> float:
+    """Return the current price, waiting up to `timeout` for a first tick.
+
+    Raises ValueError with a user-facing message if no price arrives. Callers
+    translate that into a 400 (PLAN.md section 8).
+    """
+    deadline = time.monotonic() + timeout
+    while True:
+        price = cache.get_price(ticker)
+        if price is not None:
+            return price
+        if time.monotonic() >= deadline:
+            raise ValueError(f"No price available for {ticker} yet, please try again")
+        await asyncio.sleep(0.2)

backend/app/market/interface.py

@@ -8,50 +8,42 @@
 class MarketDataSource(ABC):
     """Contract for market data providers.
 
-    Implementations push price updates into a shared PriceCache on their own
-    schedule. Downstream code never calls the data source directly for prices —
-    it reads from the cache.
+    Implementations push prices into a shared PriceCache on their own schedule.
+    Downstream code never asks a source for a price — it reads the cache.
 
     Lifecycle:
         source = create_market_data_source(cache)
-        await source.start(["AAPL", "GOOGL", ...])
-        # ... app runs ...
+        await source.start(["AAPL", "GOOGL", ...])   # cache populated on return
         await source.add_ticker("TSLA")
         await source.remove_ticker("GOOGL")
-        # ... app shutting down ...
-        await source.stop()
+        await source.stop()                          # idempotent
     """
 
+    @property
+    @abstractmethod
+    def source_name(self) -> str:
+        """Short identifier for logs and /api/health: 'simulator' or 'massive'."""
+
     @abstractmethod
     async def start(self, tickers: list[str]) -> None:
-        """Begin producing price updates for the given tickers.
+        """Begin producing prices for `tickers`.
 
-        Starts a background task that periodically writes to the PriceCache.
-        Must be called exactly once. Calling start() twice is undefined behavior.
+        Must populate the cache (prices and seeded history) before returning, so
+        the first HTTP request never sees an empty cache. Called exactly once.
         """
 
     @abstractmethod
     async def stop(self) -> None:
-        """Stop the background task and release resources.
-
-        Safe to call multiple times. After stop(), the source will not write
-        to the cache again.
-        """
+        """Stop the background task. Idempotent, and never writes afterwards."""
 
     @abstractmethod
     async def add_ticker(self, ticker: str) -> None:
-        """Add a ticker to the active set. No-op if already present.
-
-        The next update cycle will include this ticker.
-        """
+        """Track a ticker. No-op if already tracked. Seeds price and history."""
 
     @abstractmethod
     async def remove_ticker(self, ticker: str) -> None:
-        """Remove a ticker from the active set. No-op if not present.
-
-        Also removes the ticker from the PriceCache.
-        """
+        """Stop tracking a ticker and remove it from the cache. No-op if absent."""
 
     @abstractmethod
     def get_tickers(self) -> list[str]:
-        """Return the current list of actively tracked tickers."""
+        """Currently tracked tickers. Synchronous — called from request handlers."""