Add market data backend code review document

.github/workflows/claude-code-review.yml

@@ -27,7 +27,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
         with:
           fetch-depth: 1
 

.github/workflows/claude.yml

@@ -19,21 +19,23 @@ jobs:
       (github.event_name == 'issues' && (contains(github.event.issue.body, '@claude') || contains(github.event.issue.title, '@claude')))
     runs-on: ubuntu-latest
     permissions:
-      contents: read
+      contents: write
       pull-requests: read
       issues: read
       id-token: write
       actions: read # Required for Claude to read CI results on PRs
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
         with:
           fetch-depth: 1
 
       - name: Run Claude Code
         id: claude
         uses: anthropics/claude-code-action@v1
         with:
+          # Authenticate via the Claude.ai Pro subscription's usage allowance
+          # instead of pay-per-token API billing (the API key ran out of credits).
           claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }}
 
           # This is an optional setting that allows Claude to read CI results on PRs
@@ -43,8 +45,7 @@ jobs:
           # Optional: Give a custom prompt to Claude. If this is not specified, Claude will perform the instructions specified in the comment that tagged it.
           # prompt: 'Update the pull request description to include a summary of changes.'
 
-          # Optional: Add claude_args to customize behavior and configuration
-          # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md
-          # or https://code.claude.com/docs/en/cli-reference for available options
-          # claude_args: '--allowed-tools Bash(gh pr:*)'
+          # Use Sonnet instead of the default Opus[1m] to keep run cost down —
+          # the original attempt spent $5.34 over 81 turns on Opus without finishing.
+          claude_args: '--model claude-sonnet-5'
 

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/app/market/interface.py

@@ -42,7 +42,9 @@ async def stop(self) -> None:
     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.
+        The next update cycle will include this ticker. Behavior before
+        start() has been called is implementation-defined — callers should
+        always start() before adding tickers.
         """
 
     @abstractmethod

backend/app/market/massive_client.py

@@ -40,7 +40,7 @@ def __init__(
 
     async def start(self, tickers: list[str]) -> None:
         self._client = RESTClient(api_key=self._api_key)
-        self._tickers = list(tickers)
+        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()
@@ -99,10 +99,10 @@ async def _poll_once(self) -> None:
             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
+                    # Massive timestamps are Unix nanoseconds → convert to seconds
+                    timestamp = snap.last_trade.sip_timestamp / 1_000_000_000.0
                     self._cache.update(
-                        ticker=snap.ticker,
+                        ticker=snap.ticker.upper().strip(),
                         price=price,
                         timestamp=timestamp,
                     )

backend/app/market/simulator.py

@@ -217,6 +217,7 @@ def __init__(
         self._task: asyncio.Task | None = None
 
     async def start(self, tickers: list[str]) -> None:
+        tickers = [t.upper().strip() for t in tickers]
         self._sim = GBMSimulator(
             tickers=tickers,
             event_probability=self._event_prob,
@@ -240,6 +241,7 @@ async def stop(self) -> None:
         logger.info("Simulator stopped")
 
     async def add_ticker(self, ticker: str) -> None:
+        ticker = ticker.upper().strip()
         if self._sim:
             self._sim.add_ticker(ticker)
             # Seed cache immediately so the ticker has a price right away
@@ -249,6 +251,7 @@ async def add_ticker(self, ticker: str) -> None:
             logger.info("Simulator: added ticker %s", ticker)
 
     async def remove_ticker(self, ticker: str) -> None:
+        ticker = ticker.upper().strip()
         if self._sim:
             self._sim.remove_ticker(ticker)
         self._cache.remove(ticker)

backend/app/market/stream.py

@@ -14,14 +14,16 @@
 
 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.
+    This factory pattern lets us inject the PriceCache without globals. The
+    router is constructed here (not at module level) so calling this factory
+    multiple times — e.g. once per app instance in tests — never registers
+    routes on a shared router object.
     """
+    router = APIRouter(prefix="/api/stream", tags=["streaming"])
 
     @router.get("/prices")
     async def stream_prices(request: Request) -> StreamingResponse:

