Address market data review findings

backend/app/market/massive_client.py

@@ -37,6 +37,7 @@ def __init__(
         self._tickers: list[str] = []
         self._task: asyncio.Task | None = None
         self._client: RESTClient | None = None
+        self._consecutive_failures: int = 0
 
     async def start(self, tickers: list[str]) -> None:
         self._client = RESTClient(api_key=self._api_key)
@@ -64,6 +65,9 @@ async def stop(self) -> None:
         logger.info("Massive poller stopped")
 
     async def add_ticker(self, ticker: str) -> None:
+        # The new ticker has no price until the next poll cycle (up to
+        # poll_interval seconds later). We intentionally do not fetch it
+        # immediately to conserve the API rate-limit budget.
         ticker = ticker.upper().strip()
         if ticker not in self._tickers:
             self._tickers.append(ticker)
@@ -81,9 +85,15 @@ def get_tickers(self) -> list[str]:
     # --- Internal ---
 
     async def _poll_loop(self) -> None:
-        """Poll on interval. First poll already happened in start()."""
+        """Poll on interval. First poll already happened in start().
+
+        On consecutive failures, back off exponentially (capped) so a sustained
+        error such as a 429 rate-limit does not keep hammering the API at the
+        base interval.
+        """
         while True:
-            await asyncio.sleep(self._interval)
+            delay = self._interval * (2 ** min(self._consecutive_failures, 4))
+            await asyncio.sleep(delay)
             await self._poll_once()
 
     async def _poll_once(self) -> None:
@@ -93,8 +103,10 @@ async def _poll_once(self) -> None:
 
         try:
             # The Massive RESTClient is synchronous — run in a thread to
-            # avoid blocking the event loop.
-            snapshots = await asyncio.to_thread(self._fetch_snapshots)
+            # avoid blocking the event loop. Pass a copy of the ticker list so
+            # concurrent add/remove cannot mutate it mid-request.
+            snapshots = await asyncio.to_thread(self._fetch_snapshots, list(self._tickers))
+            self._consecutive_failures = 0
             processed = 0
             for snap in snapshots:
                 try:
@@ -116,13 +128,18 @@ async def _poll_once(self) -> None:
             logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers))
 
         except Exception as e:
-            logger.error("Massive poll failed: %s", e)
-            # Don't re-raise — the loop will retry on the next interval.
+            self._consecutive_failures += 1
+            logger.error("Massive poll failed (%d in a row): %s", self._consecutive_failures, e)
+            # Don't re-raise — the loop will retry with backoff on the next cycle.
             # Common failures: 401 (bad key), 429 (rate limit), network errors.
 
-    def _fetch_snapshots(self) -> list:
-        """Synchronous call to the Massive REST API. Runs in a thread."""
+    def _fetch_snapshots(self, tickers: list[str]) -> list:
+        """Synchronous call to the Massive REST API. Runs in a thread.
+
+        Receives a snapshot copy of the ticker list so concurrent
+        add/remove on the event loop cannot mutate it mid-request.
+        """
         return self._client.get_snapshot_all(
             market_type=SnapshotMarketType.STOCKS,
-            tickers=self._tickers,
+            tickers=tickers,
         )

backend/app/market/stream.py

@@ -14,14 +14,15 @@
 
 logger = logging.getLogger(__name__)
 
-router = APIRouter(prefix="/api/stream", tags=["streaming"])
-
 
 def create_stream_router(price_cache: PriceCache) -> APIRouter:
     """Create the SSE streaming router with a reference to the price cache.
 
-    This factory pattern lets us inject the PriceCache without globals.
+    The router is created fresh on each call so the factory is self-contained:
+    no shared module-global state, and calling it twice never double-registers
+    the route.
     """
+    router = APIRouter(prefix="/api/stream", tags=["streaming"])
 
     @router.get("/prices")
     async def stream_prices(request: Request) -> StreamingResponse:

backend/tests/market/test_stream.py

@@ -0,0 +1,74 @@
+"""Tests for the SSE streaming endpoint."""
+
+import json
+
+from app.market.cache import PriceCache
+from app.market.stream import _generate_events, create_stream_router
+
+
+class FakeRequest:
+    """Minimal stand-in for a Starlette Request.
+
+    is_disconnected() returns False for `alive_iterations` calls, then True,
+    so the generator loop terminates deterministically.
+    """
+
+    def __init__(self, alive_iterations: int = 1) -> None:
+        self._alive = alive_iterations
+        self.client = None
+
+    async def is_disconnected(self) -> bool:
+        if self._alive > 0:
+            self._alive -= 1
+            return False
+        return True
+
+
+async def test_create_stream_router_fresh_each_call():
+    """Each call returns a self-contained router with exactly one /prices route."""
+    cache = PriceCache()
+    r1 = create_stream_router(cache)
+    r2 = create_stream_router(cache)
+
+    assert r1 is not r2
+    paths = [route.path for route in r1.routes]
+    assert paths.count("/api/stream/prices") == 1
+
+
+async def test_generate_events_emits_retry_then_prices():
+    cache = PriceCache()
+    cache.update("AAPL", 190.0)
+
+    gen = _generate_events(cache, FakeRequest(alive_iterations=1), interval=0.0)
+
+    first = await anext(gen)
+    assert first == "retry: 1000\n\n"
+
+    payload = await anext(gen)
+    assert payload.startswith("data: ")
+    data = json.loads(payload[len("data: ") :].strip())
+    assert data["AAPL"]["price"] == 190.0
+    assert data["AAPL"]["direction"] == "flat"
+
+
+async def test_generate_events_only_sends_on_version_change():
+    cache = PriceCache()
+    cache.update("AAPL", 190.0)
+
+    # Allow several loop iterations but never change the cache after the first send.
+    gen = _generate_events(cache, FakeRequest(alive_iterations=3), interval=0.0)
+
+    events = [event async for event in gen]
+
+    # retry directive + exactly one data event (no duplicate sends for an unchanged cache)
+    data_events = [e for e in events if e.startswith("data: ")]
+    assert len(data_events) == 1
+
+
+async def test_generate_events_stops_on_disconnect():
+    cache = PriceCache()
+    # Disconnected immediately: only the retry directive is yielded.
+    gen = _generate_events(cache, FakeRequest(alive_iterations=0), interval=0.0)
+
+    events = [event async for event in gen]
+    assert events == ["retry: 1000\n\n"]