Claude/market data review 20260818

.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/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()

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_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

@@ -184,6 +184,21 @@ 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()

backend/tests/market/test_seed_prices.py

@@ -0,0 +1,60 @@
+"""Tests for seed_prices.py constants."""
+
+from app.market.seed_prices import (
+    CORRELATION_GROUPS,
+    CROSS_GROUP_CORR,
+    DEFAULT_PARAMS,
+    INTRA_FINANCE_CORR,
+    INTRA_TECH_CORR,
+    SEED_PRICES,
+    TICKER_PARAMS,
+    TSLA_CORR,
+)
+
+
+class TestSeedPrices:
+    def test_default_watchlist_tickers_present(self):
+        """The 10 default watchlist tickers from PLAN.md §7 must all be seeded."""
+        expected = {
+            "AAPL", "GOOGL", "MSFT", "AMZN", "TSLA",
+            "NVDA", "META", "JPM", "V", "NFLX",
+        }
+        assert expected == set(SEED_PRICES.keys())
+
+    def test_seed_prices_are_positive(self):
+        for ticker, price in SEED_PRICES.items():
+            assert price > 0, f"{ticker} seed price must be positive"
+
+    def test_every_seed_ticker_has_params(self):
+        """Every ticker with a seed price must also have GBM params."""
+        assert set(SEED_PRICES.keys()) == set(TICKER_PARAMS.keys())
+
+    def test_ticker_params_are_valid(self):
+        for ticker, params in TICKER_PARAMS.items():
+            assert "sigma" in params
+            assert "mu" in params
+            assert params["sigma"] > 0, f"{ticker} sigma must be positive"
+
+    def test_default_params_valid(self):
+        assert "sigma" in DEFAULT_PARAMS
+        assert "mu" in DEFAULT_PARAMS
+        assert DEFAULT_PARAMS["sigma"] > 0
+
+    def test_correlation_groups_reference_known_tickers(self):
+        """Every ticker in a correlation group must have a seed price."""
+        for group, tickers in CORRELATION_GROUPS.items():
+            for ticker in tickers:
+                assert ticker in SEED_PRICES, (
+                    f"{ticker} in correlation group {group!r} has no seed price"
+                )
+
+    def test_correlation_groups_are_disjoint(self):
+        """A ticker should not belong to more than one sector group."""
+        tech = CORRELATION_GROUPS["tech"]
+        finance = CORRELATION_GROUPS["finance"]
+        assert tech.isdisjoint(finance)
+
+    def test_correlation_coefficients_are_valid(self):
+        """Correlation coefficients must be valid correlation values."""
+        for corr in (INTRA_TECH_CORR, INTRA_FINANCE_CORR, CROSS_GROUP_CORR, TSLA_CORR):
+            assert -1.0 <= corr <= 1.0

backend/tests/market/test_simulator_source.py

@@ -124,6 +124,42 @@ async def test_custom_update_interval(self):
 
         await source.stop()
 
+    async def test_start_normalizes_ticker_case(self):
+        """Tickers passed to start() are normalized to uppercase, matching
+        MassiveDataSource, so the unified interface behaves consistently
+        regardless of which source is active."""
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["aapl", " googl "])
+
+        assert set(source.get_tickers()) == {"AAPL", "GOOGL"}
+        assert cache.get("AAPL") is not None
+        assert cache.get("aapl") is None
+
+        await source.stop()
+
+    async def test_add_ticker_normalizes_case(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["AAPL"])
+
+        await source.add_ticker("tsla")
+        assert "TSLA" in source.get_tickers()
+        assert cache.get("TSLA") is not None
+
+        await source.stop()
+
+    async def test_remove_ticker_normalizes_case(self):
+        cache = PriceCache()
+        source = SimulatorDataSource(price_cache=cache, update_interval=0.1)
+        await source.start(["AAPL", "TSLA"])
+
+        await source.remove_ticker("tsla")
+        assert "TSLA" not in source.get_tickers()
+        assert cache.get("TSLA") is None
+
+        await source.stop()
+
     async def test_custom_event_probability(self):
         """Test creating source with custom event probability."""
         cache = PriceCache()

backend/tests/market/test_stream.py

@@ -0,0 +1,98 @@
+"""Tests for the SSE streaming endpoint (stream.py)."""
+
+import json
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.market.cache import PriceCache
+from app.market.stream import _generate_events, create_stream_router
+
+
+def _make_request(disconnected_after: int = 1):
+    """Mock Request whose is_disconnected() returns False `disconnected_after`
+    times, then True — simulates a client that stays connected for N checks.
+    """
+    request = MagicMock()
+    request.client.host = "127.0.0.1"
+    calls = {"n": 0}
+
+    async def is_disconnected():
+        calls["n"] += 1
+        return calls["n"] > disconnected_after
+
+    request.is_disconnected = is_disconnected
+    return request
+
+
+@pytest.mark.asyncio
+class TestGenerateEvents:
+    async def test_first_yield_is_retry_directive(self):
+        cache = PriceCache()
+        request = _make_request()
+        gen = _generate_events(cache, request, interval=0.01)
+        first = await gen.__anext__()
+        assert first == "retry: 1000\n\n"
+
+    async def test_sends_seeded_ticker_data(self):
+        cache = PriceCache()
+        cache.update("AAPL", 190.50)
+        request = _make_request(disconnected_after=1)
+
+        events = [event async for event in _generate_events(cache, request, interval=0.01)]
+
+        data_events = [e for e in events if e.startswith("data: ")]
+        assert len(data_events) == 1
+        payload = json.loads(data_events[0][len("data: "):].strip())
+        assert "AAPL" in payload
+        assert payload["AAPL"]["price"] == 190.50
+
+    async def test_no_data_event_when_cache_empty(self):
+        cache = PriceCache()
+        request = _make_request(disconnected_after=1)
+
+        events = [event async for event in _generate_events(cache, request, interval=0.01)]
+
+        assert [e for e in events if e.startswith("data: ")] == []
+
+    async def test_skips_unchanged_version(self):
+        """No new data event is sent while the cache version stays the same."""
+        cache = PriceCache()
+        cache.update("AAPL", 190.50)
+        request = _make_request(disconnected_after=3)
+
+        events = [event async for event 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_stops_on_disconnect(self):
+        cache = PriceCache()
+        request = _make_request(disconnected_after=0)
+
+        events = [event async for event in _generate_events(cache, request, interval=0.01)]
+
+        assert events == ["retry: 1000\n\n"]
+
+
+class TestCreateStreamRouter:
+    def test_returns_a_new_router_each_call(self):
+        """Regression test: the router must not be a shared module-level global."""
+        cache = PriceCache()
+        router1 = create_stream_router(cache)
+        router2 = create_stream_router(cache)
+        assert router1 is not router2
+
+    def test_each_router_has_exactly_one_route(self):
+        """Calling the factory twice must not accumulate duplicate routes."""
+        cache = PriceCache()
+        router1 = create_stream_router(cache)
+        router2 = create_stream_router(cache)
+        assert len(router1.routes) == 1
+        assert len(router2.routes) == 1
+
+    def test_route_path(self):
+        cache = PriceCache()
+        router = create_stream_router(cache)
+        paths = [route.path for route in router.routes]
+        assert "/api/stream/prices" in paths