backend/tests/market/test_cache.py

@@ -1,5 +1,7 @@
 """Tests for PriceCache."""
 
+from concurrent.futures import ThreadPoolExecutor
+
 from app.market.cache import PriceCache
 
 
@@ -101,3 +103,39 @@ def test_price_rounding(self):
         cache = PriceCache()
         update = cache.update("AAPL", 190.12345)
         assert update.price == 190.12
+
+    def test_concurrent_writers_do_not_corrupt_state(self):
+        """Stress the lock with real OS threads (relevant because the Massive
+        path's synchronous SDK calls run via asyncio.to_thread, i.e. a real
+        thread, alongside the simulator/SSE reader on the event loop).
+
+        Many threads hammer update()/get()/remove() on overlapping tickers
+        concurrently; afterwards the cache must be internally consistent:
+        no lost/corrupted PriceUpdate, and the version counter must equal
+        exactly the number of update() calls that ran (no missed increments,
+        no double counting)."""
+        cache = PriceCache()
+        tickers = [f"T{i}" for i in range(8)]
+        writes_per_ticker = 200
+
+        def hammer(ticker: str) -> None:
+            for i in range(writes_per_ticker):
+                cache.update(ticker, price=100.0 + i)
+                cache.get(ticker)
+                cache.get_all()
+                if i % 50 == 0:
+                    cache.remove(ticker)
+
+        with ThreadPoolExecutor(max_workers=len(tickers)) as pool:
+            list(pool.map(hammer, tickers))
+
+        # Every ticker ends with a well-formed, non-corrupted update.
+        for ticker in tickers:
+            update = cache.get(ticker)
+            assert update is not None
+            assert update.ticker == ticker
+            assert update.price == 100.0 + (writes_per_ticker - 1)
+
+        # version is bumped exactly once per update() call, with no lost
+        # or duplicated increments under concurrent access.
+        assert cache.version == len(tickers) * writes_per_ticker

backend/tests/market/test_interface.py

@@ -0,0 +1,57 @@
+"""Tests for the MarketDataSource abstract interface contract."""
+
+import pytest
+
+from app.market.interface import MarketDataSource
+from app.market.massive_client import MassiveDataSource
+from app.market.simulator import SimulatorDataSource
+
+
+class TestMarketDataSourceABC:
+    """MarketDataSource is an ABC — it must not be directly instantiable."""
+
+    def test_cannot_instantiate_abstract_class(self):
+        with pytest.raises(TypeError):
+            MarketDataSource()
+
+    def test_incomplete_implementation_cannot_be_instantiated(self):
+        """A subclass missing an abstract method must fail to instantiate."""
+
+        class IncompleteSource(MarketDataSource):
+            async def start(self, tickers):
+                pass
+
+            async def stop(self):
+                pass
+
+            async def add_ticker(self, ticker):
+                pass
+
+            # remove_ticker and get_tickers intentionally omitted
+
+        with pytest.raises(TypeError):
+            IncompleteSource()
+
+
+class TestConformance:
+    """Both concrete implementations must satisfy the full interface contract."""
+
+    @pytest.mark.parametrize(
+        "cls,kwargs",
+        [
+            (SimulatorDataSource, {"price_cache": None}),
+            (MassiveDataSource, {"api_key": "test-key", "price_cache": None}),
+        ],
+    )
+    def test_is_instance_of_market_data_source(self, cls, kwargs):
+        from app.market.cache import PriceCache
+
+        kwargs = {**kwargs, "price_cache": PriceCache()}
+        instance = cls(**kwargs)
+        assert isinstance(instance, MarketDataSource)
+
+    @pytest.mark.parametrize("cls", [SimulatorDataSource, MassiveDataSource])
+    def test_implements_all_abstract_methods(self, cls):
+        for method_name in MarketDataSource.__abstractmethods__:
+            assert hasattr(cls, method_name)
+            assert callable(getattr(cls, method_name))

backend/tests/market/test_massive.py

@@ -3,18 +3,19 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from massive.rest.models.snapshot import TickerSnapshot
 
 from app.market.cache import PriceCache
 from app.market.massive_client import MassiveDataSource
 
 
