From b605abac5cade71197fdaedf8d642d56fa8a5681 Mon Sep 17 00:00:00 2001 From: jkirkmannz <38672718+jkirkmannz@users.noreply.github.com> Date: Sat, 12 Sep 2026 14:43:07 +1200 Subject: [PATCH 1/5] "Update Claude PR Assistant workflow" --- .github/workflows/claude.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 *)' From 9058d670f8bd90271c5722074503cd2c0f845eeb Mon Sep 17 00:00:00 2001 From: jkirkmannz <38672718+jkirkmannz@users.noreply.github.com> Date: Sat, 12 Sep 2026 14:43:08 +1200 Subject: [PATCH 2/5] "Update Claude Code Review workflow" --- .github/workflows/claude-code-review.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 From 912dabf25c87f64802fc6c2383cbf055e306881d Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 12 Sep 2026 03:33:01 +0000 Subject: [PATCH 3/5] Add detailed market data backend design document Consolidates the interface, simulator, and Massive API design docs into one implementation-accurate reference, matching the as-built code in backend/app/market/ after code review fixes were applied. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01VxEsuAL4E6q2LaTfu9BqPW --- planning/MARKET_DATA_DESIGN.md | 1388 ++++++++++++++++++++++++++++++++ 1 file changed, 1388 insertions(+) create mode 100644 planning/MARKET_DATA_DESIGN.md 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", +] +``` From 5a5ac828dbdb9207899a935487bfb121b7c784a6 Mon Sep 17 00:00:00 2001 From: jkirkmannz Date: Sat, 12 Sep 2026 16:11:51 +1200 Subject: [PATCH 4/5] Fix market data review findings and add MARKET_DATA_REVIEW.md - Normalize ticker case in MassiveDataSource.start() to match add_ticker/remove_ticker, fixing a silent remove-ticker failure - Derive GBMSimulator dt from SimulatorDataSource.update_interval instead of hardcoding a 500ms assumption - Build stream.py's APIRouter inside create_stream_router() instead of at module scope, avoiding double route registration on repeat calls - Remove deprecated no-op event_loop_policy fixture from conftest.py - Add regression tests for the ticker-normalization and dt-scaling fixes - Apply ruff format to affected files Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_011jXykWYaYoRkBpojjASfru --- backend/app/market/massive_client.py | 2 +- backend/app/market/simulator.py | 5 + backend/app/market/stream.py | 5 +- backend/tests/conftest.py | 10 -- backend/tests/market/test_massive.py | 21 +++++ backend/tests/market/test_models.py | 44 ++++++--- backend/tests/market/test_simulator.py | 4 +- backend/tests/market/test_simulator_source.py | 23 ++++- planning/MARKET_DATA_REVIEW.md | 91 +++++++++++++++++++ 9 files changed, 175 insertions(+), 30 deletions(-) create mode 100644 planning/MARKET_DATA_REVIEW.md 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/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_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..02f7f8a89 100644 --- a/backend/tests/market/test_simulator.py +++ b/backend/tests/market/test_simulator.py @@ -126,6 +126,6 @@ def test_prices_rounded_to_two_decimals(self): result = sim.step() price_str = str(result["AAPL"]) # Check that we have at most 2 decimal places - if '.' in price_str: - decimal_part = price_str.split('.')[1] + if "." in price_str: + decimal_part = price_str.split(".")[1] assert len(decimal_part) <= 2 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/planning/MARKET_DATA_REVIEW.md b/planning/MARKET_DATA_REVIEW.md new file mode 100644 index 000000000..830db8ea6 --- /dev/null +++ b/planning/MARKET_DATA_REVIEW.md @@ -0,0 +1,91 @@ +# Market Data Backend — Code Review + +**Date:** 2026-09-12 +**Reviewer:** Claude +**Scope:** `backend/app/market/` (8 source modules) and `backend/tests/market/` (6 test files, 75 tests) + +This is the second review pass. The first pass (findings below, in §1) surfaced six issues; all six have been fixed in this pass, along with two new regression tests locking in the two behavioral fixes. This document supersedes the review that was in this same file before those fixes; the original findings are preserved in `planning/archive/` for history. + +--- + +## 1. Fixes Applied This 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 (reproduced: `remove_ticker("aapl")` silently failed to remove `"aapl"` when `start()` had stored it un-normalized). | Medium | `start()` now does `[t.upper().strip() for t in tickers]`, matching `add_ticker`/`remove_ticker`. Added `test_start_normalizes_ticker_case` to lock this in. | +| 2 | `GBMSimulator`'s `dt` was hardcoded to assume a 500ms tick regardless of `SimulatorDataSource.update_interval` — changing the interval would silently scale the effective annualized volatility with no warning. | Medium | `SimulatorDataSource.start()` now derives `dt = self._interval / GBMSimulator.TRADING_SECONDS_PER_YEAR` and passes it explicitly. Added `test_dt_scales_with_update_interval` to lock this in. | +| 3 | `stream.py` built its `APIRouter` at module scope and decorated onto it inside `create_stream_router()` — calling the factory twice (e.g., across tests, or a future reload path) would double-register `/prices` on the same shared router. | Low | `router = APIRouter(...)` moved inside `create_stream_router()`, so each call gets its own router. | +| 4 | `PriceCache.version` was read without `self._lock`, unlike every other accessor. | Low | **Not changed** — re-assessed as not worth it; see §5. | +| 5 | `tests/conftest.py` defined an `event_loop_policy` fixture that just returned the default policy, contributing a `DeprecationWarning` on every async test run (`asyncio.DefaultEventLoopPolicy` is deprecated, slated for removal in Python 3.16). | Trivial | Fixture removed; `conftest.py` is now just the module docstring. | +| 6 | `ruff format --check` flagged 3 test files as not matching the formatter's line-wrapping. | Trivial | Ran `ruff format` on the affected files (and once more on `stream.py` after edit #3 left it missing a blank line). | + +Item 4 is intentionally left as-is — see §5 for reasoning. + +--- + +## 2. Test Results (after fixes) + +**75 tests collected, 75 passed, 0 failed.** (`uv run pytest -q --cov=app --cov-report=term-missing`, with `massive==2.2.0` actually installed via `uv sync --extra dev`.) + +Two tests were added this pass (73 → 75): `test_start_normalizes_ticker_case` (massive) and `test_dt_scales_with_update_interval` (simulator source). No `DeprecationWarning`s remain in the run. + +| 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 (L85-87), real (unmocked) `_fetch_snapshots` body (L125) | +| stream.py | 31% | Still untested — needs an ASGI test client, not added this pass (see §5) | +| **Total** | **91%** | Unchanged from before — the new tests cover previously-untested *behavior*, not previously-uncovered *lines* | + +**Lint:** `ruff check app/ tests/` — clean. `ruff check --select F401` (unused imports) — clean. + +**Format:** `ruff format --check app/ tests/` — clean, all 19 files formatted (was 3 files + `stream.py` dirty before this pass). + +--- + +## 3. Architecture Assessment + +Unchanged from the prior review: this remains a clean strategy-pattern implementation (`MarketDataSource` ABC → `SimulatorDataSource` / `MassiveDataSource`) writing into a single shared, thread-safe `PriceCache`, matching `planning/PLAN.md` §6. The fixes in this pass were surgical — no structural changes, no new modules, no altered public API shapes. `GBMSimulator.__init__` already accepted a `dt` parameter; the fix just wires the caller to pass the right value instead of relying on the default. + +--- + +## 4. Re-Verification of Everything From the Original (2026-02-10) Review + +For completeness, since this file has now gone through two review passes beyond the original: + +| Original finding | Status | +|---|---| +| `pyproject.toml` missing wheel packaging config | Fixed (prior pass) | +| Massive tests fragile without `massive` installed | Fixed (prior pass) — confirmed again this pass, ran with `massive` installed | +| `_generate_events` return type annotation | Fixed (prior pass) | +| `PriceCache.version` not under lock | Still open — see §5, deliberately not fixed | +| `SimulatorDataSource.get_tickers` accessed private state | Fixed (prior pass) | +| Module-level `router` in `stream.py` | **Fixed this pass** | +| Unused imports in tests | Fixed (prior pass) | +| Missing SSE integration test | Still open — see §5 | +| No `PriceCache` concurrency test | Still open — see §5 | +| No full-10-ticker `GBMSimulator` test | Still open, but manually verified working (no numerical issue) | +| `DEFAULT_CORR` vs `CROSS_GROUP_CORR` naming | Fixed (prior pass) | + +--- + +## 5. Remaining Open Items (Deliberately Not Fixed) + +None of these are regressions or newly discovered problems — they're the same low-priority items from before, re-assessed and left open on purpose: + +- **`PriceCache.version` unlocked read.** A single `int` read is atomic under CPython's GIL, and this project targets standard CPython, not a no-GIL build. Adding a lock here would be defensive code against a scenario that doesn't apply; not worth the (tiny) overhead or the inconsistency of "sometimes we lock, sometimes the JIT/GIL makes it safe anyway." +- **No SSE integration test for `stream.py`.** Testing this properly needs an ASGI test client (`httpx` + `ASGITransport`, or FastAPI's `TestClient`), which isn't currently a dependency, and there's no FastAPI `app` yet to mount the router into — that's the next phase of backend work, not this one. Adding a dependency and a test harness for a router that isn't wired into a real app yet would be premature; revisit once `main.py`/the FastAPI app exists. +- **No `PriceCache` concurrent-writers test.** The lock usage is straightforward (one `Lock`, held for the full duration of every method) and inspection gives high confidence it's correct. A multi-threaded stress test would mostly be testing Python's `threading.Lock`, not this code's logic. +- **No test with the full 10-ticker default set.** Manually verified this pass (again) that `GBMSimulator(tickers=list(SEED_PRICES.keys()))` builds a valid 10×10 Cholesky decomposition and steps cleanly. Still a coverage gap worth closing eventually, but it's a "nice to have," not a bug. + +--- + +## 6. Verdict + +All six issues identified in the first review pass are fixed, verified by a clean 75-test run, clean lint, and clean format check. The two behavioral fixes (ticker-case normalization, `dt`/`update_interval` coupling) now have dedicated regression tests so they can't silently regress. Nothing here blocks moving on to the rest of the backend (portfolio, watchlist, chat, and the FastAPI app that will actually mount `create_stream_router()`). + +No further action needed on the market data subsystem before that next phase begins. From b3257cbea82d0eb8befe58545d826d0e97d6783e Mon Sep 17 00:00:00 2001 From: jkirkmannz Date: Sat, 12 Sep 2026 16:28:33 +1200 Subject: [PATCH 5/5] Add remaining market data test coverage from review follow-ups - Add SSE integration tests for stream.py by driving _generate_events directly with a fake Request, since both httpx.ASGITransport and FastAPI's TestClient deadlock against this endpoint's unbounded while-True/disconnect-driven loop (confirmed via faulthandler). Coverage for stream.py goes from 31% to 94%. - Add a PriceCache concurrent-writers test (8 threads x 200 updates) asserting the version counter exactly matches, to catch lost updates under real thread interleaving. - Add a full 10-ticker GBMSimulator test covering the mixed tech/finance/TSLA correlation matrix, not just the 1-2 ticker cases. - Add httpx as a dev dependency for ASGI testing. Overall coverage: 91% -> 97%, 84/84 tests passing. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_011jXykWYaYoRkBpojjASfru --- backend/pyproject.toml | 1 + backend/tests/market/test_cache.py | 28 +++++ backend/tests/market/test_simulator.py | 17 +++ backend/tests/market/test_stream.py | 167 +++++++++++++++++++++++++ backend/uv.lock | 30 +++++ planning/MARKET_DATA_REVIEW.md | 108 ++++++++-------- 6 files changed, 295 insertions(+), 56 deletions(-) create mode 100644 backend/tests/market/test_stream.py 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/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_simulator.py b/backend/tests/market/test_simulator.py index 02f7f8a89..75d1fe870 100644 --- a/backend/tests/market/test_simulator.py +++ b/backend/tests/market/test_simulator.py @@ -129,3 +129,20 @@ def test_prices_rounded_to_two_decimals(self): 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_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_REVIEW.md b/planning/MARKET_DATA_REVIEW.md index 830db8ea6..145c4ee46 100644 --- a/planning/MARKET_DATA_REVIEW.md +++ b/planning/MARKET_DATA_REVIEW.md @@ -2,90 +2,86 @@ **Date:** 2026-09-12 **Reviewer:** Claude -**Scope:** `backend/app/market/` (8 source modules) and `backend/tests/market/` (6 test files, 75 tests) +**Scope:** `backend/app/market/` (8 source modules) and `backend/tests/market/` (7 test files, 84 tests) -This is the second review pass. The first pass (findings below, in §1) surfaced six issues; all six have been fixed in this pass, along with two new regression tests locking in the two behavioral fixes. This document supersedes the review that was in this same file before those fixes; the original findings are preserved in `planning/archive/` for history. +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. Fixes Applied This Pass +## 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 (reproduced: `remove_ticker("aapl")` silently failed to remove `"aapl"` when `start()` had stored it un-normalized). | Medium | `start()` now does `[t.upper().strip() for t in tickers]`, matching `add_ticker`/`remove_ticker`. Added `test_start_normalizes_ticker_case` to lock this in. | -| 2 | `GBMSimulator`'s `dt` was hardcoded to assume a 500ms tick regardless of `SimulatorDataSource.update_interval` — changing the interval would silently scale the effective annualized volatility with no warning. | Medium | `SimulatorDataSource.start()` now derives `dt = self._interval / GBMSimulator.TRADING_SECONDS_PER_YEAR` and passes it explicitly. Added `test_dt_scales_with_update_interval` to lock this in. | -| 3 | `stream.py` built its `APIRouter` at module scope and decorated onto it inside `create_stream_router()` — calling the factory twice (e.g., across tests, or a future reload path) would double-register `/prices` on the same shared router. | Low | `router = APIRouter(...)` moved inside `create_stream_router()`, so each call gets its own router. | -| 4 | `PriceCache.version` was read without `self._lock`, unlike every other accessor. | Low | **Not changed** — re-assessed as not worth it; see §5. | -| 5 | `tests/conftest.py` defined an `event_loop_policy` fixture that just returned the default policy, contributing a `DeprecationWarning` on every async test run (`asyncio.DefaultEventLoopPolicy` is deprecated, slated for removal in Python 3.16). | Trivial | Fixture removed; `conftest.py` is now just the module docstring. | -| 6 | `ruff format --check` flagged 3 test files as not matching the formatter's line-wrapping. | Trivial | Ran `ruff format` on the affected files (and once more on `stream.py` after edit #3 left it missing a blank line). | - -Item 4 is intentionally left as-is — see §5 for reasoning. +| 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. Test Results (after fixes) - -**75 tests collected, 75 passed, 0 failed.** (`uv run pytest -q --cov=app --cov-report=term-missing`, with `massive==2.2.0` actually installed via `uv sync --extra dev`.) +## 2. Improvements Added (This Pass) -Two tests were added this pass (73 → 75): `test_start_normalizes_ticker_case` (massive) and `test_dt_scales_with_update_interval` (simulator source). No `DeprecationWarning`s remain in the run. +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: -| 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 (L85-87), real (unmocked) `_fetch_snapshots` body (L125) | -| stream.py | 31% | Still untested — needs an ASGI test client, not added this pass (see §5) | -| **Total** | **91%** | Unchanged from before — the new tests cover previously-untested *behavior*, not previously-uncovered *lines* | +### 2.1 SSE integration test for `stream.py` (was 31% coverage, no tests) -**Lint:** `ruff check app/ tests/` — clean. `ruff check --select F401` (unused imports) — clean. +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. -**Format:** `ruff format --check app/ tests/` — clean, all 19 files formatted (was 3 files + `stream.py` dirty before this pass). +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 -## 3. Architecture Assessment +`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). -Unchanged from the prior review: this remains a clean strategy-pattern implementation (`MarketDataSource` ABC → `SimulatorDataSource` / `MassiveDataSource`) writing into a single shared, thread-safe `PriceCache`, matching `planning/PLAN.md` §6. The fixes in this pass were surgical — no structural changes, no new modules, no altered public API shapes. `GBMSimulator.__init__` already accepted a `dt` parameter; the fix just wires the caller to pass the right value instead of relying on the default. +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 -## 4. Re-Verification of Everything From the Original (2026-02-10) Review +`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. -For completeness, since this file has now gone through two review passes beyond the original: +### 2.3 Full 10-ticker `GBMSimulator` test -| Original finding | Status | -|---|---| -| `pyproject.toml` missing wheel packaging config | Fixed (prior pass) | -| Massive tests fragile without `massive` installed | Fixed (prior pass) — confirmed again this pass, ran with `massive` installed | -| `_generate_events` return type annotation | Fixed (prior pass) | -| `PriceCache.version` not under lock | Still open — see §5, deliberately not fixed | -| `SimulatorDataSource.get_tickers` accessed private state | Fixed (prior pass) | -| Module-level `router` in `stream.py` | **Fixed this pass** | -| Unused imports in tests | Fixed (prior pass) | -| Missing SSE integration test | Still open — see §5 | -| No `PriceCache` concurrency test | Still open — see §5 | -| No full-10-ticker `GBMSimulator` test | Still open, but manually verified working (no numerical issue) | -| `DEFAULT_CORR` vs `CROSS_GROUP_CORR` naming | Fixed (prior pass) | +`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. --- -## 5. Remaining Open Items (Deliberately Not Fixed) +## 3. Test Results (Final) -None of these are regressions or newly discovered problems — they're the same low-priority items from before, re-assessed and left open on purpose: +**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. -- **`PriceCache.version` unlocked read.** A single `int` read is atomic under CPython's GIL, and this project targets standard CPython, not a no-GIL build. Adding a lock here would be defensive code against a scenario that doesn't apply; not worth the (tiny) overhead or the inconsistency of "sometimes we lock, sometimes the JIT/GIL makes it safe anyway." -- **No SSE integration test for `stream.py`.** Testing this properly needs an ASGI test client (`httpx` + `ASGITransport`, or FastAPI's `TestClient`), which isn't currently a dependency, and there's no FastAPI `app` yet to mount the router into — that's the next phase of backend work, not this one. Adding a dependency and a test harness for a router that isn't wired into a real app yet would be premature; revisit once `main.py`/the FastAPI app exists. -- **No `PriceCache` concurrent-writers test.** The lock usage is straightforward (one `Lock`, held for the full duration of every method) and inspection gives high confidence it's correct. A multi-threaded stress test would mostly be testing Python's `threading.Lock`, not this code's logic. -- **No test with the full 10-ticker default set.** Manually verified this pass (again) that `GBMSimulator(tickers=list(SEED_PRICES.keys()))` builds a valid 10×10 Cholesky decomposition and steps cleanly. Still a coverage gap worth closing eventually, but it's a "nice to have," not a bug. +| 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. --- -## 6. Verdict +## 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. + +--- -All six issues identified in the first review pass are fixed, verified by a clean 75-test run, clean lint, and clean format check. The two behavioral fixes (ticker-case normalization, `dt`/`update_interval` coupling) now have dedicated regression tests so they can't silently regress. Nothing here blocks moving on to the rest of the backend (portfolio, watchlist, chat, and the FastAPI app that will actually mount `create_stream_router()`). +## 5. Verdict -No further action needed on the market data subsystem before that next phase begins. +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.