diff --git a/.github/workflows/claude-code-review.yml b/.github/workflows/claude-code-review.yml index b5e8cfd4d..6822593dd 100644 --- a/.github/workflows/claude-code-review.yml +++ b/.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 diff --git a/.github/workflows/claude.yml b/.github/workflows/claude.yml index d300267f1..69313a0e8 100644 --- a/.github/workflows/claude.yml +++ b/.github/workflows/claude.yml @@ -19,14 +19,14 @@ 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 @@ -34,6 +34,8 @@ jobs: 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' diff --git a/backend/app/market/cache.py b/backend/app/market/cache.py index 4d0215778..03370e717 100644 --- a/backend/app/market/cache.py +++ b/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 diff --git a/backend/app/market/interface.py b/backend/app/market/interface.py index 0f3b7d8c9..307b1707c 100644 --- a/backend/app/market/interface.py +++ b/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 diff --git a/backend/app/market/massive_client.py b/backend/app/market/massive_client.py index 00bc7b2aa..1de062d90 100644 --- a/backend/app/market/massive_client.py +++ b/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, ) diff --git a/backend/app/market/simulator.py b/backend/app/market/simulator.py index b6803f592..ae66a1265 100644 --- a/backend/app/market/simulator.py +++ b/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) diff --git a/backend/app/market/stream.py b/backend/app/market/stream.py index 7fd974b7c..4942fdb79 100644 --- a/backend/app/market/stream.py +++ b/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: diff --git a/backend/tests/market/test_cache.py b/backend/tests/market/test_cache.py index b5ab3d55d..830427fac 100644 --- a/backend/tests/market/test_cache.py +++ b/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 diff --git a/backend/tests/market/test_interface.py b/backend/tests/market/test_interface.py new file mode 100644 index 000000000..ab61fa7d0 --- /dev/null +++ b/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)) diff --git a/backend/tests/market/test_massive.py b/backend/tests/market/test_massive.py index cdd7dbd24..a8d1b7631 100644 --- a/backend/tests/market/test_massive.py +++ b/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): diff --git a/backend/tests/market/test_models.py b/backend/tests/market/test_models.py index 21600dfd6..1e0d3042d 100644 --- a/backend/tests/market/test_models.py +++ b/backend/tests/market/test_models.py @@ -10,7 +10,9 @@ 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 @@ -18,47 +20,65 @@ def test_price_update_creation(self): 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 diff --git a/backend/tests/market/test_seed_prices.py b/backend/tests/market/test_seed_prices.py new file mode 100644 index 000000000..9ce7a5a0e --- /dev/null +++ b/backend/tests/market/test_seed_prices.py @@ -0,0 +1,68 @@ +"""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 diff --git a/backend/tests/market/test_simulator.py b/backend/tests/market/test_simulator.py index 1845ec16b..3e7d37674 100644 --- a/backend/tests/market/test_simulator.py +++ b/backend/tests/market/test_simulator.py @@ -13,6 +13,22 @@ def test_step_returns_all_tickers(self): result = sim.step() assert set(result.keys()) == {"AAPL", "GOOGL"} + def test_full_default_watchlist_steps_successfully(self): + """The production correlation structure (tech intra=0.6, finance + intra=0.5, TSLA=0.3, cross=0.3) is a non-trivial block matrix, not + simple equicorrelation. Confirm Cholesky decomposition succeeds and + step() runs cleanly for the real 10-ticker default watchlist rather + than only the 1-2 ticker cases exercised elsewhere in this file.""" + tickers = list(SEED_PRICES.keys()) + sim = GBMSimulator(tickers=tickers) + assert sim._cholesky is not None + + for _ in range(100): + result = sim.step() + assert set(result.keys()) == set(tickers) + for price in result.values(): + assert price > 0 + def test_prices_are_positive(self): """GBM prices can never go negative (exp() is always positive).""" sim = GBMSimulator(tickers=["AAPL"]) @@ -126,6 +142,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 diff --git a/backend/tests/market/test_simulator_source.py b/backend/tests/market/test_simulator_source.py index 515ce7290..564449476 100644 --- a/backend/tests/market/test_simulator_source.py +++ b/backend/tests/market/test_simulator_source.py @@ -1,11 +1,12 @@ """Integration tests for SimulatorDataSource.""" import asyncio +from unittest.mock import patch import pytest from app.market.cache import PriceCache -from app.market.simulator import SimulatorDataSource +from app.market.simulator import GBMSimulator, SimulatorDataSource @pytest.mark.asyncio @@ -94,19 +95,37 @@ async def test_empty_start(self): await source.stop() async def test_exception_resilience(self): - """Test that simulator continues running after errors.""" + """Test that a single bad tick doesn't kill the background loop. + + Patches GBMSimulator.step to raise once, then asserts the loop + survives the exception (caught by the `except Exception` in + _run_loop) and resumes writing to the cache on the next tick. + """ cache = PriceCache() source = SimulatorDataSource(price_cache=cache, update_interval=0.05) - - # Start with a valid ticker await source.start(["AAPL"]) - # Wait for some updates - await asyncio.sleep(0.15) + version_before_failure = cache.version + real_step = GBMSimulator.step + call_count = 0 + + def flaky_step(self): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise RuntimeError("simulated tick failure") + return real_step(self) + + with patch.object(GBMSimulator, "step", flaky_step): + # Wait long enough for the failing tick plus at least one more. + await asyncio.sleep(0.2) - # Task should still be running + # Task survived the exception rather than dying. assert source._task is not None assert not source._task.done() + # And the cache kept receiving updates after the failure. + assert cache.version > version_before_failure + assert call_count >= 2 await source.stop() @@ -124,13 +143,47 @@ 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() # 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 diff --git a/backend/tests/market/test_stream.py b/backend/tests/market/test_stream.py new file mode 100644 index 000000000..266c11e6a --- /dev/null +++ b/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 diff --git a/planning/MARKET_DATA_DESIGN.md b/planning/MARKET_DATA_DESIGN.md new file mode 100644 index 000000000..a1281b90a --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,1303 @@ +# Market Data Backend — Detailed Design + +Implementation-ready design for the FinAlly market data subsystem: the unified provider interface, the in-memory price cache, the GBM simulator, the Massive (Polygon.io) API client, the SSE streaming endpoint, and how the rest of the backend wires into all of it. + +**Status:** This subsystem is implemented and tested (see `planning/MARKET_DATA_SUMMARY.md` for the build/test summary). This document is the reference design — the code snippets below are the actual, current implementation in `backend/app/market/`, not a proposal. Section 10 (FastAPI lifecycle integration) is forward-looking: `backend/app/main.py` does not exist yet, so that section specifies how the not-yet-built app entrypoint and other routers (portfolio, watchlist) should wire into this subsystem. + +--- + +## Table of Contents + +1. [Design Goals](#1-design-goals) +2. [File Structure](#2-file-structure) +3. [Data Model — `models.py`](#3-data-model) +4. [Abstract Interface — `interface.py`](#4-abstract-interface) +5. [Price Cache — `cache.py`](#5-price-cache) +6. [Seed Prices & Ticker Parameters — `seed_prices.py`](#6-seed-prices--ticker-parameters) +7. [GBM Simulator — `simulator.py`](#7-gbm-simulator) +8. [Massive API Client — `massive_client.py`](#8-massive-api-client) +9. [Factory — `factory.py`](#9-factory) +10. [SSE Streaming Endpoint — `stream.py`](#10-sse-streaming-endpoint) +11. [FastAPI Lifecycle Integration (forward-looking)](#11-fastapi-lifecycle-integration-forward-looking) +12. [Watchlist Coordination](#12-watchlist-coordination) +13. [Error Handling & Edge Cases](#13-error-handling--edge-cases) +14. [Testing Strategy](#14-testing-strategy) +15. [Configuration Summary](#15-configuration-summary) + +--- + +## 1. Design Goals + +- **Source-agnostic downstream code.** SSE streaming, portfolio valuation, and trade execution never know whether prices come from the simulator or Massive. They only see `PriceCache` and `PriceUpdate`. +- **Push, not pull.** Data sources write into a shared cache on their own schedule (500ms for the simulator, 15s for Massive free tier). Consumers read the cache at their own cadence. This decouples producer timing from consumer timing. +- **Zero external dependencies for the default path.** With no `MASSIVE_API_KEY`, the simulator runs entirely in-process — no network calls, no API key required, works offline. +- **Graceful degradation.** A bad Massive poll, an invalid API key, or a malformed snapshot never crashes the background task or blanks out prices — the cache simply retains the last good value. +- **Cheap to extend.** Adding a new ticker or a new data source (e.g., a websocket-based provider later) should not require touching the SSE endpoint, portfolio code, or the cache. + +--- + +## 2. File Structure + +``` +backend/ + app/ + market/ + __init__.py # Re-exports: PriceUpdate, PriceCache, MarketDataSource, + # create_market_data_source, create_stream_router + models.py # PriceUpdate dataclass + interface.py # MarketDataSource ABC + cache.py # PriceCache (thread-safe in-memory store) + seed_prices.py # SEED_PRICES, TICKER_PARAMS, DEFAULT_PARAMS, CORRELATION_GROUPS + simulator.py # GBMSimulator + SimulatorDataSource + massive_client.py # MassiveDataSource + factory.py # create_market_data_source() + stream.py # SSE endpoint (FastAPI router factory) + tests/ + market/ + test_models.py + test_cache.py + test_simulator.py + test_simulator_source.py + test_factory.py + test_massive.py +``` + +Each module has a single responsibility. `app/market/__init__.py` re-exports the public API so the rest of the backend imports from `app.market` without reaching into submodules: + +```python +from app.market import PriceCache, PriceUpdate, MarketDataSource, create_market_data_source, create_stream_router +``` + +--- + +## 3. Data Model + +**File: `backend/app/market/models.py`** + +`PriceUpdate` is the only data structure that leaves the market data layer. Every downstream consumer — SSE streaming, portfolio valuation, trade execution — works exclusively with this type. + +```python +"""Data models for market data.""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass(frozen=True, slots=True) +class PriceUpdate: + """Immutable snapshot of a single ticker's price at a point in time.""" + + ticker: str + price: float + previous_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: + """Absolute price change from previous update.""" + return round(self.price - self.previous_price, 4) + + @property + def change_percent(self) -> float: + """Percentage change from previous update.""" + if self.previous_price == 0: + return 0.0 + return round((self.price - self.previous_price) / self.previous_price * 100, 4) + + @property + def direction(self) -> str: + """'up', 'down', or 'flat'.""" + if self.price > self.previous_price: + return "up" + elif self.price < self.previous_price: + return "down" + return "flat" + + def to_dict(self) -> dict: + """Serialize for JSON / SSE transmission.""" + return { + "ticker": self.ticker, + "price": self.price, + "previous_price": self.previous_price, + "timestamp": self.timestamp, + "change": self.change, + "change_percent": self.change_percent, + "direction": self.direction, + } +``` + +### Design decisions + +- **`frozen=True`** — Price updates are immutable value objects, safe to share across async tasks without copying. +- **`slots=True`** — Memory optimization; the system creates many of these per second. +- **Computed properties** (`change`, `change_percent`, `direction`) — Derived from `price` and `previous_price`, so they can never fall out of sync. There is no stored `direction` field that could go stale. +- **`to_dict()`** — Single serialization point, used by both the SSE endpoint and (eventually) REST responses like `/api/watchlist` and `/api/portfolio`. + +--- + +## 4. Abstract Interface + +**File: `backend/app/market/interface.py`** + +```python +"""Abstract interface for market data sources.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + + +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. + + Lifecycle: + source = create_market_data_source(cache) + await source.start(["AAPL", "GOOGL", ...]) + # ... app runs ... + await source.add_ticker("TSLA") + await source.remove_ticker("GOOGL") + # ... app shutting down ... + await source.stop() + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing price updates for the given tickers. + + Starts a background task that periodically writes to the PriceCache. + Must be called exactly once. Calling start() twice is undefined behavior. + """ + + @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. + """ + + @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. + """ + + @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. + """ + + @abstractmethod + def get_tickers(self) -> list[str]: + """Return the current list of actively tracked tickers.""" +``` + +### Why the source writes to the cache instead of returning prices + +This push model decouples timing. The simulator ticks every 500ms; Massive polls every 15s (free tier). SSE always reads from the cache at its own 500ms cadence regardless of which source is active — it never needs to know the source's update interval, and adding a third data source later requires no change to the SSE layer. + +--- + +## 5. Price Cache + +**File: `backend/app/market/cache.py`** + +The price cache is the central hub: data sources write to it; SSE streaming, portfolio valuation, and trade execution read from it. It must be thread-safe because a data source's background work may run in a thread pool executor (`asyncio.to_thread`, used by the Massive client) while SSE reads happen on the async event loop. + +```python +"""Thread-safe in-memory price cache.""" + +from __future__ import annotations + +import time +from threading import Lock + +from .models import PriceUpdate + + +class PriceCache: + """Thread-safe in-memory cache of the latest price for each ticker. + + Writers: SimulatorDataSource or MassiveDataSource (one at a time). + Readers: SSE streaming endpoint, portfolio valuation, trade execution. + """ + + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._lock = Lock() + self._version: int = 0 # Monotonically increasing; bumped on every update + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + """Record a new price for a ticker. Returns the created PriceUpdate. + + Automatically computes direction and change from the previous price. + If this is the first update for the ticker, previous_price == price (direction='flat'). + """ + with self._lock: + ts = timestamp or time.time() + prev = self._prices.get(ticker) + previous_price = prev.price if prev else price + + update = PriceUpdate( + ticker=ticker, + price=round(price, 2), + previous_price=round(previous_price, 2), + timestamp=ts, + ) + self._prices[ticker] = update + self._version += 1 + return update + + 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_all(self) -> dict[str, PriceUpdate]: + """Snapshot of all current prices. Returns a shallow copy.""" + 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 remove(self, ticker: str) -> None: + """Remove a ticker from the cache (e.g., when removed from watchlist).""" + with self._lock: + self._prices.pop(ticker, None) + + @property + def version(self) -> int: + """Current version counter. Useful for SSE change detection.""" + return self._version + + def __len__(self) -> int: + with self._lock: + return len(self._prices) + + def __contains__(self, ticker: str) -> bool: + with self._lock: + return ticker in self._prices +``` + +### Why a version counter? + +The SSE loop polls the cache every ~500ms. Without a version counter it would serialize and re-send every price on every tick, even when nothing changed (e.g., Massive only updates every 15s). The counter lets the SSE loop skip a send when nothing is new: + +```python +last_version = -1 +while True: + if price_cache.version != last_version: + last_version = price_cache.version + yield format_sse(price_cache.get_all()) + await asyncio.sleep(0.5) +``` + +### Thread safety rationale + +`threading.Lock` is used instead of `asyncio.Lock` because: +- The Massive client's synchronous `get_snapshot_all()` call runs inside `asyncio.to_thread()`, which executes in a real OS thread — `asyncio.Lock` would not protect against concurrent access from that thread. +- `threading.Lock` works correctly whether the caller is a sync thread or the async event loop. +- The critical sections are tiny (a dict read or write plus an int increment), so contention is negligible even at the target scale (≤ dozens of tickers, one writer, a handful of SSE readers). + +--- + +## 6. Seed Prices & Ticker Parameters + +**File: `backend/app/market/seed_prices.py`** + +Constants only — no logic, no imports beyond stdlib types. Shared by the simulator (initial prices, GBM parameters, correlation structure) and available as a fallback if the Massive client needs seed values before its first successful poll. + +```python +"""Seed prices and per-ticker parameters for the market simulator.""" + +# Realistic starting prices for the default watchlist (as of project creation) +SEED_PRICES: dict[str, float] = { + "AAPL": 190.00, + "GOOGL": 175.00, + "MSFT": 420.00, + "AMZN": 185.00, + "TSLA": 250.00, + "NVDA": 800.00, + "META": 500.00, + "JPM": 195.00, + "V": 280.00, + "NFLX": 600.00, +} + +# Per-ticker GBM parameters +# sigma: annualized volatility (higher = more price movement) +# mu: annualized drift / expected return +TICKER_PARAMS: dict[str, dict[str, float]] = { + "AAPL": {"sigma": 0.22, "mu": 0.05}, + "GOOGL": {"sigma": 0.25, "mu": 0.05}, + "MSFT": {"sigma": 0.20, "mu": 0.05}, + "AMZN": {"sigma": 0.28, "mu": 0.05}, + "TSLA": {"sigma": 0.50, "mu": 0.03}, # High volatility + "NVDA": {"sigma": 0.40, "mu": 0.08}, # High volatility, strong drift + "META": {"sigma": 0.30, "mu": 0.05}, + "JPM": {"sigma": 0.18, "mu": 0.04}, # Low volatility (bank) + "V": {"sigma": 0.17, "mu": 0.04}, # Low volatility (payments) + "NFLX": {"sigma": 0.35, "mu": 0.05}, +} + +# Default parameters for tickers not in the list above (dynamically added) +DEFAULT_PARAMS: dict[str, float] = {"sigma": 0.25, "mu": 0.05} + +# Correlation groups for the simulator's Cholesky decomposition +# Tickers in the same group have higher intra-group correlation +CORRELATION_GROUPS: dict[str, set[str]] = { + "tech": {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"}, + "finance": {"JPM", "V"}, +} + +# Correlation coefficients +INTRA_TECH_CORR = 0.6 # Tech stocks move together +INTRA_FINANCE_CORR = 0.5 # Finance stocks move together +CROSS_GROUP_CORR = 0.3 # Between sectors / unknown tickers +TSLA_CORR = 0.3 # TSLA does its own thing +``` + +Tickers not in `SEED_PRICES` (added dynamically via the watchlist or LLM chat) start at a random price between $50–$300 and use `DEFAULT_PARAMS`. + +--- + +## 7. GBM Simulator + +**File: `backend/app/market/simulator.py`** + +Two classes live here: +- **`GBMSimulator`** — pure math engine, stateful, holds current prices and advances them one step at a time. +- **`SimulatorDataSource`** — the `MarketDataSource` implementation that wraps `GBMSimulator` in an async loop and writes results to the `PriceCache`. + +### 7.1 The math + +Geometric Brownian Motion is the standard model behind Black-Scholes: prices evolve continuously with random noise, never go negative, and follow a lognormal distribution — the same statistical shape observed in real markets. + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +Where `S(t)` is the current price, `mu` is annualized drift, `sigma` is annualized volatility, `dt` is the time step as a fraction of a trading year, and `Z` is a (correlated) standard normal draw. + +For 500ms updates over a 252-day, 6.5-hour trading year: + +``` +dt = 0.5 / (252 * 6.5 * 3600) ≈ 8.48e-8 +``` + +This tiny `dt` produces small, realistic per-tick moves that accumulate naturally into believable intraday ranges. + +### 7.2 Correlated moves via Cholesky decomposition + +Real stocks don't move independently. Given a correlation matrix `C`, the Cholesky factor `L = cholesky(C)` transforms independent standard normals into correlated ones: `Z_correlated = L @ Z_independent`. The correlation structure used here: + +| Pairing | Correlation | +|---|---| +| Two tech tickers | 0.6 | +| Two finance tickers | 0.5 | +| Anything involving TSLA | 0.3 (it moves on its own) | +| Cross-sector or unknown tickers | 0.3 | + +### 7.3 Random shock events + +Every step, each ticker has a small (~0.1%) chance of a sudden 2–5% move, for visual drama on the dashboard. At 10 tickers and 2 ticks/sec, expect roughly one event every ~50 seconds. + +### 7.4 Implementation + +```python +"""GBM-based market simulator.""" + +from __future__ import annotations + +import asyncio +import logging +import math +import random + +import numpy as np + +from .cache import PriceCache +from .interface import MarketDataSource +from .seed_prices import ( + CORRELATION_GROUPS, + CROSS_GROUP_CORR, + DEFAULT_PARAMS, + INTRA_FINANCE_CORR, + INTRA_TECH_CORR, + SEED_PRICES, + TICKER_PARAMS, + TSLA_CORR, +) + +logger = logging.getLogger(__name__) + + +class GBMSimulator: + """Geometric Brownian Motion simulator for correlated stock prices. + + Math: + S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) + + The tiny dt (~8.5e-8 for 500ms ticks over 252 trading days * 6.5h/day) + produces sub-cent moves per tick that accumulate naturally over time. + """ + + # 252 trading days * 6.5 hours/day * 3600 seconds/hour = 5,896,800 seconds + TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 + DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8 + + def __init__( + self, + tickers: list[str], + dt: float = DEFAULT_DT, + event_probability: float = 0.001, + ) -> None: + self._dt = dt + self._event_prob = event_probability + + self._tickers: list[str] = [] + self._prices: dict[str, float] = {} + self._params: dict[str, dict[str, float]] = {} + self._cholesky: np.ndarray | None = None + + for ticker in tickers: + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + # --- Public API --- + + def step(self) -> dict[str, float]: + """Advance all tickers by one time step. Returns {ticker: new_price}. + + This is the hot path — called every 500ms. Keep it fast. + """ + n = len(self._tickers) + if n == 0: + return {} + + z_independent = np.random.standard_normal(n) + z_correlated = self._cholesky @ z_independent if self._cholesky is not None else z_independent + + result: dict[str, float] = {} + for i, ticker in enumerate(self._tickers): + params = self._params[ticker] + mu, sigma = params["mu"], params["sigma"] + + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z_correlated[i] + self._prices[ticker] *= math.exp(drift + diffusion) + + # Random event: ~0.1% chance per tick per ticker + if random.random() < self._event_prob: + shock_magnitude = random.uniform(0.02, 0.05) + shock_sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock_magnitude * shock_sign + logger.debug( + "Random event on %s: %.1f%% %s", + ticker, shock_magnitude * 100, "up" if shock_sign > 0 else "down", + ) + + result[ticker] = round(self._prices[ticker], 2) + + return result + + def add_ticker(self, ticker: str) -> None: + """Add a ticker to the simulation. Rebuilds the correlation matrix.""" + if ticker in self._prices: + return + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + def remove_ticker(self, ticker: str) -> None: + """Remove a ticker from the simulation. Rebuilds the correlation matrix.""" + if ticker not in self._prices: + return + self._tickers.remove(ticker) + del self._prices[ticker] + del self._params[ticker] + self._rebuild_cholesky() + + def get_price(self, ticker: str) -> float | None: + """Current price for a ticker, or None if not tracked.""" + return self._prices.get(ticker) + + def get_tickers(self) -> list[str]: + """Return the list of currently tracked tickers.""" + return list(self._tickers) + + # --- Internals --- + + def _add_ticker_internal(self, ticker: str) -> None: + """Add a ticker without rebuilding Cholesky (for batch initialization).""" + if ticker in self._prices: + return + self._tickers.append(ticker) + self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) + self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) + + def _rebuild_cholesky(self) -> None: + """Rebuild the Cholesky decomposition of the ticker correlation matrix. + + Called whenever tickers are added or removed. O(n^2) but n < 50. + """ + n = len(self._tickers) + if n <= 1: + self._cholesky = None + return + + corr = np.eye(n) + for i in range(n): + for j in range(i + 1, n): + rho = self._pairwise_correlation(self._tickers[i], self._tickers[j]) + corr[i, j] = rho + corr[j, i] = rho + + self._cholesky = np.linalg.cholesky(corr) + + @staticmethod + def _pairwise_correlation(t1: str, t2: str) -> float: + """Determine correlation between two tickers based on sector grouping.""" + tech = CORRELATION_GROUPS["tech"] + finance = CORRELATION_GROUPS["finance"] + + if t1 == "TSLA" or t2 == "TSLA": + return TSLA_CORR + if t1 in tech and t2 in tech: + return INTRA_TECH_CORR + if t1 in finance and t2 in finance: + return INTRA_FINANCE_CORR + return CROSS_GROUP_CORR + + +class SimulatorDataSource(MarketDataSource): + """MarketDataSource backed by the GBM simulator. + + Runs a background asyncio task that calls GBMSimulator.step() every + `update_interval` seconds and writes results to the PriceCache. + """ + + def __init__( + self, + price_cache: PriceCache, + update_interval: float = 0.5, + event_probability: float = 0.001, + ) -> None: + self._cache = price_cache + self._interval = update_interval + self._event_prob = event_probability + self._sim: GBMSimulator | None = None + self._task: asyncio.Task | None = None + + async def start(self, tickers: list[str]) -> None: + self._sim = GBMSimulator(tickers=tickers, event_probability=self._event_prob) + # Seed the cache with initial prices so SSE has data immediately + for ticker in tickers: + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + logger.info("Simulator started with %d tickers", len(tickers)) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("Simulator stopped") + + async def add_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.add_ticker(ticker) + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + logger.info("Simulator: added ticker %s", ticker) + + async def remove_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.remove_ticker(ticker) + self._cache.remove(ticker) + logger.info("Simulator: removed ticker %s", ticker) + + def get_tickers(self) -> list[str]: + return self._sim.get_tickers() if self._sim else [] + + async def _run_loop(self) -> None: + """Core loop: step the simulation, write to cache, sleep.""" + while True: + try: + if self._sim: + prices = self._sim.step() + for ticker, price in prices.items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +### Key behaviors + +- **Immediate seeding.** `start()` populates the cache with seed prices before the loop begins, so the SSE endpoint has data to send on its very first tick — no blank-screen delay on page load. +- **Graceful cancellation.** `stop()` cancels the background task and awaits it, catching `CancelledError`, for clean shutdown during FastAPI lifespan teardown. +- **Exception resilience.** The loop catches exceptions per-step so a single bad tick (e.g., a numerical edge case) never kills the entire feed. + +--- + +## 8. Massive API Client + +**File: `backend/app/market/massive_client.py`** + +Polls the Massive (formerly Polygon.io) REST snapshot endpoint on a configurable interval. The synchronous `massive` SDK client runs inside `asyncio.to_thread()` so it never blocks the event loop. + +```python +"""Massive (Polygon.io) API client for real market data.""" + +from __future__ import annotations + +import asyncio +import logging + +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +from .cache import PriceCache +from .interface import MarketDataSource + +logger = logging.getLogger(__name__) + + +class MassiveDataSource(MarketDataSource): + """MarketDataSource backed by the Massive (Polygon.io) REST API. + + Polls GET /v2/snapshot/locale/us/markets/stocks/tickers for all watched + tickers in a single API call, then writes results to the PriceCache. + + Rate limits: + - Free tier: 5 req/min → poll every 15s (default) + - Paid tiers: higher limits → poll every 2-5s + """ + + def __init__( + self, + api_key: str, + price_cache: PriceCache, + poll_interval: float = 15.0, + ) -> None: + self._api_key = api_key + self._cache = price_cache + self._interval = poll_interval + self._tickers: list[str] = [] + self._task: asyncio.Task | None = None + self._client: RESTClient | None = None + + async def start(self, tickers: list[str]) -> None: + self._client = RESTClient(api_key=self._api_key) + self._tickers = list(tickers) + + # Do an immediate first poll so the cache has data right away + await self._poll_once() + + self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") + logger.info( + "Massive poller started: %d tickers, %.1fs interval", + len(tickers), self._interval, + ) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + self._client = None + logger.info("Massive poller stopped") + + async def add_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + if ticker not in self._tickers: + self._tickers.append(ticker) + logger.info("Massive: added ticker %s (will appear on next poll)", ticker) + + async def remove_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + self._tickers = [t for t in self._tickers if t != ticker] + self._cache.remove(ticker) + logger.info("Massive: removed ticker %s", ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + # --- Internal --- + + async def _poll_loop(self) -> None: + """Poll on interval. First poll already happened in start().""" + while True: + await asyncio.sleep(self._interval) + await self._poll_once() + + async def _poll_once(self) -> None: + """Execute one poll cycle: fetch snapshots, update cache.""" + if not self._tickers or not self._client: + return + + try: + # The Massive RESTClient is synchronous — run in a thread to + # avoid blocking the event loop. + snapshots = await asyncio.to_thread(self._fetch_snapshots) + processed = 0 + 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 + self._cache.update(ticker=snap.ticker, price=price, timestamp=timestamp) + processed += 1 + except (AttributeError, TypeError) as e: + logger.warning( + "Skipping snapshot for %s: %s", getattr(snap, "ticker", "???"), e, + ) + 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 retries on the next interval. + # 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.""" + return self._client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=self._tickers, + ) +``` + +### Massive API reference (as used here) + +**Client init** — `RESTClient(api_key=...)`, or `RESTClient()` to read `MASSIVE_API_KEY` from the environment automatically. + +**Primary endpoint — snapshot, all tickers in one call:** + +```python +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +client = RESTClient() +snapshots = client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"], +) +for snap in snapshots: + print(f"{snap.ticker}: ${snap.last_trade.price}") +``` + +Getting all watched tickers in a single call is what keeps the free tier (5 req/min) viable — polling ticker-by-ticker would blow the budget instantly with a 10-ticker watchlist. + +Key fields extracted per snapshot: `snap.ticker`, `snap.last_trade.price`, `snap.last_trade.timestamp` (Unix **milliseconds**, converted to seconds before writing to the cache). + +**Rate limits:** + +| Tier | Limit | FinAlly poll interval | +|---|---|---| +| Free | 5 req/min | 15s | +| Paid | much higher | 2–5s | + +### Error handling philosophy + +The poller is intentionally resilient — a live trading terminal should never crash because of a flaky upstream API: + +| Error | Behavior | +|---|---| +| 401 Unauthorized (bad key) | Logged as error; poller keeps running so a corrected `.env` + restart recovers cleanly. | +| 429 Rate limited | Logged as error; next poll retries after `poll_interval`. | +| Network timeout | Logged as error; retried automatically on the next cycle. | +| Malformed snapshot for one ticker | That ticker is skipped with a warning; other tickers in the same batch are still processed. | +| All tickers fail | Cache retains last-known prices — SSE keeps streaming stale-but-present data rather than going blank. | + +### Lazy dependency, not lazy import + +Unlike an earlier draft of this design, `massive_client.py` imports `RESTClient` and `SnapshotMarketType` at module level. `massive>=1.0.0` is declared as a core dependency in `pyproject.toml`, so it is always installed — `uv sync` pulls it in regardless of whether `MASSIVE_API_KEY` is set. The **optionality is behavioral, not import-time**: `factory.py` (below) only *constructs* a `MassiveDataSource` when the key is present, so the client is never instantiated — and never makes a network call — in simulator mode. + +--- + +## 9. Factory + +**File: `backend/app/market/factory.py`** + +```python +"""Factory for creating market data sources.""" + +from __future__ import annotations + +import logging +import os + +from .cache import PriceCache +from .interface import MarketDataSource +from .massive_client import MassiveDataSource +from .simulator import SimulatorDataSource + +logger = logging.getLogger(__name__) + + +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """Create the appropriate market data source based on environment variables. + + - MASSIVE_API_KEY set and non-empty → MassiveDataSource (real market data) + - Otherwise → SimulatorDataSource (GBM simulation) + + Returns an unstarted source. Caller must await source.start(tickers). + """ + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + + if api_key: + logger.info("Market data source: Massive API (real data)") + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + else: + logger.info("Market data source: GBM Simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +### Usage at app startup + +```python +price_cache = PriceCache() +source = create_market_data_source(price_cache) +await source.start(initial_tickers) # e.g., ["AAPL", "GOOGL", ...] +``` + +This single function is the only place in the codebase that branches on `MASSIVE_API_KEY`. Everything downstream — SSE, portfolio valuation, watchlist routes, the LLM chat tool that executes trades — only ever sees a `MarketDataSource` and a `PriceCache`. + +--- + +## 10. SSE Streaming Endpoint + +**File: `backend/app/market/stream.py`** + +A FastAPI route that holds open a long-lived HTTP connection and pushes price updates to the browser as `text/event-stream`. + +```python +"""SSE streaming endpoint for live price updates.""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import AsyncGenerator + +from fastapi import APIRouter, Request +from fastapi.responses import StreamingResponse + +from .cache import PriceCache + +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. + """ + + @router.get("/prices") + async def stream_prices(request: Request) -> StreamingResponse: + """SSE endpoint for live price updates. + + Streams all tracked ticker prices every ~500ms. The client connects + with EventSource and receives events in the format: + + data: {"AAPL": {"ticker": "AAPL", "price": 190.50, ...}, ...} + + Includes a retry directive so the browser auto-reconnects on + disconnection (EventSource built-in behavior). + """ + return StreamingResponse( + _generate_events(price_cache, request), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", # Disable nginx buffering if proxied + }, + ) + + return router + + +async def _generate_events( + price_cache: PriceCache, + request: Request, + interval: float = 0.5, +) -> AsyncGenerator[str, None]: + """Async generator that yields SSE-formatted price events. + + Sends all prices every `interval` seconds. Stops when the client + disconnects (detected via request.is_disconnected()). + """ + yield "retry: 1000\n\n" # Tell the client to retry after 1s if the connection drops + + last_version = -1 + client_ip = request.client.host if request.client else "unknown" + logger.info("SSE client connected: %s", client_ip) + + try: + while True: + if await request.is_disconnected(): + logger.info("SSE client disconnected: %s", client_ip) + break + + current_version = price_cache.version + if current_version != last_version: + last_version = current_version + prices = price_cache.get_all() + + if prices: + data = {ticker: update.to_dict() for ticker, update in prices.items()} + yield f"data: {json.dumps(data)}\n\n" + + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("SSE stream cancelled for: %s", client_ip) +``` + +### Wire format + +``` +data: {"AAPL":{"ticker":"AAPL","price":190.50,"previous_price":190.42,"timestamp":1707580800.5,"change":0.08,"change_percent":0.042,"direction":"up"},"GOOGL":{"ticker":"GOOGL","price":175.12,...}} + +``` + +Frontend consumption (per PLAN.md §10, `EventSource` is the required client API): + +```javascript +const eventSource = new EventSource('/api/stream/prices'); +eventSource.onmessage = (event) => { + const prices = JSON.parse(event.data); + // prices: { "AAPL": { ticker, price, previous_price, change, change_percent, direction, timestamp }, ... } + // Use `direction` to trigger the green/red flash animation, + // and accumulate each ticker's `price` client-side to build sparklines. +}; +``` + +### Why poll-and-push instead of event-driven? + +The endpoint polls the cache on a fixed interval rather than being notified by the data source. This is simpler, and — more importantly — it produces evenly-spaced updates regardless of source cadence, which matters because the frontend accumulates these into sparkline charts; even spacing keeps that visualization clean, whether the underlying source is a 500ms simulator tick or a 15s Massive poll (in the latter case, most 500ms ticks are no-ops thanks to the version check, and the sparkline simply has fewer, evenly-spaced points). + +--- + +## 11. FastAPI Lifecycle Integration (forward-looking) + +`backend/app/main.py` has not been written yet — the rest of the platform (portfolio, watchlist, chat, database) is still to be built per `PLAN.md`. This section specifies how that entrypoint should start and stop the market data subsystem using FastAPI's `lifespan` context manager, and how other routers should access it. + +```python +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from app.market import PriceCache, MarketDataSource, create_market_data_source, create_stream_router + + +@asynccontextmanager +async def lifespan(app: FastAPI): + """Manage startup and shutdown of background services.""" + + # --- STARTUP --- + + # 1. Create the shared price cache + price_cache = PriceCache() + app.state.price_cache = price_cache + + # 2. Create the market data source (reads MASSIVE_API_KEY) + source = create_market_data_source(price_cache) + app.state.market_source = source + + # 3. Load initial tickers from the database watchlist (lazy-init DB if needed) + initial_tickers = await load_watchlist_tickers() # reads from SQLite `watchlist` table + await source.start(initial_tickers) + + # 4. Register the SSE streaming router + app.include_router(create_stream_router(price_cache)) + + yield # App is running + + # --- SHUTDOWN --- + await source.stop() + + +app = FastAPI(title="FinAlly", lifespan=lifespan) + + +# Dependencies for injecting shared state into route handlers +def get_price_cache() -> PriceCache: + return app.state.price_cache + + +def get_market_source() -> MarketDataSource: + return app.state.market_source +``` + +### Accessing market data from other routers + +Portfolio, trade execution, and watchlist routes access the cache and source via FastAPI dependency injection — never by importing a module-level singleton: + +```python +from fastapi import APIRouter, Depends, HTTPException + +router = APIRouter(prefix="/api") + + +@router.post("/portfolio/trade") +async def execute_trade( + trade: TradeRequest, + price_cache: PriceCache = Depends(get_price_cache), +): + current_price = price_cache.get_price(trade.ticker) + if current_price is None: + raise HTTPException(404, f"No price available for {trade.ticker}") + # ... validate cash/shares, insert into `trades` and `positions`, at current_price ... + + +@router.post("/watchlist") +async def add_to_watchlist( + payload: WatchlistAdd, + source: MarketDataSource = Depends(get_market_source), +): + # ... insert into `watchlist` table ... + await source.add_ticker(payload.ticker) + # ... return ticker + current price if already cached ... + + +@router.delete("/watchlist/{ticker}") +async def remove_from_watchlist( + ticker: str, + source: MarketDataSource = Depends(get_market_source), +): + # ... delete from `watchlist` table (see §12 for the open-position edge case) ... + await source.remove_ticker(ticker) +``` + +The LLM chat tool-execution path (structured `trades` / `watchlist_changes` from the model, per `PLAN.md` §9) should route through these same functions rather than duplicating trade/watchlist logic — auto-executed LLM actions and manually-triggered REST actions must share one code path so validation stays consistent. + +--- + +## 12. Watchlist Coordination + +When the watchlist changes — via the REST API or LLM chat — the market data source must be told, so it tracks the right set of tickers. + +### Flow: adding a ticker + +``` +User (or LLM) → POST /api/watchlist {ticker: "PYPL"} + → INSERT INTO watchlist (SQLite) + → await source.add_ticker("PYPL") + Simulator: adds to GBMSimulator, rebuilds Cholesky, seeds cache immediately + Massive: appends to ticker list, appears on the next poll (up to `poll_interval` delay) + → Response: ticker + current price (if already cached) +``` + +### Flow: removing a ticker + +``` +User (or LLM) → DELETE /api/watchlist/PYPL + → DELETE FROM watchlist (SQLite) + → await source.remove_ticker("PYPL") + Simulator: removes from GBMSimulator, rebuilds Cholesky, removes from cache + Massive: removes from ticker list, removes from cache + → Response: success +``` + +### Edge case: ticker has an open position + +If the user removes a ticker from the watchlist while still holding shares, the data source must keep tracking it — otherwise portfolio valuation and the positions table lose their price feed. The watchlist route must check for an open position before calling `remove_ticker()`: + +```python +@router.delete("/watchlist/{ticker}") +async def remove_from_watchlist( + ticker: str, + source: MarketDataSource = Depends(get_market_source), +): + await db.delete_watchlist_entry(ticker) + + position = await db.get_position(ticker) + if position is None or position.quantity == 0: + await source.remove_ticker(ticker) + # else: keep tracking — portfolio valuation still needs live prices for it + + return {"status": "ok"} +``` + +Symmetrically, buying a ticker that isn't currently on the watchlist (possible via LLM-initiated trades) should call `source.add_ticker()` even if no watchlist row is created, so the new position gets priced immediately. + +--- + +## 13. Error Handling & Edge Cases + +### 13.1 Startup with an empty watchlist + +If the database has no watchlist rows (e.g. the user deleted every ticker), `start([])` is called. Both sources handle this gracefully: the simulator produces no prices, the Massive poller skips its API call entirely (`if not self._tickers: return`). The SSE endpoint sends no events until a ticker is added, at which point tracking begins immediately. + +### 13.2 Price cache miss during trade execution + +A ticker can be requested for trading before it has a cached price (just added, Massive hasn't polled yet). The trade route must surface this as a clear client error rather than crashing or trading at a null price: + +```python +price = price_cache.get_price(ticker) +if price is None: + raise HTTPException( + status_code=400, + detail=f"Price not yet available for {ticker}. Please wait a moment and try again.", + ) +``` + +The simulator avoids this window entirely by seeding the cache synchronously inside `add_ticker()`. Massive has an inherent gap of up to `poll_interval`; the 400 response with a clear message is the correct behavior for that gap, not a bug to "fix" with blocking/waiting logic. + +### 13.3 Invalid Massive API key + +If the key is set but wrong, the first poll fails with 401. The poller logs the error and keeps retrying every `poll_interval` — it does not crash or exit. The SSE endpoint keeps streaming (connection stays "connected" in the UI's status indicator) but with empty or stale price data, since the cache never gets populated. The fix is operational: correct `.env` and restart the container. + +### 13.4 Thread safety under load + +`PriceCache`'s `threading.Lock` is a plain mutex — one thread holds it at a time. At the target scale (≤ dozens of tickers, one writer, a handful of concurrent SSE readers in this single-user app), contention is negligible; the critical section is a dict read/write plus an int increment. A `ReadWriteLock` would only matter at a scale (hundreds of tickers, many concurrent readers) this project never needs. + +### 13.5 Simulator numerical stability + +- Prices are rounded to 2 decimal places inside `GBMSimulator.step()`. +- The exponential formulation (`exp(drift + diffusion)`) is numerically stable and always yields a positive result — GBM prices cannot go negative or hit exactly zero. +- Tiny `dt` (~8.5e-8) keeps per-tick moves small; volatility accumulates correctly over many ticks rather than producing implausible single-tick jumps (outside of the deliberate random shock events). + +--- + +## 14. Testing Strategy + +Tests live in `backend/tests/market/`, one module per source file, following `PLAN.md` §12's backend testing guidance (pytest, `pytest-asyncio`). + +| Test module | What it covers | +|---|---| +| `test_models.py` | `PriceUpdate` computed properties (`change`, `change_percent`, `direction`) and `to_dict()` serialization, including edge cases like `previous_price == 0`. | +| `test_cache.py` | `update`/`get`/`get_all`/`get_price`/`remove`, first-update-is-flat behavior, version counter increments, `__len__`/`__contains__`. | +| `test_simulator.py` | GBM math properties: prices always positive, `step()` returns all tracked tickers, add/remove ticker rebuilds Cholesky correctly, duplicate add / missing remove are no-ops, unknown tickers get a random seed in `[50, 300]`, prices drift after many steps. | +| `test_simulator_source.py` | Integration: `start()` seeds the cache immediately (no blank window), prices actually change over time, `stop()` is idempotent (safe to call twice), `add_ticker`/`remove_ticker` propagate to both the simulator and the cache. | +| `test_factory.py` | `create_market_data_source` returns `SimulatorDataSource` when `MASSIVE_API_KEY` is unset/empty, `MassiveDataSource` when set — via `monkeypatch.setenv`/`delenv`, not real network calls. | +| `test_massive.py` | `_poll_once` updates the cache from mocked snapshot objects; a malformed snapshot for one ticker is skipped without affecting others; an exception from `_fetch_snapshots` is swallowed (poller doesn't crash) and leaves the cache untouched for that ticker. | + +Representative patterns: + +```python +# test_simulator.py +def test_prices_are_positive(): + """GBM prices can never go negative (exp() is always positive).""" + sim = GBMSimulator(tickers=["AAPL"]) + for _ in range(10_000): + prices = sim.step() + assert prices["AAPL"] > 0 + + +def test_cholesky_rebuilds_on_add(): + sim = GBMSimulator(tickers=["AAPL"]) + assert sim._cholesky is None # Only 1 ticker, no correlation matrix + sim.add_ticker("GOOGL") + assert sim._cholesky is not None +``` + +```python +# test_massive.py — mock the SDK boundary, not the network +def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock: + snap = MagicMock() + snap.ticker = ticker + snap.last_trade.price = price + snap.last_trade.timestamp = timestamp_ms + return snap + + +async def test_malformed_snapshot_skipped(): + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL", "BAD"] + + good = _make_snapshot("AAPL", 190.50, 1707580800000) + bad = MagicMock(ticker="BAD", last_trade=None) # triggers AttributeError + + with patch.object(source, "_fetch_snapshots", return_value=[good, bad]): + await source._poll_once() + + assert cache.get_price("AAPL") == 190.50 + assert cache.get_price("BAD") is None +``` + +Because `massive_client.py` imports `RESTClient` and `SnapshotMarketType` at module level (§8), and `massive` is a core dependency, `test_massive.py` runs without needing `create=True` patch tricks — patch targets exist at import time as long as `uv sync` (not `uv sync --no-dev` against a stripped lockfile) has installed the declared dependency. + +SSE (`stream.py`) is best covered with an ASGI test client once `main.py` exists (e.g. `httpx.AsyncClient` against the FastAPI app), asserting that a connected client receives at least one `data:` event containing a known seeded ticker. + +--- + +## 15. Configuration Summary + +| Parameter | Location | Default | Description | +|---|---|---|---| +| `MASSIVE_API_KEY` | Environment variable | `""` (empty) | If set and non-empty, use Massive API; otherwise use the simulator. | +| `update_interval` | `SimulatorDataSource.__init__` | `0.5` (seconds) | Time between simulator ticks. | +| `poll_interval` | `MassiveDataSource.__init__` | `15.0` (seconds) | Time between Massive API polls (free-tier safe; lower for paid tiers). | +| `event_probability` | `GBMSimulator.__init__` | `0.001` | Chance of a random shock event per ticker per tick. | +| `dt` | `GBMSimulator.__init__` | `~8.48e-8` | GBM time step, as a fraction of a trading year. | +| SSE push interval | `_generate_events()` | `0.5` (seconds) | Time between cache-version checks / pushes to the client. | +| SSE retry directive | `_generate_events()` | `1000` (ms) | Browser `EventSource` reconnection delay after a dropped connection. | + +### Public package API (`backend/app/market/__init__.py`) + +```python +from .cache import PriceCache +from .factory import create_market_data_source +from .interface import MarketDataSource +from .models import PriceUpdate +from .stream import create_stream_router + +__all__ = [ + "PriceUpdate", + "PriceCache", + "MarketDataSource", + "create_market_data_source", + "create_stream_router", +] +``` + +### Quick reference for downstream code + +```python +from app.market import PriceCache, create_market_data_source + +# Startup +cache = PriceCache() +source = create_market_data_source(cache) # Reads MASSIVE_API_KEY +await source.start(["AAPL", "GOOGL", "MSFT", ...]) + +# Read prices +update = cache.get("AAPL") # PriceUpdate or None +price = cache.get_price("AAPL") # float or None +all_prices = cache.get_all() # dict[str, PriceUpdate] + +# Dynamic watchlist +await source.add_ticker("TSLA") +await source.remove_ticker("GOOGL") + +# Shutdown +await source.stop() +``` diff --git a/planning/MARKET_DATA_REVIEW.md b/planning/MARKET_DATA_REVIEW.md new file mode 100644 index 000000000..2f861762a --- /dev/null +++ b/planning/MARKET_DATA_REVIEW.md @@ -0,0 +1,153 @@ +# Market Data Backend — Code Review + +**Date:** 2026-08-18 +**Scope:** `backend/app/market/` (9 source files) and `backend/tests/market/` (9 test files) +**Reviewer environment:** `uv` installed fresh, `uv sync --extra dev` pulled real `massive==2.2.0`, `numpy==2.4.2`, `fastapi==0.128.7`, etc. (not the versions implied by `planning/MARKET_DATA_DESIGN.md`, which matters — see §3.1). + +This supersedes `planning/archive/MARKET_DATA_REVIEW.md` (2026-02-10). All 7 issues from that review have been fixed (build config, `get_tickers()` encapsulation, SSE return type, module-level router, unused imports, Massive test fragility). This pass re-reviewed the code fresh and ran everything against real dependencies rather than trusting the prior writeup's conclusions. + +--- + +## 1. Test Results + +**99 tests collected, 99 passed, 0 failed.** (Up from 73 in the prior review — new `test_interface.py`, `test_seed_prices.py`, and `test_stream.py` modules have been added.) + +``` +uv run --extra dev pytest -v --cov=app --cov-report=term-missing +... +99 passed in 4.33s +``` + +**Coverage: 97% overall** (up from 84%): + +| Module | Coverage | Missing | +|---|---|---| +| models.py | 100% | | +| cache.py | 100% | | +| interface.py | 100% | | +| seed_prices.py | 100% | | +| factory.py | 100% | | +| simulator.py | 98% | `_add_ticker_internal` duplicate-guard, `_run_loop` exception branch (see §3.3) | +| massive_client.py | 94% | `_poll_loop`'s own sleep-loop wrapper, `_fetch_snapshots` body (mocked in every test) | +| stream.py | 92% | `StreamingResponse(...)` construction line, `CancelledError` branch | + +**Lint:** `ruff check app/ tests/` — clean, no violations. + +**Format:** `ruff format --check` — 5 test files would be reformatted (`test_models.py`, `test_seed_prices.py`, `test_simulator.py`, `test_simulator_source.py`, `test_stream.py`). Trivial, not caught by `ruff check` since these are whitespace-only diffs, not lint rule violations. + +--- + +## 2. Architecture Assessment + +The strategy-pattern design holds up well: + +``` +MarketDataSource (ABC) +├── SimulatorDataSource (GBM simulator) +└── MassiveDataSource (Polygon.io REST poller) + │ + ▼ + PriceCache (shared, thread-safe) + │ + ▼ + SSE stream → Frontend +``` + +Confirmed by direct testing, not just inspection: +- GBM prices stay strictly positive over 10,000 steps (existing test, verified). +- Cholesky decomposition succeeds for the **full realistic 10-ticker default watchlist** (7 tech + 2 finance + TSLA) — verified manually since no test exercises this combination (see §3.2). +- Cholesky also succeeds under stress with 110 tickers (10 seeded + 100 dynamically-added unknowns) — no `LinAlgError`, correlation structure is safely PSD across the sizes this app will ever see. +- All previously-fixed issues stayed fixed: `pyproject.toml` has `[tool.hatch.build.targets.wheel] packages = ["app"]`, `massive_client.py` imports `RESTClient`/`SnapshotMarketType` at module level, `GBMSimulator.get_tickers()` is public, `stream.py`'s `_generate_events` is correctly typed `AsyncGenerator[str, None]`, and `create_stream_router()` builds a fresh `APIRouter` per call instead of reusing a module-level singleton. + +--- + +## 3. Issues Found + +### 3.1 Massive API integration is broken against the real SDK (Severity: **High**) + +`massive_client.py:103`: + +```python +price = snap.last_trade.price +timestamp = snap.last_trade.timestamp / 1000.0 +``` + +The installed `massive` package (`v2.2.0`, resolved from the `>=1.0.0` constraint in `pyproject.toml`) has **no `timestamp` attribute on `LastTrade`**. The actual dataclass (`massive/rest/models/trades.py`) exposes `sip_timestamp`, `participant_timestamp`, and `trf_timestamp` — never a bare `timestamp` — and those fields are **Unix nanoseconds**, not milliseconds. + +Reproduced end-to-end against the real SDK model (not a mock): + +```python +from massive.rest.models.snapshot import TickerSnapshot +snap = TickerSnapshot.from_dict({"ticker": "AAPL", "lastTrade": {"p": 190.50, "t": 1707580800123456789}}) +# ... source._poll_once() with this snapshot ... +``` +``` +Skipping snapshot for AAPL: 'LastTrade' object has no attribute 'timestamp' +cache price for AAPL: None +``` + +**Impact:** with a valid `MASSIVE_API_KEY` and a successful API call, `_poll_once` hits `AttributeError` on *every* snapshot, every poll cycle, forever. The existing `except (AttributeError, TypeError)` handler swallows it and logs a per-ticker warning, so the app never crashes — it just silently never populates the cache with real prices. The SSE connection stays "connected" in the UI (per §13.3 of the design doc, this is meant to describe an *invalid key* scenario) but with an empty/stale feed even when the key is perfectly valid and the network call succeeds. This is worse than the documented failure mode because there's no operational signal beyond a repeating log warning — a user pointing FinAlly at real data would see a blank watchlist and have no obvious reason why. + +**Why the test suite didn't catch this:** every test in `test_massive.py` builds its fake snapshot with `MagicMock()` (`snap.last_trade.timestamp = timestamp_ms`), and `MagicMock` auto-vivifies any attribute you assign or access — so the mock happily has a `.timestamp` attribute that the real `LastTrade` dataclass does not. The tests validate the *code's own logic* (skip-on-error, timestamp division, cache writes) but never validate that the attribute names asserted against actually exist on the real SDK response shape. This is a textbook mock/reality divergence. + +**Fix:** use `snap.last_trade.sip_timestamp` (or `participant_timestamp`, whichever semantic is preferred) and divide by `1_000_000_000.0` (nanoseconds → seconds), not `1000.0`. Also worth adding one test that constructs a real `TickerSnapshot`/`LastTrade` via `from_dict()` (as done for this repro) rather than a bare `MagicMock`, so a future SDK-shape mismatch fails a test instead of failing silently in production. + +### 3.2 No test exercises the full default 10-ticker watchlist (Severity: Medium) + +`test_simulator.py` only ever constructs `GBMSimulator` with 1–2 tickers. The production correlation structure (tech intra=0.6, finance intra=0.5, TSLA=0.3, cross=0.3) is a non-trivial block matrix, not simple equicorrelation — it's the kind of structure where Cholesky decomposition can fail to be positive-semidefinite for the wrong parameter combination. It happens to work (verified manually in this review), but nothing in the test suite would catch it if a future correlation constant change broke that invariant. This was already flagged in the prior archived review (§4.2) and is still open. + +**Fix:** add a test that builds `GBMSimulator(tickers=list(SEED_PRICES.keys()))` and asserts `step()` succeeds without raising, covering the real production shape. + +### 3.3 `_run_loop` exception-handling branch is untested (Severity: Low) + +`simulator.py:271-272` (the `except Exception: logger.exception(...)` in `SimulatorDataSource._run_loop`) is never exercised — confirmed by the coverage report and by reading `test_simulator_source.py::test_exception_resilience`, which despite its name never injects a failure. It just asserts the background task is still running after a normal sleep. The resilience the design doc claims for this loop (a bad tick can't kill the feed) is real code but not verified by any test. + +**Fix:** patch `GBMSimulator.step` (or the cache's `update`) to raise once, then assert the task survives and continues writing to the cache on the next tick. + +### 3.4 `MassiveDataSource` ticker-case inconsistency between write paths (Severity: Low) + +`start()`, `add_ticker()`, and `remove_ticker()` all normalize tickers via `.upper().strip()` before touching `self._tickers` or the cache. `_poll_once()` does not: it writes `self._cache.update(ticker=snap.ticker, ...)` using whatever casing the API response returns, unnormalized. In practice Polygon/Massive symbols are already uppercase, so this is low real-world risk, but it's an inconsistency in the codebase's own normalization discipline — if the API ever returned a differently-cased ticker, it would create a second cache entry rather than updating the existing one, silently splitting a ticker's price history in two. + +### 3.5 Pre-start asymmetry between the two `MarketDataSource` implementations (Severity: Trivial) + +`SimulatorDataSource.add_ticker()`/`remove_ticker()` are silent no-ops if called before `start()` (`self._sim` is `None`, guarded by `if self._sim:` with no else/log). `MassiveDataSource.add_ticker()` has no such guard — it will happily append to `self._tickers` even pre-`start()`. Neither behavior is wrong per the ABC's documented contract (`start()` must be called first), but the two implementations diverge in what happens if a caller violates that contract, which could produce different debugging experiences depending on which source is active. Worth a one-line note in `interface.py` or aligning the guard, not urgent. + +### 3.6 `PriceCache.version` still reads outside the lock (Severity: Trivial, carried over) + +Unchanged from the prior review (§3.4 there): `version` is a plain property read without `self._lock`. Harmless under CPython's GIL for a single `int` read; the design doc (§13.4) explicitly accepts this tradeoff at the project's target scale. Not a regression, just noting it's still the case and still fine. + +### 3.7 No concurrent-writer test for `PriceCache` (Severity: Trivial, carried over) + +Also unchanged from the prior review (§4.2 there). The lock usage reads correctly by inspection and there's no evidence of an actual bug, but a multi-thread stress test would give empirical confidence rather than relying on code review alone — relevant because the Massive path's synchronous SDK calls run via `asyncio.to_thread`, i.e. a real OS thread, while the simulator and SSE reader operate on the event loop. + +--- + +## 4. What's Solid + +- **Strategy pattern is clean** — `PriceCache` genuinely decouples both producers from all consumers; nothing downstream branches on which source is active. +- **GBM math is correct and numerically stable** — verified prices stay positive over 10k steps, per-tick moves are appropriately small given the `dt` scale, and the exponential formulation can't underflow to zero or go negative. +- **Correlated moves work in practice**, not just in theory — manually confirmed against both the real 10-ticker production set and a 110-ticker stress case. +- **All 7 issues from the prior (archived) review are genuinely fixed**, not just marked fixed — verified each one directly in the current source rather than trusting the summary doc. +- **Defensive error handling is real** — a malformed snapshot for one ticker doesn't take down the batch; an API failure doesn't crash the poll loop; `stop()` is idempotent on both implementations. +- **97% coverage, 99/99 tests green, lint clean.** The test suite is broad and mostly well-targeted — it just has one dangerous blind spot (§3.1) common to any test suite that mocks a third-party SDK's response shape instead of constructing it. + +--- + +## 5. Verdict + +The market data subsystem is well-architected and the simulator path (the default, no-API-key mode most users and all E2E tests will exercise) is solid, correct, and thoroughly tested. **The Massive/real-data path is currently non-functional** due to §3.1 — this is the one finding that should block calling the Massive integration "done," since it means the `MASSIVE_API_KEY` feature described in `PLAN.md` §5–§6 doesn't actually work today despite 100% of its tests passing. + +**Must fix before considering Massive integration complete:** +1. §3.1 — `snap.last_trade.timestamp` → `snap.last_trade.sip_timestamp`, and the unit conversion from nanoseconds (`/ 1_000_000_000.0`) instead of milliseconds (`/ 1000.0`). Add at least one test built from a real `TickerSnapshot.from_dict()` payload, not a `MagicMock`, to guard against this class of regression. + +**Should fix:** +2. §3.2 — add a test covering the full default 10-ticker `SEED_PRICES` set through `GBMSimulator`. +3. §3.3 — make `test_exception_resilience` actually inject a failure. +4. §3.4 — normalize ticker casing in `_poll_once()` to match the other write paths. + +**Nice to have:** +5. §3.5 — align pre-start behavior between `SimulatorDataSource` and `MassiveDataSource`, or document the difference. +6. Run `ruff format` on the 5 flagged test files. +7. §3.7 — a concurrent-writer stress test for `PriceCache`, for empirical (not just inspection-based) confidence. + +Simulator-only deployments (the default path with no `MASSIVE_API_KEY`) can proceed without blocking on this review. Anything depending on real market data via Massive should not be considered ready until §3.1 is fixed.