-def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock:
+def _make_snapshot(ticker: str, price: float, timestamp_ns: int) -> MagicMock:
     """Create a mock Massive snapshot object."""
     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 = timestamp_ns
     return snap
 
 
@@ -34,8 +35,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, 1707580800000000000),
+            _make_snapshot("GOOGL", 175.25, 1707580800000000000),
         ]
 
         with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):
@@ -55,7 +56,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, 1707580800000000000)
         bad_snap = MagicMock()
         bad_snap.ticker = "BAD"
         bad_snap.last_trade = None  # Will cause AttributeError
@@ -84,7 +85,7 @@ 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."""
+        """Test that timestamps are converted from nanoseconds to seconds."""
         cache = PriceCache()
         source = MassiveDataSource(
             api_key="test-key",
@@ -94,7 +95,7 @@ async def test_timestamp_conversion(self):
         source._tickers = ["AAPL"]
         source._client = MagicMock()  # Satisfy the _poll_once guard
 
-        mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)]
+        mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000000000)]
 
         with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):
             await source._poll_once()
@@ -103,6 +104,54 @@ async def test_timestamp_conversion(self):
         assert update is not None
         assert update.timestamp == 1707580800.0  # Converted to seconds
 
+    async def test_poll_with_real_sdk_snapshot_shape(self):
+        """Regression test for the sip_timestamp/nanoseconds bug (see
+        planning/MARKET_DATA_REVIEW.md §3.1): build a real TickerSnapshot via
+        from_dict() instead of a MagicMock, so a mismatch between the code's
+        assumed attribute names and the actual SDK dataclass fails here
+        instead of failing silently in production."""
+        cache = PriceCache()
+        source = MassiveDataSource(
+            api_key="test-key",
+            price_cache=cache,
+            poll_interval=60.0,
+        )
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()  # Satisfy the _poll_once guard
+
+        real_snap = TickerSnapshot.from_dict(
+            {"ticker": "AAPL", "lastTrade": {"p": 190.50, "t": 1707580800123456789}}
+        )
+
+        with patch.object(source, "_fetch_snapshots", return_value=[real_snap]):
+            await source._poll_once()
+
+        update = cache.get("AAPL")
+        assert update is not None
+        assert update.price == 190.50
+        assert update.timestamp == pytest.approx(1707580800.123456789)
+
+    async def test_poll_normalizes_ticker_case(self):
+        """_poll_once() must normalize ticker casing the same way
+        start()/add_ticker()/remove_ticker() do, so a differently-cased
+        symbol in the API response doesn't create a second cache entry."""
+        cache = PriceCache()
+        source = MassiveDataSource(
+            api_key="test-key",
+            price_cache=cache,
+            poll_interval=60.0,
+        )
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()  # Satisfy the _poll_once guard
+
+        mock_snapshots = [_make_snapshot("aapl", 190.50, 1707580800000000000)]
+
+        with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):
+            await source._poll_once()
+
+        assert cache.get_price("AAPL") == 190.50
+        assert cache.get("aapl") is None
+
     async def test_add_ticker(self):
         """Test adding a ticker."""
         cache = PriceCache()
@@ -184,12 +233,27 @@ async def test_stop_cancels_task(self):
         await source.stop()
         assert source._task is None
 
+    async def test_start_normalizes_ticker_case(self):
+        """start() must normalize tickers the same way add_ticker/remove_ticker
+        do, so the same call produces the same casing regardless of which
+        method (or which data source) is used."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+
+        with patch("app.market.massive_client.RESTClient"):
+            with patch.object(source, "_fetch_snapshots", return_value=[]):
+                await source.start(["aapl", " googl "])
+
+        assert source.get_tickers() == ["AAPL", "GOOGL"]
+
+        await source.stop()
+
     async def test_start_immediate_poll(self):
         """Test that start() does an immediate poll before starting the loop."""
         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, 1707580800000000000)]
 
         with patch("app.market.massive_client.RESTClient"):
             with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots):