diff --git a/.github/workflows/claude-code-review.yml b/.github/workflows/claude-code-review.yml index b5e8cfd4d..37e66f3fd 100644 --- a/.github/workflows/claude-code-review.yml +++ b/.github/workflows/claude-code-review.yml @@ -38,7 +38,8 @@ jobs: claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }} plugin_marketplaces: 'https://github.com/anthropics/claude-code.git' plugins: 'code-review@claude-code-plugins' - prompt: '/code-review:code-review ${{ github.repository }}/pull/${{ github.event.pull_request.number }}' + prompt: '/code-review:code-review --comment ${{ github.repository }}/pull/${{ github.event.pull_request.number }}' + claude_args: '--allowedTools "mcp__github_inline_comment__create_inline_comment"' # 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 diff --git a/.github/workflows/claude.yml b/.github/workflows/claude.yml index d300267f1..6b15fac7a 100644 --- a/.github/workflows/claude.yml +++ b/.github/workflows/claude.yml @@ -46,5 +46,5 @@ jobs: # 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:*)' + # claude_args: '--allowed-tools Bash(gh pr *)' diff --git a/backend/app/market/massive_client.py b/backend/app/market/massive_client.py index 00bc7b2aa..525003ae7 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() diff --git a/backend/app/market/simulator.py b/backend/app/market/simulator.py index b6803f592..2753679a0 100644 --- a/backend/app/market/simulator.py +++ b/backend/app/market/simulator.py @@ -217,8 +217,13 @@ def __init__( self._task: asyncio.Task | None = None async def start(self, tickers: list[str]) -> None: + # dt must track the actual tick interval, not the 500ms default, or the + # annualized vol/drift in seed_prices.py silently becomes wrong whenever + # update_interval is anything other than 0.5s. + dt = self._interval / GBMSimulator.TRADING_SECONDS_PER_YEAR self._sim = GBMSimulator( tickers=tickers, + dt=dt, event_probability=self._event_prob, ) # Seed the cache with initial prices so SSE has data immediately diff --git a/backend/app/market/stream.py b/backend/app/market/stream.py index 7fd974b7c..c2358760c 100644 --- a/backend/app/market/stream.py +++ b/backend/app/market/stream.py @@ -14,14 +14,15 @@ logger = logging.getLogger(__name__) -router = APIRouter(prefix="/api/stream", tags=["streaming"]) - def create_stream_router(price_cache: PriceCache) -> APIRouter: """Create the SSE streaming router with a reference to the price cache. This factory pattern lets us inject the PriceCache without globals. + A fresh APIRouter is built on each call so repeated calls (e.g. across + tests) never double-register the route on a shared instance. """ + router = APIRouter(prefix="/api/stream", tags=["streaming"]) @router.get("/prices") async def stream_prices(request: Request) -> StreamingResponse: diff --git a/backend/pyproject.toml b/backend/pyproject.toml index e172cca22..40dcabf23 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -18,6 +18,7 @@ dev = [ "pytest-asyncio>=0.24.0", "pytest-cov>=5.0.0", "ruff>=0.7.0", + "httpx>=0.27.0", ] [build-system] diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 14545f124..898ca8746 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -1,11 +1 @@ """Pytest configuration and fixtures.""" - -import pytest - - -@pytest.fixture -def event_loop_policy(): - """Use the default event loop policy for all async tests.""" - import asyncio - - return asyncio.DefaultEventLoopPolicy() diff --git a/backend/tests/market/test_cache.py b/backend/tests/market/test_cache.py index b5ab3d55d..e7d21eaca 100644 --- a/backend/tests/market/test_cache.py +++ b/backend/tests/market/test_cache.py @@ -1,5 +1,7 @@ """Tests for PriceCache.""" +import threading + from app.market.cache import PriceCache @@ -101,3 +103,29 @@ def test_price_rounding(self): cache = PriceCache() update = cache.update("AAPL", 190.12345) assert update.price == 190.12 + + def test_concurrent_updates_are_not_lost(self): + """Many threads hammering update() concurrently must not lose or + corrupt writes — the version counter should exactly equal the number + of update() calls, and every ticker should end up with a valid entry. + """ + cache = PriceCache() + tickers = [f"T{i}" for i in range(10)] + updates_per_thread = 200 + num_threads = 8 + + def worker(thread_id: int) -> None: + for i in range(updates_per_thread): + ticker = tickers[(thread_id + i) % len(tickers)] + cache.update(ticker, 100.0 + i) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(num_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert cache.version == num_threads * updates_per_thread + assert len(cache) == len(tickers) + for ticker in tickers: + assert cache.get(ticker) is not None diff --git a/backend/tests/market/test_massive.py b/backend/tests/market/test_massive.py index cdd7dbd24..a200607b4 100644 --- a/backend/tests/market/test_massive.py +++ b/backend/tests/market/test_massive.py @@ -199,3 +199,24 @@ async def test_start_immediate_poll(self): assert cache.get_price("AAPL") == 190.50 await source.stop() + + async def test_start_normalizes_ticker_case(self): + """Test that start() uppercases/strips tickers, matching add/remove_ticker. + + Regression test: previously start() stored tickers verbatim while + remove_ticker() normalized before filtering, so a ticker passed to + start() in lowercase could never be removed. + """ + 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.remove_ticker("aapl") + assert source.get_tickers() == ["GOOGL"] + + await source.stop() 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_simulator.py b/backend/tests/market/test_simulator.py index 1845ec16b..75d1fe870 100644 --- a/backend/tests/market/test_simulator.py +++ b/backend/tests/market/test_simulator.py @@ -126,6 +126,23 @@ 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 + + def test_full_default_watchlist_builds_valid_cholesky(self): + """The full 10-ticker default watchlist's correlation matrix (mixing + tech, finance, and TSLA's special-cased correlation) must produce a + valid Cholesky decomposition and step cleanly, not just 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 + assert sim._cholesky.shape == (len(tickers), len(tickers)) + + for _ in range(50): + result = sim.step() + assert set(result.keys()) == set(tickers) + for price in result.values(): + assert price > 0 diff --git a/backend/tests/market/test_simulator_source.py b/backend/tests/market/test_simulator_source.py index 515ce7290..3c99c4ba1 100644 --- a/backend/tests/market/test_simulator_source.py +++ b/backend/tests/market/test_simulator_source.py @@ -5,7 +5,7 @@ import pytest from app.market.cache import PriceCache -from app.market.simulator import SimulatorDataSource +from app.market.simulator import GBMSimulator, SimulatorDataSource @pytest.mark.asyncio @@ -128,11 +128,26 @@ 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 await asyncio.sleep(0.2) await source.stop() + + async def test_dt_scales_with_update_interval(self): + """The GBM dt must track update_interval, not always assume 500ms. + + Regression test: previously GBMSimulator was always constructed with + its 500ms-derived DEFAULT_DT regardless of update_interval, so a + faster/slower tick rate silently changed the simulator's effective + annualized volatility instead of just its update frequency. + """ + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + + expected_dt = 0.1 / GBMSimulator.TRADING_SECONDS_PER_YEAR + assert source._sim._dt == pytest.approx(expected_dt) + + await source.stop() diff --git a/backend/tests/market/test_stream.py b/backend/tests/market/test_stream.py new file mode 100644 index 000000000..105af9a8b --- /dev/null +++ b/backend/tests/market/test_stream.py @@ -0,0 +1,167 @@ +"""Integration tests for the SSE price streaming endpoint. + +`_generate_events` is a `while True` loop that only exits when the client +disconnects. Both `httpx.ASGITransport` and Starlette's `TestClient` fully +await an ASGI call to completion before handing back anything to consume — +neither delivers a disconnect mid-stream, so driving this endpoint through +either one deadlocks (confirmed empirically). Instead, these tests drive +`_generate_events` directly with a minimal fake `Request` whose +`is_disconnected()` we control, and separately exercise the routed endpoint +function (returned by `create_stream_router`) to cover response/header +wiring without consuming its unbounded body. +""" + +import asyncio +import json + +import pytest +from fastapi.responses import StreamingResponse + +from app.market.cache import PriceCache +from app.market.stream import _generate_events, create_stream_router + + +class _FakeClient: + host = "test-client" + + +class _FakeRequest: + """Minimal stand-in for fastapi.Request — only what _generate_events uses.""" + + def __init__(self) -> None: + self.client = _FakeClient() + self._disconnected = False + + async def is_disconnected(self) -> bool: + return self._disconnected + + def disconnect(self) -> None: + self._disconnected = True + + +def _parse_data_event(event: str) -> dict: + assert event.startswith("data: ") + assert event.endswith("\n\n") + return json.loads(event[len("data: ") : -2]) + + +class TestCreateStreamRouter: + """Tests for the router factory itself (not the streaming body).""" + + def test_builds_independent_routers(self): + """Each call must return its own APIRouter, not share module state. + + Regression test: create_stream_router() used to decorate onto a + shared module-level router, so calling it twice would register + /prices twice on the same object. + """ + cache = PriceCache() + router_a = create_stream_router(cache) + router_b = create_stream_router(cache) + + assert router_a is not router_b + assert len(router_a.routes) == 1 + assert len(router_b.routes) == 1 + + @pytest.mark.asyncio + async def test_endpoint_returns_streaming_response_with_sse_headers(self): + cache = PriceCache() + cache.update("AAPL", 190.00) + router = create_stream_router(cache) + endpoint = router.routes[0].endpoint + + response = await endpoint(_FakeRequest()) + + assert isinstance(response, StreamingResponse) + assert response.media_type == "text/event-stream" + assert response.headers["cache-control"] == "no-cache" + assert response.headers["connection"] == "keep-alive" + assert response.headers["x-accel-buffering"] == "no" + + +@pytest.mark.asyncio +class TestGenerateEvents: + """Tests for the _generate_events async generator directly.""" + + async def test_first_event_is_retry_directive(self): + cache = PriceCache() + request = _FakeRequest() + gen = _generate_events(cache, request, interval=0.01) + + first = await gen.__anext__() + assert first == "retry: 1000\n\n" + + request.disconnect() + with pytest.raises(StopAsyncIteration): + await gen.__anext__() + + async def test_emits_seeded_prices(self): + cache = PriceCache() + cache.update("AAPL", 190.50) + cache.update("GOOGL", 175.25) + request = _FakeRequest() + gen = _generate_events(cache, request, interval=0.01) + + await gen.__anext__() # retry directive + payload = _parse_data_event(await gen.__anext__()) + + assert payload["AAPL"]["price"] == 190.50 + assert payload["AAPL"]["direction"] == "flat" + assert payload["GOOGL"]["price"] == 175.25 + + request.disconnect() + with pytest.raises(StopAsyncIteration): + await gen.__anext__() + + async def test_reflects_subsequent_updates(self): + cache = PriceCache() + cache.update("AAPL", 190.00) + request = _FakeRequest() + gen = _generate_events(cache, request, interval=0.01) + + await gen.__anext__() # retry directive + first_payload = _parse_data_event(await gen.__anext__()) + assert first_payload["AAPL"]["price"] == 190.00 + + cache.update("AAPL", 191.00) + second_payload = _parse_data_event(await gen.__anext__()) + assert second_payload["AAPL"]["price"] == 191.00 + assert second_payload["AAPL"]["direction"] == "up" + + request.disconnect() + with pytest.raises(StopAsyncIteration): + await gen.__anext__() + + async def test_no_data_event_while_cache_stays_empty(self): + """With an empty cache, the version never changes, so no data event + should ever be produced — only the initial retry directive.""" + cache = PriceCache() + request = _FakeRequest() + gen = _generate_events(cache, request, interval=0.02) + + first = await gen.__anext__() + assert first == "retry: 1000\n\n" + + async def disconnect_after_several_ticks() -> None: + await asyncio.sleep(0.1) # ~5 ticks at a 0.02s interval + request.disconnect() + + asyncio.create_task(disconnect_after_several_ticks()) + + # If a data event had been produced, anext() would return it instead + # of the loop eventually hitting the disconnect and raising here. + with pytest.raises(StopAsyncIteration): + await gen.__anext__() + + async def test_stops_on_disconnect_mid_stream(self): + cache = PriceCache() + cache.update("AAPL", 190.00) + request = _FakeRequest() + gen = _generate_events(cache, request, interval=0.01) + + await gen.__anext__() # retry directive + await gen.__anext__() # initial data event + + request.disconnect() + with pytest.raises(StopAsyncIteration): + await gen.__anext__() diff --git a/backend/uv.lock b/backend/uv.lock index 67d471b2d..fd4977954 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -177,6 +177,7 @@ dependencies = [ [package.optional-dependencies] dev = [ + { name = "httpx" }, { name = "pytest" }, { name = "pytest-asyncio" }, { name = "pytest-cov" }, @@ -186,6 +187,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "fastapi", specifier = ">=0.115.0" }, + { name = "httpx", marker = "extra == 'dev'", specifier = ">=0.27.0" }, { name = "massive", specifier = ">=1.0.0" }, { name = "numpy", specifier = ">=2.0.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.3.0" }, @@ -206,6 +208,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" }, ] +[[package]] +name = "httpcore" +version = "1.0.9" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "certifi" }, + { name = "h11" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/94/82699a10bca87a5556c9c59b5963f2d039dbd239f25bc2a63907a05a14cb/httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8", size = 85484, upload-time = "2025-04-24T22:06:22.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" }, +] + [[package]] name = "httptools" version = "0.7.1" @@ -235,6 +250,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/53/cf/878f3b91e4e6e011eff6d1fa9ca39f7eb17d19c9d7971b04873734112f30/httptools-0.7.1-cp314-cp314-win_amd64.whl", hash = "sha256:cfabda2a5bb85aa2a904ce06d974a3f30fb36cc63d7feaddec05d2050acede96", size = 88205, upload-time = "2025-10-10T03:55:00.389Z" }, ] +[[package]] +name = "httpx" +version = "0.28.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "certifi" }, + { name = "httpcore" }, + { name = "idna" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b1/df/48c586a5fe32a0f01324ee087459e112ebb7224f646c0b5023f5e79e9956/httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc", size = 141406, upload-time = "2024-12-06T15:37:23.222Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" }, +] + [[package]] name = "idna" version = "3.11" diff --git a/planning/MARKET_DATA_DESIGN.md b/planning/MARKET_DATA_DESIGN.md new file mode 100644 index 000000000..658f747e1 --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,1388 @@ +# Market Data Backend — Design Document + +Implementation-ready design for the FinAlly market data subsystem: a unified +interface with two interchangeable implementations (a GBM simulator and a +Massive/Polygon.io REST poller), a thread-safe price cache, and an SSE +streaming endpoint. This document reflects the **as-built** code in +`backend/app/market/` (8 modules) — it is the authoritative reference for +any agent wiring the rest of the backend (portfolio, watchlist, chat) on top +of live prices. + +Status: this subsystem is complete and tested (see +`planning/MARKET_DATA_SUMMARY.md`). `backend/app/main.py` — the FastAPI app +that wires this subsystem into a running server — does not exist yet; §10 +below is the design for building it. + +--- + +## Table of Contents + +1. [Architecture](#1-architecture) +2. [File Structure](#2-file-structure) +3. [Data Model — `models.py`](#3-data-model) +4. [Price Cache — `cache.py`](#4-price-cache) +5. [Abstract Interface — `interface.py`](#5-abstract-interface) +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. [FastAPI Integration — `stream.py` and `main.py`](#10-fastapi-integration) +11. [Watchlist Coordination](#11-watchlist-coordination) +12. [Testing Strategy](#12-testing-strategy) +13. [Error Handling & Edge Cases](#13-error-handling--edge-cases) +14. [Configuration Summary](#14-configuration-summary) + +--- + +## 1. Architecture + +``` + MarketDataSource (ABC) + / \ + SimulatorDataSource MassiveDataSource + (GBM, in-process, (Polygon.io REST + no external deps) poller, needs API key) + \ / + v v + PriceCache + (thread-safe, in-memory, + versioned for change detection) + / | \ + v v v + SSE /api/stream Portfolio Trade + /prices valuation execution +``` + +**Strategy pattern.** Both data sources implement `MarketDataSource`. +Downstream code (SSE streaming, portfolio valuation, trade execution) never +knows or cares which one is active — it only ever talks to the shared +`PriceCache`. + +**Push model, not pull.** A data source is not asked for a price; it writes +prices into the cache on its own schedule (simulator: every 500ms, Massive: +every 15s by default). Readers poll the cache at whatever cadence they need, +decoupled from the producer's timing. + +**Selection is environment-driven.** `create_market_data_source()` picks the +implementation based on whether `MASSIVE_API_KEY` is set — no code change +needed to switch between simulated and real data (see [PLAN.md §6](PLAN.md)). + +--- + +## 2. File Structure + +``` +backend/ + app/ + market/ + __init__.py # Re-exports: PriceUpdate, PriceCache, MarketDataSource, + # create_market_data_source, create_stream_router + models.py # PriceUpdate dataclass + cache.py # PriceCache (thread-safe in-memory store) + interface.py # MarketDataSource ABC + 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) + market_data_demo.py # Rich terminal demo (uv run market_data_demo.py) + 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 +``` + +--- + +## 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 +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; once + created they never change, so they're safe to share across async tasks + without copying. +- **`slots=True`** — memory optimization; many of these are created per + second. +- **Computed properties** (`change`, `change_percent`, `direction`) are + derived from `price`/`previous_price` so they can never drift out of sync + with each other — there is no stale `direction` field to forget to update. +- **`to_dict()`** is the single serialization point used by both the SSE + endpoint and any future REST API response. + +--- + +## 4. Price Cache + +**File: `backend/app/market/cache.py`** + +The central data hub. Data sources write to it; SSE streaming and (later) +portfolio valuation / trade execution read from it. It must be thread-safe +because the Massive client's synchronous calls run in a thread-pool +executor via `asyncio.to_thread`, while SSE reads happen on the event loop. + +```python +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 send all prices on every tick even when nothing +changed (e.g. Massive only updates every 15s). The counter lets the SSE +loop skip sends when there's nothing 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()` runs via + `asyncio.to_thread()`, which executes in a real OS thread — + `asyncio.Lock` would not protect against that. +- `threading.Lock` works correctly from both sync threads and the async + event loop, so one cache implementation serves both data sources. + +--- + +## 5. Abstract Interface + +**File: `backend/app/market/interface.py`** + +```python +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 at 500ms, Massive +polls at 15s, but SSE always reads from the cache at its own 500ms cadence. +The SSE layer never needs to know which data source is active or what its +update interval is. + +--- + +## 6. Seed Prices & Ticker Parameters + +**File: `backend/app/market/seed_prices.py`** + +Constants only — no logic, no imports beyond stdlib. Shared by the +simulator (initial prices and GBM parameters) and available as a fallback +reference for any code that wants a sane starting price for an unknown +ticker. + +```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 added dynamically that aren't in `SEED_PRICES`/`TICKER_PARAMS` fall +back to a random seed price between $50–$300 and `DEFAULT_PARAMS`. + +--- + +## 7. GBM Simulator + +**File: `backend/app/market/simulator.py`** + +Two classes live here: `GBMSimulator` (pure math engine, stateful) and +`SimulatorDataSource` (the `MarketDataSource` implementation that wraps it +in an async loop and writes to the `PriceCache`). + +### 7.1 The math + +Geometric Brownian Motion is the standard model underlying Black-Scholes: +prices evolve continuously with random noise, can never go negative, and +follow the lognormal distribution seen in real markets. + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +- `S(t)` — current price +- `mu` — annualized drift (expected return), e.g. `0.05` +- `sigma` — annualized volatility, e.g. `0.20` +- `dt` — time step as a fraction of a trading year +- `Z` — a (correlated) standard normal random variable + +For 500ms ticks over a 252-day, 6.5-hour trading year: + +``` +dt = 0.5 / (252 * 6.5 * 3600) ≈ 8.48e-8 +``` + +This tiny `dt` produces sub-cent moves per tick that accumulate naturally +into realistic intraday ranges over time. Prices can never go negative +because the update is multiplicative through `exp()`. + +### 7.2 Correlated moves via Cholesky decomposition + +Real stocks don't move independently — tech stocks tend to move together. +Given a correlation matrix `C`, compute `L = cholesky(C)`; for independent +standard normals `Z_independent`, `Z_correlated = L @ Z_independent` gives +draws with the desired correlation structure. Cholesky decomposition +requires the matrix be positive semi-definite, which holds for any valid +correlation matrix (all diagonal 1s, off-diagonal in `[-1, 1]`, symmetric). + +Correlation structure used here: + +| Pair | Correlation | +|---|---| +| Tech ↔ tech (AAPL, GOOGL, MSFT, AMZN, META, NVDA, NFLX) | 0.6 | +| Finance ↔ finance (JPM, V) | 0.5 | +| TSLA ↔ anything | 0.3 (it does its own thing) | +| Cross-sector / unknown tickers | 0.3 | + +### 7.3 Random shock events + +Each tick, each ticker has a small probability (default `0.001`) of a +sudden 2–5% move — visual drama for the demo. With 10 tickers at 2 +ticks/sec, expect an event roughly every 50 seconds. + +### 7.4 Full implementation + +```python +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) + """ + + # 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}. + + 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] = 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"] + + # TSLA is in the tech set but behaves independently + 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). +- **Graceful cancellation** — `stop()` cancels the task and awaits it, + swallowing `CancelledError`, for clean shutdown during FastAPI lifespan + teardown. +- **Exception resilience** — the loop catches exceptions per-step so one + bad tick doesn't kill the entire feed. +- **`GBMSimulator.get_tickers()`** is a public method — `SimulatorDataSource` + never reaches into a private attribute to expose the ticker list. + +--- + +## 8. Massive API Client + +**File: `backend/app/market/massive_client.py`** + +Polls the Massive (formerly Polygon.io) REST API snapshot endpoint on a +configurable interval. The client is synchronous, so it runs inside +`asyncio.to_thread()` to avoid blocking the event loop. + +### 8.1 Massive API primer + +- **Package**: `massive` (declared as a core dependency in + `backend/pyproject.toml`; `uv add massive`) +- **Auth**: `RESTClient(api_key=...)` — reads `MASSIVE_API_KEY` automatically + if omitted +- **Rate limits**: free tier 5 req/min → poll every 15s; paid tiers support + polling every 2–5s +- **Primary endpoint**: `GET /v2/snapshot/locale/us/markets/stocks/tickers` + — returns current data for *all requested tickers in one call*, which is + what keeps us within the free-tier rate limit regardless of watchlist size + +```python +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +client = RESTClient(api_key="...") +snapshots = client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "GOOGL", "MSFT"], +) +for snap in snapshots: + print(snap.ticker, snap.last_trade.price, snap.last_trade.timestamp) +``` + +Relevant response fields per ticker: `last_trade.price` (current price used +for trading/display), `last_trade.timestamp` (Unix **milliseconds**), +`day.previous_close` / `day.change_percent` (available if a day-change UI +element is added later). + +### 8.2 Implementation + +```python +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 -> 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, + ) +``` + +Note: imports are at module level (not lazy) because `massive` is a core +dependency of `backend/pyproject.toml` — the simulator path still has zero +*runtime* dependency on it being *configured* (no API key → this module is +simply never instantiated), but the package itself is always installed. + +### 8.3 Error handling philosophy + +The poller is intentionally resilient — a bad response never takes down the +price feed: + +| Error | Behavior | +|---|---| +| **401 Unauthorized** | Logged as error; poller keeps running (user might fix `.env` and restart). | +| **429 Rate Limited** | Logged as error; next poll retries after `poll_interval` seconds. | +| **Network timeout** | Logged as error; retries automatically on next cycle. | +| **Malformed snapshot** | That ticker is skipped with a warning; other tickers in the same response are still processed. | +| **All tickers fail** | Cache retains last-known prices; SSE keeps streaming stale data (better than no data). | + +--- + +## 9. Factory + +**File: `backend/app/market/factory.py`** + +```python +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", ...] +``` + +--- + +## 10. FastAPI Integration + +### 10.1 SSE Streaming Endpoint — `stream.py` (built) + +**File: `backend/app/market/stream.py`** + +A FastAPI route that holds open a long-lived HTTP connection and pushes +price updates to the client as `text/event-stream`. + +```python +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" # Reconnect after 1s on drop + + 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** — each event looks like: + +``` +data: {"AAPL":{"ticker":"AAPL","price":190.50,"previous_price":190.42,"timestamp":1707580800.5,"change":0.08,"change_percent":0.042,"direction":"up"},"GOOGL":{...}} + +``` + +Client-side (`EventSource`, native browser API — no library needed): + +```javascript +const eventSource = new EventSource('/api/stream/prices'); +eventSource.onmessage = (event) => { + const prices = JSON.parse(event.data); + // prices is { "AAPL": { ticker, price, previous_price, change, change_percent, direction, timestamp }, ... } +}; +``` + +**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 +directly. This is simpler and produces evenly-spaced updates, which matters +because the frontend accumulates them into sparkline charts — regular +spacing keeps that visualization clean regardless of which backend data +source is active. + +### 10.2 Lifecycle Integration — `main.py` (not yet built) + +`backend/app/main.py` does not exist yet. This is the design for wiring +the market data subsystem into the FastAPI app via the `lifespan` context +manager, so whoever builds the rest of the backend (portfolio, watchlist, +chat routes) has a concrete pattern to follow. + +```python +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from app.market import PriceCache, create_market_data_source, create_stream_router +from app.market.interface import MarketDataSource + + +@asynccontextmanager +async def lifespan(app: FastAPI): + """Manage startup and shutdown of background services.""" + + # --- STARTUP --- + price_cache = PriceCache() + app.state.price_cache = price_cache + + source = create_market_data_source(price_cache) + app.state.market_source = source + + # Load initial tickers from the database watchlist (lazily initializes + # the DB and seeds default tickers on first run — see PLAN.md §7) + initial_tickers = await load_watchlist_tickers() + await source.start(initial_tickers) + + stream_router = create_stream_router(price_cache) + app.include_router(stream_router) + + yield # App is running + + # --- SHUTDOWN --- + await source.stop() + + +app = FastAPI(title="FinAlly", lifespan=lifespan) + + +def get_price_cache() -> PriceCache: + return app.state.price_cache + + +def get_market_source() -> MarketDataSource: + return app.state.market_source +``` + +Other routes access the price cache and data source via dependency +injection: + +```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}") + # ... execute trade 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) + # ... + + +@router.delete("/watchlist/{ticker}") +async def remove_from_watchlist( + ticker: str, + source: MarketDataSource = Depends(get_market_source), +): + # ... delete from watchlist table ... + await source.remove_ticker(ticker) + # ... +``` + +--- + +## 11. Watchlist Coordination + +When the watchlist changes (via REST API or LLM chat action), the market +data source must be told so it tracks the right set of tickers. + +### Adding a ticker + +``` +User (or LLM) -> POST /api/watchlist {ticker: "PYPL"} + -> Insert into watchlist table (SQLite) + -> await source.add_ticker("PYPL") + Simulator: adds to GBMSimulator, rebuilds Cholesky, seeds cache immediately + Massive: appends to ticker list, appears on next poll (up to poll_interval delay) + -> Return success (ticker + current price if available) +``` + +### Removing a ticker + +``` +User (or LLM) -> DELETE /api/watchlist/PYPL + -> Delete from watchlist table (SQLite) + -> await source.remove_ticker("PYPL") + Simulator: removes from GBMSimulator, rebuilds Cholesky, removes from cache + Massive: removes from ticker list, removes from cache + -> Return success +``` + +### Edge case: ticker still has an open position + +If the user removes a ticker from the watchlist but still holds shares, the +data source must keep tracking it so portfolio valuation stays accurate. +The watchlist route is responsible for this check — the market data layer +itself has no concept of "positions": + +```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) + + return {"status": "ok"} +``` + +--- + +## 12. Testing Strategy + +**File location: `backend/tests/market/`** — 6 modules, 73 tests, 84% +overall coverage (see `planning/MARKET_DATA_SUMMARY.md` for the full +breakdown). Summary of what each module verifies: + +| Module | Focus | +|---|---| +| `test_models.py` | `PriceUpdate` computed properties (`change`, `change_percent`, `direction`), `to_dict()` serialization, immutability | +| `test_cache.py` | update/get/get_all/remove, first-update-is-flat, direction on up/down, version increments on every write | +| `test_simulator.py` | `GBMSimulator.step()` always returns all tickers, prices stay positive over 10k steps, add/remove ticker rebuilds Cholesky, unknown ticker gets a random seed in range, empty ticker list is a no-op | +| `test_simulator_source.py` | `SimulatorDataSource.start()` seeds the cache before the first tick, prices change over time, `stop()` is idempotent, `add_ticker`/`remove_ticker` propagate to both the simulator and the cache | +| `test_factory.py` | `MASSIVE_API_KEY` set → `MassiveDataSource`; unset/empty → `SimulatorDataSource` | +| `test_massive.py` | `_poll_once` updates the cache from mocked snapshots, malformed snapshots are skipped without aborting the batch, API exceptions don't crash the poller | + +Representative test (full suite is in the repo): + +```python +# backend/tests/market/test_simulator.py +class TestGBMSimulator: + def test_prices_are_positive(self): + """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(self): + 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 +# backend/tests/market/test_massive.py +@pytest.mark.asyncio +class TestMassiveDataSource: + async def test_malformed_snapshot_skipped(self): + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL", "BAD"] + + good_snap = _make_snapshot("AAPL", 190.50, 1707580800000) + bad_snap = MagicMock(ticker="BAD", last_trade=None) # triggers AttributeError + + with patch.object(source, "_fetch_snapshots", return_value=[good_snap, bad_snap]): + await source._poll_once() + + assert cache.get_price("AAPL") == 190.50 + assert cache.get_price("BAD") is None +``` + +Run locally: + +```bash +cd backend +uv run --extra dev pytest -v +uv run --extra dev pytest --cov=app +``` + +### Gaps to be aware of + +- `stream.py` has low direct coverage (31%) — exercising the SSE generator + properly requires a running ASGI test client (e.g. `httpx.AsyncClient` + against the FastAPI `app`), which isn't possible until `main.py` exists. + Add an SSE integration test once the app is wired up in §10.2. +- No dedicated concurrent-writer stress test for `PriceCache` (lock + correctness is verified by inspection, not empirically under contention). + +--- + +## 13. Error Handling & Edge Cases + +### 13.1 Startup with an empty watchlist + +If the database has no watchlist entries, `start()` receives an empty +list. Both data sources handle this gracefully — the simulator produces no +prices, the Massive poller skips its API call entirely. The SSE endpoint +simply sends no events until a ticker is added, at which point tracking +starts immediately. + +### 13.2 Price cache miss during a trade + +If a user tries to trade a ticker with no cached price yet (just added, +Massive hasn't polled it): + +```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 entirely by seeding the cache synchronously +inside `add_ticker()`. The Massive client may have a brief gap until its +next poll — the 400 with a clear message is the correct response there. + +### 13.3 Massive API key invalid + +If the key is set but wrong, the first poll fails with 401. The poller +logs the error and keeps retrying every `poll_interval`. SSE keeps +streaming (connected, just empty). The fix is correcting `.env` and +restarting the container. + +### 13.4 Thread safety under load + +`PriceCache` uses a `threading.Lock` (a mutex). Under expected load (10 +tickers, 2 updates/sec, one SSE reader per browser tab) contention is +negligible — the critical section is a dict lookup plus assignment. If this +ever became a bottleneck (hundreds of tickers, many concurrent readers) a +`ReadWriteLock` would be the fix, but that's unnecessary for this project's +scale. + +### 13.5 Simulator numerical precision + +The tiny `dt` produces very small per-tick moves; this is not a precision +concern because prices are rounded to 2 decimals in `GBMSimulator.step()`, +the `exp(drift + diffusion)` formulation is numerically stable, and prices +are always positive by construction (exponential of a real number). + +--- + +## 14. Configuration Summary + +| Parameter | Location | Default | Description | +|---|---|---|---| +| `MASSIVE_API_KEY` | Environment variable | `""` (empty) | If set, use Massive API; otherwise use the simulator | +| `update_interval` | `SimulatorDataSource.__init__` | `0.5` s | Time between simulator ticks | +| `poll_interval` | `MassiveDataSource.__init__` | `15.0` s | Time between Massive API polls (free tier: 5 req/min) | +| `event_probability` | `GBMSimulator.__init__` | `0.001` | Chance of a random shock event per ticker per tick | +| `dt` | `GBMSimulator.__init__` | `~8.5e-8` | GBM time step (fraction of a trading year) | +| SSE push interval | `_generate_events()` | `0.5` s | Time between SSE pushes to a connected client | +| SSE retry directive | `_generate_events()` | `1000` ms | Browser `EventSource` reconnection delay | + +### Package `__init__.py` + +**File: `backend/app/market/__init__.py`** + +```python +"""Market data subsystem for FinAlly. + +Public API: + PriceUpdate - Immutable price snapshot dataclass + PriceCache - Thread-safe in-memory price store + MarketDataSource - Abstract interface for data providers + create_market_data_source - Factory that selects simulator or Massive + create_stream_router - FastAPI router factory for SSE endpoint +""" + +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", +] +``` diff --git a/planning/MARKET_DATA_REVIEW.md b/planning/MARKET_DATA_REVIEW.md new file mode 100644 index 000000000..145c4ee46 --- /dev/null +++ b/planning/MARKET_DATA_REVIEW.md @@ -0,0 +1,87 @@ +# Market Data Backend — Code Review + +**Date:** 2026-09-12 +**Reviewer:** Claude +**Scope:** `backend/app/market/` (8 source modules) and `backend/tests/market/` (7 test files, 84 tests) + +This is the third pass on this document. The first pass found six issues (§1 below); the second pass fixed all six and added regression tests for the two behavioral bugs. This third pass closes out the remaining items that were previously left open on purpose (§2): an SSE integration test, a `PriceCache` concurrency test, and a full-10-ticker `GBMSimulator` test. The market data subsystem is now considered complete and ready for the rest of the backend to build on. + +--- + +## 1. Issues Fixed (Second Pass) + +| # | Issue | Severity | Fix | +|---|---|---|---| +| 1 | `MassiveDataSource.start()` didn't normalize ticker case, but `add_ticker`/`remove_ticker` did — a ticker passed to `start()` in lowercase could never be removed later. | Medium | `start()` now does `[t.upper().strip() for t in tickers]`. Locked in by `test_start_normalizes_ticker_case`. | +| 2 | `GBMSimulator`'s `dt` was hardcoded to assume a 500ms tick regardless of `SimulatorDataSource.update_interval`. | Medium | `SimulatorDataSource.start()` derives `dt = self._interval / GBMSimulator.TRADING_SECONDS_PER_YEAR`. Locked in by `test_dt_scales_with_update_interval`. | +| 3 | `stream.py` built its `APIRouter` at module scope; calling `create_stream_router()` twice would double-register `/prices`. | Low | `router = APIRouter(...)` now built inside `create_stream_router()`. Locked in by `test_builds_independent_routers`. | +| 4 | `PriceCache.version` read without `self._lock`. | Low | Left as-is — see §4. | +| 5 | `tests/conftest.py`'s `event_loop_policy` fixture was a deprecated no-op, producing a `DeprecationWarning` on every async test. | Trivial | Fixture removed. | +| 6 | Formatting drift flagged by `ruff format --check` in several test files. | Trivial | `ruff format` applied. | + +--- + +## 2. Improvements Added (This Pass) + +These were the items previously listed as "worth doing, not urgent" / "deliberately not fixed" because they needed infrastructure the subsystem didn't have yet at the time (an ASGI test harness) or were judged lower value. All three are now done: + +### 2.1 SSE integration test for `stream.py` (was 31% coverage, no tests) + +Getting this right took a real detour worth recording: the natural first attempt — `httpx.ASGITransport` and, separately, FastAPI's `TestClient` — both **deadlock** against this endpoint. `_generate_events` is an unbounded `while True` loop that only exits when it observes `request.is_disconnected()`. Both of those test clients fully run the ASGI call to completion (buffering the entire response) *before* handing anything back to the caller to consume — there is no mechanism for the client to signal a disconnect mid-stream, so the server-side generator never sees one and the client never gets anything back. Confirmed this empirically with `faulthandler`-dumped stack traces showing both hung inside the initial `send()`/`handle_request()` call, before the streaming body was ever reached. + +The fix was to test `_generate_events` directly: a minimal fake `Request` (just `.client.host` and a controllable `is_disconnected()`) drives the async generator with `__anext__()`, so the test controls disconnection deterministically instead of depending on transport-level streaming semantics that don't exist in either test client. `create_stream_router()`'s route-building and the endpoint's `StreamingResponse`/headers are tested separately by calling the routed endpoint function directly (via `router.routes[0].endpoint`) without consuming its body. + +New file: `tests/market/test_stream.py`, 7 tests: +- Router factory builds independent routers per call (locks in fix #3 above) +- Endpoint returns a `StreamingResponse` with the right media type and headers +- First event is the `retry: 1000` directive +- Initial data event reflects whatever's already in the cache +- A cache update after the connection opens streams through as a new event +- No data event is ever produced while the cache stays empty across several ticks +- The generator stops (raises `StopAsyncIteration`) on disconnect + +`stream.py` coverage: **31% → 94%** (only the `asyncio.CancelledError` logging branch remains uncovered — that requires actually cancelling the task rather than a clean disconnect, which is a real server-shutdown path, not something worth engineering a test around). + +Added `httpx>=0.27.0` to `[project.optional-dependencies].dev` — it's what `fastapi.testclient.TestClient` needs even though it isn't used directly in the final tests; harmless to keep since it's dev-only and a natural fit for future API testing. + +### 2.2 `PriceCache` concurrent-writers test + +`tests/market/test_cache.py::test_concurrent_updates_are_not_lost` spins up 8 real OS threads (matching how the cache is actually used — `MassiveDataSource` calls into it via `asyncio.to_thread`), each performing 200 `update()` calls across 10 shared tickers, then asserts `cache.version` exactly equals `8 * 200 = 1600`. This is a meaningful assertion, not a smoke test: a broken lock (or a `+=` race) would show up here as a version count *less than* 1600 — a lost update — with high probability under real thread interleaving. Ran the full suite three times back-to-back to confirm no flakiness. + +### 2.3 Full 10-ticker `GBMSimulator` test + +`tests/market/test_simulator.py::test_full_default_watchlist_builds_valid_cholesky` builds a `GBMSimulator` with all 10 tickers from `SEED_PRICES` (mixing the tech group, the finance group, and TSLA's special-cased correlation all at once — the case none of the existing 1-2 ticker tests exercised), asserts the Cholesky decomposition is a proper 10×10 matrix, and runs 50 steps confirming all tickers stay present and positive. This had already been manually verified working in the first review pass; it's now a permanent regression test. + +--- + +## 3. Test Results (Final) + +**84 tests collected, 84 passed, 0 failed.** (`uv run pytest -q --cov=app --cov-report=term-missing`, `massive` installed via `uv sync --extra dev`.) Verified stable across 3 consecutive runs. + +| Module | Coverage | Notes | +|---|---|---| +| models.py | 100% | | +| cache.py | 100% | | +| interface.py | 100% | | +| seed_prices.py | 100% | | +| factory.py | 100% | | +| simulator.py | 98% | Uncovered: L149 duplicate-add guard, L273-274 exception path in `_run_loop` | +| massive_client.py | 94% | Uncovered: `_poll_loop`'s `while True` body, real (unmocked) `_fetch_snapshots` body | +| stream.py | 94% | Uncovered: `asyncio.CancelledError` logging branch (server-shutdown path) | +| **Total** | **97%** | Up from 91% at the start of this pass | + +**Lint:** `ruff check app/ tests/` — clean. +**Format:** `ruff format --check app/ tests/` — clean, all 20 files formatted. +**Dependency sanity:** `uv sync` (prod-only) and `uv sync --extra dev` both verified to install cleanly from a fresh lockfile resolution. + +--- + +## 4. Remaining Open Item + +- **`PriceCache.version` unlocked read.** Still deliberately left as-is. A single `int` read is atomic under CPython's GIL, this project targets standard CPython, and adding a lock here would be defensive code against a scenario (a no-GIL Python build) this project doesn't target. Re-affirmed in this pass; no plan to change unless the project's Python target changes. + +--- + +## 5. Verdict + +The market data backend is complete, tested, and ready. All issues from both prior review passes are resolved except the one item in §4, which is a deliberate judgment call rather than an oversight. Coverage is 97% overall, with every module except the two background polling loops (whose bodies are `await asyncio.sleep()` + a call already tested directly) above 90%. Nothing here should block building the rest of the backend — portfolio, watchlist, chat, and the FastAPI `app` that will mount `create_stream_router()` for real.