Address market data backend review findings

README.md

@@ -78,7 +78,7 @@ Full rationale for these choices is in [`planning/PLAN.md`](planning/PLAN.md) §
 
 ## What's built so far: market data
 
-A self-contained market data subsystem lives in `backend/app/market/` — a `PriceCache`, a GBM-based simulator with correlated, per-sector price moves, a Massive/Polygon.io REST client behind the same interface, and an SSE stream factory. It's fully tested (73 tests, 91% coverage overall — `stream.py` is the weak spot at 33%, everything else is 94-100%) and has a standalone terminal demo:
+A self-contained market data subsystem lives in `backend/app/market/` — a `PriceCache`, a GBM-based simulator with correlated, per-sector price moves, a Massive/Polygon.io REST client behind the same interface, and an SSE stream factory. It's fully tested (79 tests, 99% coverage overall — every module is 94-100%, `stream.py` included) and has a standalone terminal demo:
 
 ```bash
 cd backend

backend/app/market/cache.py

@@ -27,7 +27,7 @@ def update(self, ticker: str, price: float, timestamp: float | None = None) -> P
         If this is the first update for the ticker, previous_price == price (direction='flat').
         """
         with self._lock:
-            ts = timestamp or time.time()
+            ts = timestamp if timestamp is not None else time.time()
             prev = self._prices.get(ticker)
             previous_price = prev.price if prev else price
 

backend/pyproject.toml

@@ -18,6 +18,7 @@ dev = [
     "pytest-asyncio>=0.24.0",
     "pytest-cov>=5.0.0",
     "ruff>=0.7.0",
+    "httpx>=0.27.0",
 ]
 
 [build-system]

backend/tests/market/test_models.py

@@ -10,55 +10,75 @@ 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, timestamp=1234567890.0
+        )
         assert update.ticker == "AAPL"
         assert update.price == 190.50
         assert update.previous_price == 190.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)
+        update = PriceUpdate(
+            ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0
+        )
         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)
+        update = PriceUpdate(
+            ticker="AAPL", price=189.50, previous_price=190.00, timestamp=1234567890.0
+        )
         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)
+        update = PriceUpdate(
+            ticker="AAPL", price=190.00, previous_price=100.00, timestamp=1234567890.0
+        )
         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)
+        update = PriceUpdate(
+            ticker="AAPL", price=100.00, previous_price=200.00, timestamp=1234567890.0
+        )
         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, timestamp=1234567890.0
+        )
         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)
+        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)
+        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)
+        update = PriceUpdate(
+            ticker="AAPL", price=190.00, previous_price=190.00, timestamp=1234567890.0
+        )
         assert update.direction == "flat"
 
     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, timestamp=1234567890.0
+        )
         result = update.to_dict()
 
         assert result["ticker"] == "AAPL"
@@ -71,7 +91,9 @@ def test_to_dict(self):
 
     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, timestamp=1234567890.0
+        )
 
         with pytest.raises(AttributeError):
             update.price = 200.00  # Should raise error

backend/tests/market/test_simulator.py

@@ -126,6 +126,6 @@ def test_prices_rounded_to_two_decimals(self):
         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

backend/tests/market/test_simulator_source.py

@@ -94,43 +94,52 @@ async def test_empty_start(self):
         await source.stop()
 
     async def test_exception_resilience(self):
-        """Test that simulator continues running after errors."""
+        """Test that the loop survives a step() failure and keeps ticking."""
         cache = PriceCache()
-        source = SimulatorDataSource(price_cache=cache, update_interval=0.05)
-
-        # Start with a valid ticker
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.02)
         await source.start(["AAPL"])
 
-        # Wait for some updates
+        real_step = source._sim.step
+        calls = {"count": 0}
+
+        def flaky_step():
+            calls["count"] += 1
+            if calls["count"] == 1:
+                raise RuntimeError("simulated step failure")
+            return real_step()
+
+        source._sim.step = flaky_step
+
+        # Wait long enough for the failing tick plus subsequent successful ticks
         await asyncio.sleep(0.15)
 
-        # Task should still be running
+        # Task survived the exception and kept running
         assert source._task is not None
         assert not source._task.done()
+        # And it kept producing updates after the injected failure
+        assert calls["count"] > 1
 
         await source.stop()
 
     async def test_custom_update_interval(self):
         """Test using a custom update interval."""
         cache = PriceCache()
-        source = SimulatorDataSource(price_cache=cache, update_interval=0.01)
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.02)
         await source.start(["AAPL"])
 
         initial_version = cache.version
-        await asyncio.sleep(0.05)  # Should get ~5 updates
+        await asyncio.sleep(0.15)
 
-        # Should have multiple updates with fast interval
-        assert cache.version > initial_version + 2
+        # Should have gotten at least one update with the fast interval
+        assert cache.version > initial_version
 
         await source.stop()
 
     async def test_custom_event_probability(self):
         """Test creating source with custom event probability."""
         cache = PriceCache()
         # Very high event probability for testing
-        source = SimulatorDataSource(
-            price_cache=cache, update_interval=0.1, event_probability=1.0
-        )
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1, event_probability=1.0)
         await source.start(["AAPL"])
 
         # Just verify it starts and stops cleanly

backend/tests/market/test_stream.py

@@ -0,0 +1,144 @@
+"""Tests for the SSE streaming endpoint (app/market/stream.py).
+
+Note: full HTTP round-trip testing via httpx's ASGITransport (or FastAPI's
+TestClient, which is built on the same buffering transport in the installed
+dependency versions) is not viable here — both drive the ASGI app to
+completion inside a single `await` before returning any response to the
+caller, which deadlocks against `_generate_events`'s infinite loop. Instead
+these tests drive the real generator and route-handler coroutine directly,
+which exercises the same code paths without relying on a transport that
+can stream partial responses.
+"""
+
+import asyncio
+import json
+
+import pytest
+from fastapi.responses import StreamingResponse
+
+from app.market.cache import PriceCache
+from app.market.stream import _generate_events, create_stream_router
+
+
+class _FakeClient:
+    def __init__(self, host: str = "test-client") -> None:
+        self.host = host
+
+
+class _FakeRequest:
+    """Minimal stand-in for fastapi.Request exposing what _generate_events uses."""
+
+    def __init__(self, disconnect_after: int | None = None) -> None:
+        self.client = _FakeClient()
+        self._calls = 0
+        self._disconnect_after = disconnect_after
+
+    async def is_disconnected(self) -> bool:
+        self._calls += 1
+        if self._disconnect_after is not None and self._calls > self._disconnect_after:
+            return True
+        return False
+
+
+@pytest.mark.asyncio
+async def test_generate_events_yields_retry_directive_first():
+    cache = PriceCache()
+    request = _FakeRequest(disconnect_after=0)
+    gen = _generate_events(cache, request, interval=0.01)
+
+    first = await gen.__anext__()
+    assert first == "retry: 1000\n\n"
+
+    with pytest.raises(StopAsyncIteration):
+        await gen.__anext__()
+
+
+@pytest.mark.asyncio
+async def test_generate_events_emits_prices_on_version_change():
+    cache = PriceCache()
+    cache.update("AAPL", 190.50)
+    request = _FakeRequest(disconnect_after=1)
+    gen = _generate_events(cache, request, interval=0.01)
+
+    await gen.__anext__()  # retry directive
+    data_event = await gen.__anext__()
+
+    assert data_event.startswith("data: ")
+    payload = json.loads(data_event.removeprefix("data: ").strip())
+    assert payload["AAPL"]["price"] == 190.50
+
+    with pytest.raises(StopAsyncIteration):
+        await gen.__anext__()
+
+
+@pytest.mark.asyncio
+async def test_generate_events_sends_no_data_event_when_cache_empty():
+    cache = PriceCache()
+    request = _FakeRequest(disconnect_after=2)
+    gen = _generate_events(cache, request, interval=0.01)
+
+    events = [event async for event in gen]
+
+    # Only the retry directive - no data event, since the cache never had prices
+    assert events == ["retry: 1000\n\n"]
+
+
+@pytest.mark.asyncio
+async def test_generate_events_skips_unchanged_version():
+    """A second tick with no cache write should not repeat the data event."""
+    cache = PriceCache()
+    cache.update("AAPL", 100.0)
+    request = _FakeRequest(disconnect_after=2)
+    gen = _generate_events(cache, request, interval=0.01)
+
+    events = [event async for event in gen]
+
+    # retry directive + exactly one data event (version unchanged on 2nd tick)
+    assert len(events) == 2
+    assert events[0] == "retry: 1000\n\n"
+    assert events[1].startswith("data: ")
+
+
+@pytest.mark.asyncio
+async def test_generate_events_stops_on_cancellation():
+    """_generate_events catches CancelledError internally (to log a clean
+    disconnect) rather than propagating it, so the consuming task finishes
+    normally instead of raising or hanging."""
+    cache = PriceCache()
+    cache.update("AAPL", 100.0)
+    request = _FakeRequest()  # never disconnects on its own
+
+    async def consume():
+        async for _ in _generate_events(cache, request, interval=0.05):
+            pass
+
+    task = asyncio.create_task(consume())
+    await asyncio.sleep(0.05)
+    task.cancel()
+
+    await asyncio.wait_for(task, timeout=1)
+    assert task.done()
+    assert not task.cancelled()
+
+
+@pytest.mark.asyncio
+async def test_stream_prices_route_returns_streaming_response():
+    """create_stream_router wires the route to a StreamingResponse over _generate_events."""
+    cache = PriceCache()
+    cache.update("AAPL", 190.50)
+    router = create_stream_router(cache)
+    endpoint = router.routes[-1].endpoint
+
+    response = await endpoint(_FakeRequest())
+    try:
+        assert isinstance(response, StreamingResponse)
+        assert response.media_type == "text/event-stream"
+        assert response.headers["cache-control"] == "no-cache"
+        assert response.headers["x-accel-buffering"] == "no"
+
+        first = await response.body_iterator.__anext__()
+        assert first == "retry: 1000\n\n"
+        second = await response.body_iterator.__anext__()
+        assert second.startswith("data: ")
+    finally:
+        await response.body_iterator.aclose()