From e3da259cc169697cff71998409f7eba675fb6604 Mon Sep 17 00:00:00 2001 From: Rahul Sahu Date: Sat, 13 Jun 2026 16:59:52 +0400 Subject: [PATCH 1/2] add new files to git --- planning/MARKET_DATA_DESIGN.md | 896 +++++++++++++++++++++++++++++++++ planning/MARKET_INTERFACE.md | 210 ++++++++ planning/MARKET_SIMULATOR.md | 162 ++++++ planning/MASSIVE_API.md | 174 +++++++ 4 files changed, 1442 insertions(+) create mode 100644 planning/MARKET_DATA_DESIGN.md create mode 100644 planning/MARKET_INTERFACE.md create mode 100644 planning/MARKET_SIMULATOR.md create mode 100644 planning/MASSIVE_API.md diff --git a/planning/MARKET_DATA_DESIGN.md b/planning/MARKET_DATA_DESIGN.md new file mode 100644 index 00000000..ba96001b --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,896 @@ +# Market Data Backend — Detailed Design + +A complete, implementation-ready design for the FinAlly market data subsystem. +It covers the unified interface, the GBM simulator, the Massive (Polygon.io) REST +client, the shared price cache, the factory, and the SSE streaming endpoint, with +full code for every module. + +Companion documents: +- [MARKET_INTERFACE.md](MARKET_INTERFACE.md) — the unified-API design rationale +- [MARKET_SIMULATOR.md](MARKET_SIMULATOR.md) — the simulation approach +- [MASSIVE_API.md](MASSIVE_API.md) — the Massive API reference +- [PLAN.md](PLAN.md) §6 — product requirements + +## 1. Overview + +``` + create_market_data_source(cache) ← reads MASSIVE_API_KEY + │ + ┌───────────────┴────────────────┐ + ▼ ▼ + SimulatorDataSource MassiveDataSource + GBM, ~500ms ticks REST poll, 15s default + (default, no key) (when key is set) + │ │ + └───────────────┬────────────────┘ + ▼ writes + PriceCache ← single source of truth + thread-safe, versioned + │ reads + ┌───────────────┼────────────────┐ + ▼ ▼ ▼ + SSE /api/stream/ portfolio val. trade execution + prices +``` + +**Core principle:** producers (sources) *write* into the cache on their own +schedule; consumers *read* from the cache. They are fully decoupled — neither +the SSE endpoint nor portfolio code ever touches a data source directly. + +### Module layout (`backend/app/market/`) + +| Module | Responsibility | +|--------|----------------| +| `models.py` | `PriceUpdate` immutable dataclass | +| `cache.py` | `PriceCache` thread-safe versioned store | +| `interface.py` | `MarketDataSource` ABC | +| `seed_prices.py` | seed prices, GBM params, correlation config | +| `simulator.py` | `GBMSimulator` + `SimulatorDataSource` | +| `massive_client.py` | `MassiveDataSource` REST poller | +| `factory.py` | `create_market_data_source()` | +| `stream.py` | `create_stream_router()` SSE endpoint | +| `__init__.py` | public exports | + +### Dependencies + +```toml +# pyproject.toml +dependencies = [ + "fastapi", + "numpy", # Cholesky + vectorized normal draws in the simulator + "massive", # Massive (Polygon.io) Python client +] + +[tool.hatch.build.targets.wheel] +packages = ["app"] +``` + +## 2. `models.py` — PriceUpdate + +The unit of data flowing through the system. Immutable, frozen, slotted. + +```python +"""Data models for market data.""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass(frozen=True, slots=True) +class PriceUpdate: + """Immutable snapshot of a single ticker's price at a point in time.""" + + ticker: str + price: float + previous_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: + """Absolute price change from previous update.""" + return round(self.price - self.previous_price, 4) + + @property + def change_percent(self) -> float: + """Percentage change from previous update.""" + if self.previous_price == 0: + return 0.0 + return round((self.price - self.previous_price) / self.previous_price * 100, 4) + + @property + def direction(self) -> str: + """'up', 'down', or 'flat' — drives the frontend price-flash.""" + 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 notes: +- `change`/`change_percent`/`direction` are derived properties, not stored — one + place defines them, both sources get identical semantics. +- `frozen=True` makes instances safe to share across threads (the Massive poller + runs in a worker thread; the SSE reader runs on the loop). + +## 3. `cache.py` — PriceCache + +The seam between producers and consumers. Thread-safe because the Massive client +writes from `asyncio.to_thread` while readers run on the event loop. + +```python +"""Thread-safe in-memory price cache.""" + +from __future__ import annotations + +import time +from threading import Lock + +from .models import PriceUpdate + + +class PriceCache: + """Thread-safe in-memory cache of the latest price for each ticker. + + Writers: SimulatorDataSource or MassiveDataSource (one at a time). + Readers: SSE streaming endpoint, portfolio valuation, trade execution. + """ + + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._lock = Lock() + self._version: int = 0 # Monotonically increasing; bumped on every update + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + """Record a new price. Computes previous_price/change/direction. + + On the first update for a 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: + with self._lock: + return self._prices.get(ticker) + + def get_all(self) -> dict[str, PriceUpdate]: + """Shallow copy snapshot of all current prices.""" + with self._lock: + return dict(self._prices) + + def get_price(self, ticker: str) -> float | None: + update = self.get(ticker) + return update.price if update else None + + def remove(self, ticker: str) -> None: + with self._lock: + self._prices.pop(ticker, None) + + @property + def version(self) -> int: + """Monotonic counter, bumped on every update. 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 +``` + +The `version` counter is the trick that lets the SSE endpoint detect "something +changed" in O(1) without diffing every ticker. + +## 4. `interface.py` — MarketDataSource + +The contract. Both sources implement it identically; downstream depends only on +this ABC. + +```python +"""Abstract interface for market data sources.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class MarketDataSource(ABC): + """Contract for market data providers. + + Implementations push price updates into a shared PriceCache on their own + schedule. Downstream code reads from the cache, never from the source. + + Lifecycle: + source = create_market_data_source(cache) + await source.start(["AAPL", "GOOGL", ...]) + await source.add_ticker("TSLA") + await source.remove_ticker("GOOGL") + await source.stop() + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing updates. Starts a background task. Call once.""" + + @abstractmethod + async def stop(self) -> None: + """Stop the task, release resources. Safe to call repeatedly.""" + + @abstractmethod + async def add_ticker(self, ticker: str) -> None: + """Add a ticker to the active set. No-op if already present.""" + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Remove a ticker; also drop it from the cache. No-op if absent.""" + + @abstractmethod + def get_tickers(self) -> list[str]: + """Return the currently tracked tickers.""" +``` + +## 5. `seed_prices.py` — simulator configuration + +```python +"""Seed prices and per-ticker parameters for the market simulator.""" + +# Realistic starting prices for the default watchlist +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 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}, +} + +# Fallback for dynamically added tickers not listed above +DEFAULT_PARAMS: dict[str, float] = {"sigma": 0.25, "mu": 0.05} + +# Sector membership for Cholesky 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 +``` + +## 6. `simulator.py` — GBM Simulator + +Two classes: `GBMSimulator` (pure math, synchronous, unit-testable) and +`SimulatorDataSource` (asyncio lifecycle, writes to cache). See +[MARKET_SIMULATOR.md](MARKET_SIMULATOR.md) for the model derivation. + +### 6.1 GBMSimulator — the math engine + +```python +"""GBM-based market simulator.""" + +from __future__ import annotations + +import logging +import math +import random + +import numpy as np + +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. + + S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) + + Z is a *correlated* standard-normal draw (Cholesky of the sector matrix). + """ + + # 500ms as a fraction of a trading year: + # 252 trading days * 6.5 hours * 3600 sec = 5,896,800 sec + TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 + DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8 + + def __init__(self, tickers, dt=DEFAULT_DT, event_probability=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() + + def step(self) -> dict[str, float]: + """Advance all tickers one tick. Returns {ticker: new_price}. Hot path.""" + n = len(self._tickers) + if n == 0: + return {} + + z = np.random.standard_normal(n) + if self._cholesky is not None: + z = self._cholesky @ z # correlate the draws + + result: dict[str, float] = {} + for i, ticker in enumerate(self._tickers): + p = self._params[ticker] + mu, sigma = p["mu"], p["sigma"] + + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z[i] + self._prices[ticker] *= math.exp(drift + diffusion) + + # Rare shock event for visual drama (~0.1%/tick/ticker) + if random.random() < self._event_prob: + shock = random.uniform(0.02, 0.05) * random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock + logger.debug("Event on %s: %.1f%%", ticker, shock * 100) + + result[ticker] = round(self._prices[ticker], 2) + return result + + def add_ticker(self, ticker: str) -> None: + if ticker in self._prices: + return + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + def remove_ticker(self, ticker: str) -> None: + 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: + return self._prices.get(ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + # --- internals --- + + def _add_ticker_internal(self, ticker: str) -> None: + 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 Cholesky of the correlation matrix. O(n^2), 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: + tech = CORRELATION_GROUPS["tech"] + finance = CORRELATION_GROUPS["finance"] + if t1 == "TSLA" or t2 == "TSLA": + return TSLA_CORR + if t1 in tech and t2 in tech: + return INTRA_TECH_CORR + if t1 in finance and t2 in finance: + return INTRA_FINANCE_CORR + return CROSS_GROUP_CORR +``` + +### 6.2 SimulatorDataSource — the lifecycle wrapper + +```python +import asyncio + +from .cache import PriceCache +from .interface import MarketDataSource + + +class SimulatorDataSource(MarketDataSource): + """MarketDataSource backed by the GBM simulator. + + Runs a background asyncio task that steps the simulation every + `update_interval` seconds and writes results to the PriceCache. + """ + + def __init__(self, price_cache: PriceCache, update_interval=0.5, event_probability=0.001): + 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 immediately so SSE has data on first connect + 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) + + async def remove_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.remove_ticker(ticker) + self._cache.remove(ticker) + + def get_tickers(self) -> list[str]: + return self._sim.get_tickers() if self._sim else [] + + async def _run_loop(self) -> None: + while True: + try: + if self._sim: + for ticker, price in self._sim.step().items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +## 7. `massive_client.py` — Massive REST Poller + +Polls the full-snapshot endpoint for all tickers in **one** request (free-tier +safe). The `RESTClient` is synchronous, so the HTTP call runs in a worker thread. +See [MASSIVE_API.md](MASSIVE_API.md) for endpoint/field details. + +```python +"""Massive (Polygon.io) API client for real market data.""" + +from __future__ import annotations + +import asyncio +import logging + +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +from .cache import PriceCache +from .interface import MarketDataSource + +logger = logging.getLogger(__name__) + + +class MassiveDataSource(MarketDataSource): + """MarketDataSource backed by the Massive (Polygon.io) REST API. + + Polls GET /v2/snapshot/locale/us/markets/stocks/tickers for all watched + tickers in a single call, then writes results to the PriceCache. + + Rate limits: + - Free tier: 5 req/min -> poll every 15s (default) + - Paid tiers: higher -> poll every 2-5s + """ + + def __init__(self, api_key: str, price_cache: PriceCache, poll_interval: float = 15.0): + 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) + await self._poll_once() # immediate first poll so cache has data + 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) + + 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) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + # --- internals --- + + async def _poll_loop(self) -> None: + while True: + await asyncio.sleep(self._interval) + await self._poll_once() + + async def _poll_once(self) -> None: + if not self._tickers or not self._client: + return + try: + # RESTClient is synchronous -> run off the event loop + snapshots = await asyncio.to_thread(self._fetch_snapshots) + for snap in snapshots: + try: + self._cache.update( + ticker=snap.ticker, + price=snap.last_trade.price, + timestamp=snap.last_trade.timestamp / 1000.0, # ms -> s + ) + except (AttributeError, TypeError) as e: + logger.warning("Skipping snapshot for %s: %s", + getattr(snap, "ticker", "???"), e) + except Exception as e: + # Never crash the loop: 401 (bad key), 429 (rate limit), network errors + logger.error("Massive poll failed: %s", e) + + def _fetch_snapshots(self) -> list: + return self._client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=self._tickers, + ) +``` + +Resilience contract: a transient failure logs and retries on the next interval; +a single malformed snapshot is skipped without discarding the batch. + +## 8. `factory.py` — Source Selection + +The only module that reads the env var. Everything else depends on the ABC. + +```python +"""Factory for creating market data sources.""" + +from __future__ import annotations + +import logging +import os + +from .cache import PriceCache +from .interface import MarketDataSource +from .massive_client import MassiveDataSource +from .simulator import SimulatorDataSource + +logger = logging.getLogger(__name__) + + +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """Return MassiveDataSource if MASSIVE_API_KEY is set, else SimulatorDataSource. + + 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) + logger.info("Market data source: GBM Simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +## 9. `stream.py` — SSE Endpoint + +Streams all prices every ~500ms, but only serializes when the cache `version` +changed since the last send. Uses `EventSource`-friendly headers and a retry +directive so the browser auto-reconnects. + +```python +"""SSE streaming endpoint for live price updates.""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import AsyncGenerator + +from fastapi import APIRouter, Request +from fastapi.responses import StreamingResponse + +from .cache import PriceCache + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/stream", tags=["streaming"]) + + +def create_stream_router(price_cache: PriceCache) -> APIRouter: + """Create the SSE router with the price cache injected (no globals).""" + + @router.get("/prices") + async def stream_prices(request: Request) -> StreamingResponse: + 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]: + """Yield SSE-formatted price events; stop on client disconnect.""" + yield "retry: 1000\n\n" # browser retries after 1s if dropped + + 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 + + if price_cache.version != last_version: + last_version = price_cache.version + prices = price_cache.get_all() + if prices: + data = {t: u.to_dict() for t, u 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) +``` + +Event payload shape (one event, all tickers): + +``` +data: {"AAPL": {"ticker": "AAPL", "price": 190.50, "previous_price": 190.42, + "timestamp": 1765600000.0, "change": 0.08, "change_percent": 0.042, + "direction": "up"}, "GOOGL": {...}, ...} +``` + +## 10. `__init__.py` — Public API + +```python +"""Market data subsystem for FinAlly.""" + +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", +] +``` + +## 11. Wiring Into FastAPI + +Create the cache and source once at startup; mount the SSE router. Use the +lifespan context so the background task starts and stops cleanly. + +```python +from contextlib import asynccontextmanager +from fastapi import FastAPI + +from app.market import PriceCache, create_market_data_source, create_stream_router + +DEFAULT_WATCHLIST = ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA", + "NVDA", "META", "JPM", "V", "NFLX"] + + +@asynccontextmanager +async def lifespan(app: FastAPI): + cache = PriceCache() + source = create_market_data_source(cache) + await source.start(DEFAULT_WATCHLIST) + + app.state.price_cache = cache + app.state.market_source = source + try: + yield + finally: + await source.stop() + + +app = FastAPI(lifespan=lifespan) + + +@asynccontextmanager +async def _noop(): # placeholder; router needs the cache from state + yield + + +# Mount SSE — build the router after the cache exists. +# Simplest: build cache module-level, or attach in lifespan and include here. +``` + +A clean pattern is to build the cache/source in a small module-level singleton +so the router can be created at import time: + +```python +# app/state.py +from app.market import PriceCache, create_market_data_source + +price_cache = PriceCache() +market_source = create_market_data_source(price_cache) + +# app/main.py +from app.state import price_cache, market_source +from app.market import create_stream_router + +app.include_router(create_stream_router(price_cache)) + +@asynccontextmanager +async def lifespan(app): + await market_source.start(DEFAULT_WATCHLIST) + try: + yield + finally: + await market_source.stop() +``` + +### Consuming prices elsewhere (portfolio, trades) + +```python +from app.state import price_cache + +def position_value(ticker: str, quantity: float) -> float | None: + price = price_cache.get_price(ticker) + return price * quantity if price is not None else None +``` + +### Watchlist add/remove must hit both the source and the DB + +```python +async def add_to_watchlist(ticker: str): + await market_source.add_ticker(ticker) # start streaming it + # ... insert into watchlist table ... + +async def remove_from_watchlist(ticker: str): + await market_source.remove_ticker(ticker) # stop streaming + cache evict + # ... delete from watchlist table ... +``` + +## 12. Testing Strategy + +Unit tests live in `backend/tests/market/`. Run with `uv run --extra dev pytest`. + +| Target | What to assert | +|--------|----------------| +| `models.py` | `change`/`change_percent`/`direction` for up/down/flat; zero-prev guard; `to_dict` keys | +| `cache.py` | first update sets prev==price; version increments; `remove`; thread-safety under concurrent writes | +| `simulator.py` | prices stay positive; `step()` returns all tickers; Cholesky rebuild on add/remove; correlation lookups; shock fires when `event_probability=1.0` | +| `SimulatorDataSource` | cache seeded after `start`; loop writes; `stop` cancels cleanly; add/remove reflect in cache | +| `massive_client.py` | mock `RESTClient.get_snapshot_all`; assert ms→s conversion; malformed snapshot skipped; poll failure swallowed | +| `factory.py` | key set → Massive; unset/empty/whitespace → simulator | + +Example — simulator keeps prices positive and correlated draws applied: + +```python +def test_step_returns_positive_prices(): + sim = GBMSimulator(["AAPL", "GOOGL"], event_probability=0.0) + prices = sim.step() + assert set(prices) == {"AAPL", "GOOGL"} + assert all(p > 0 for p in prices.values()) +``` + +Example — Massive ms→s conversion with a mocked client: + +```python +def test_massive_converts_timestamp(monkeypatch): + cache = PriceCache() + src = MassiveDataSource("key", cache) + fake = SimpleNamespace(ticker="AAPL", + last_trade=SimpleNamespace(price=190.0, timestamp=1_765_600_000_000)) + src._client = SimpleNamespace(get_snapshot_all=lambda **kw: [fake]) + src._tickers = ["AAPL"] + asyncio.run(src._poll_once()) + upd = cache.get("AAPL") + assert upd.price == 190.0 + assert upd.timestamp == 1_765_600_000.0 +``` + +E2E (Playwright, `test/`): with `LLM_MOCK=true` and no `MASSIVE_API_KEY`, verify +the default watchlist streams prices, prices flash on change, and the SSE +connection reconnects after a drop. + +## 13. Design Decisions Recap + +| Decision | Rationale | +|----------|-----------| +| ABC + factory | Swap source via env var; consumers untouched | +| Cache as the seam | Producers/consumers decoupled; multi-user ready | +| `version` counter | O(1) SSE change detection, no per-ticker diff | +| Push into cache (not pull) | One model for 500ms sim and 15s REST poll | +| Single snapshot call | All tickers in one request → free-tier safe | +| `asyncio.to_thread` for Massive | Sync client never blocks the event loop | +| GBM + Cholesky | Realistic, positive, sector-correlated prices | +| Broad try/except in pollers | Transient API errors never crash the app | +| Split GBMSimulator vs DataSource | Math unit-tested without asyncio | +``` diff --git a/planning/MARKET_INTERFACE.md b/planning/MARKET_INTERFACE.md new file mode 100644 index 00000000..9879d351 --- /dev/null +++ b/planning/MARKET_INTERFACE.md @@ -0,0 +1,210 @@ +# Unified Market Data Interface + +The design of the single Python API the project uses to retrieve stock prices. +It selects the **Massive API** when `MASSIVE_API_KEY` is set, otherwise the +**simulator** ([MARKET_SIMULATOR.md](MARKET_SIMULATOR.md)). All downstream code +(SSE streaming, portfolio valuation, trade execution) is agnostic to the source. + +Grounded in the real-data shape documented in [MASSIVE_API.md](MASSIVE_API.md). + +## Design Goals + +- **One contract, two implementations** — swap data sources via an env var with + zero downstream changes (Strategy pattern). +- **Decoupled producers and consumers** — the data source *writes* prices into a + shared cache; consumers *read* from the cache. They never call each other. +- **Push-friendly** — the cache carries a version counter so the SSE endpoint can + detect changes cheaply without polling each ticker. +- **Single API call for live data** — the Massive poller fetches all tickers in + one snapshot request to respect the free-tier rate limit. + +## Architecture + +``` + create_market_data_source(cache) ← reads MASSIVE_API_KEY + │ + ┌──────────────┴───────────────┐ + ▼ ▼ +SimulatorDataSource MassiveDataSource + (GBM, ~500ms ticks) (REST poll, 15s default) + │ │ + └──────────────┬───────────────┘ + ▼ + PriceCache ← single source of truth + (thread-safe) + │ + ┌──────────────┼───────────────┐ + ▼ ▼ ▼ + SSE stream Portfolio val. Trade execution +``` + +## The Contract: `MarketDataSource` + +An abstract base class. Both sources implement it identically. + +```python +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 reads from the cache, never from the source. + + Lifecycle: + source = create_market_data_source(cache) + await source.start(["AAPL", "GOOGL", ...]) + await source.add_ticker("TSLA") + await source.remove_ticker("GOOGL") + await source.stop() + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Start the background producer task. Call exactly once.""" + + @abstractmethod + async def stop(self) -> None: + """Stop the task and release resources. Safe to call repeatedly.""" + + @abstractmethod + async def add_ticker(self, ticker: str) -> None: + """Add a ticker to the active set. No-op if already present.""" + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Remove a ticker; also drop it from the cache. No-op if absent.""" + + @abstractmethod + def get_tickers(self) -> list[str]: + """Return the currently tracked tickers.""" +``` + +Lifecycle rules: +- `start()` does an **immediate first fetch** so the cache has data before any + client connects, then launches the background task. +- `stop()` cancels the task and awaits its `CancelledError`; idempotent. +- `add_ticker` / `remove_ticker` mutate the active set live; the next cycle + reflects the change. Removal also evicts from the cache. + +## `PriceUpdate` — the unit of data + +An immutable frozen dataclass produced by the cache on every update. + +```python +@dataclass(frozen=True) +class PriceUpdate: + ticker: str + price: float + previous_price: float | None + timestamp: float # epoch seconds + + @property + def change(self) -> float: ... # price - previous_price + @property + def change_percent(self) -> float: ... + @property + def direction(self) -> str: ... # "up" | "down" | "flat" + + def to_dict(self) -> dict: ... # JSON for SSE payloads +``` + +## `PriceCache` — single source of truth + +Thread-safe in-memory store. The Massive poller writes from a worker thread; the +simulator writes from the event loop — so the lock matters. + +```python +class PriceCache: + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + """Store a new price, compute previous/change, bump version.""" + + def get(self, ticker: str) -> PriceUpdate | None: ... + def get_price(self, ticker: str) -> float | None: ... + def get_all(self) -> dict[str, PriceUpdate]: ... + def remove(self, ticker: str) -> None: ... + + @property + def version(self) -> int: + """Monotonic counter; increments on every update. SSE change detection.""" +``` + +`update()` carries the prior price into `previous_price`, so price-flash +direction is derived in one place regardless of source. + +## The Factory + +The only place that knows about the env var. Everything else depends on the ABC. + +```python +import os +from .cache import PriceCache +from .interface import MarketDataSource +from .simulator import SimulatorDataSource +from .massive_client import MassiveDataSource + + +def create_market_data_source(cache: PriceCache) -> MarketDataSource: + """Return MassiveDataSource if MASSIVE_API_KEY is set, else the simulator.""" + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + if api_key: + return MassiveDataSource(api_key=api_key, price_cache=cache) + return SimulatorDataSource(price_cache=cache) +``` + +Selection is purely presence-of-key. Empty or whitespace-only is treated as +unset → simulator. + +## Massive Implementation Notes + +`MassiveDataSource` polls the **full snapshot** endpoint for all watched tickers +in one request (see [MASSIVE_API.md](MASSIVE_API.md) §1): + +- Default poll interval **15s** (free tier safe); lower on paid tiers. +- The synchronous `RESTClient` call runs via `asyncio.to_thread` so it never + blocks the loop. +- Each snapshot maps to `cache.update(ticker, price, timestamp/1000.0)`. +- Broad `try/except` around the poll; per-snapshot parse errors skip just that + ticker. The loop always retries on the next interval — a transient 429/network + error never crashes the app. + +## Simulator Implementation Notes + +`SimulatorDataSource` steps a GBM model every ~500ms and writes all prices to the +cache. No external dependencies. Full approach in +[MARKET_SIMULATOR.md](MARKET_SIMULATOR.md). + +## Usage From The App + +```python +from app.market import PriceCache, create_market_data_source + +# Startup +cache = PriceCache() +source = create_market_data_source(cache) # picks source from env +await source.start(["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA", + "NVDA", "META", "JPM", "V", "NFLX"]) + +# Anywhere downstream — read from the cache, not the source +update = cache.get("AAPL") # PriceUpdate | None +price = cache.get_price("AAPL") # float | None +everything = cache.get_all() # dict[str, PriceUpdate] + +# Live watchlist edits +await source.add_ticker("PYPL") +await source.remove_ticker("GOOGL") + +# Shutdown +await source.stop() +``` + +## Why This Shape + +| Choice | Rationale | +|--------|-----------| +| ABC + factory | Add/replace a source without touching consumers | +| Cache as the seam | Producers and consumers fully decoupled; supports multi-user later | +| Version counter | SSE detects "something changed" in O(1), no per-ticker diffing | +| Push into cache (not pull) | Same model for a 500ms simulator and a 15s REST poll | +| Single snapshot call | One request covers all tickers → fits free-tier rate limit | diff --git a/planning/MARKET_SIMULATOR.md b/planning/MARKET_SIMULATOR.md new file mode 100644 index 00000000..88885c00 --- /dev/null +++ b/planning/MARKET_SIMULATOR.md @@ -0,0 +1,162 @@ +# Market Simulator + +The approach and code structure for simulating realistic, correlated stock +prices when no `MASSIVE_API_KEY` is set. It implements the same +`MarketDataSource` contract as the real client (see +[MARKET_INTERFACE.md](MARKET_INTERFACE.md)), so downstream code cannot tell the +difference. + +## Goals + +- **Realistic motion** — prices drift and wobble like real equities, not random + noise. +- **Correlated moves** — tech stocks tend to move together; finance together; + TSLA does its own thing. Makes the heatmap and watchlist feel alive. +- **Visual drama** — occasional sudden 2-5% shocks so the terminal has events to + flash. +- **Zero dependencies** — pure in-process background task; no network, no key. +- **Fast hot path** — runs every ~500ms; the per-tick step must stay cheap. + +## Model: Geometric Brownian Motion (GBM) + +The standard model for stock prices. Each tick advances every ticker by: + +``` +S(t+dt) = S(t) * exp( (mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z ) +``` + +| Symbol | Meaning | +|--------|---------| +| `S(t)` | current price | +| `mu` | annualized drift (expected return) | +| `sigma` | annualized volatility | +| `dt` | time step as a fraction of a trading year | +| `Z` | a *correlated* standard-normal draw | + +### Choosing `dt` + +`dt` is a 500ms tick expressed in trading-years: + +``` +TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # = 5,896,800 +DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ≈ 8.48e-8 +``` + +This tiny `dt` produces sub-cent moves per tick that accumulate naturally into +believable intraday drift — no artificial scaling needed. + +## Correlated Moves via Cholesky + +Independent random draws would make every ticker wander on its own. Real sectors +co-move. We build a correlation matrix and apply its **Cholesky decomposition** to +independent normals to get correlated ones. + +``` +z_independent ~ N(0, I) # n independent draws +z_correlated = L @ z_independent # L = cholesky(correlation_matrix) +``` + +Correlation structure (pairwise): + +| Pair | Correlation | +|------|-------------| +| Same tech sector | 0.6 | +| Same finance sector | 0.5 | +| TSLA with anything | 0.3 (independent-ish) | +| Cross-sector | 0.3 | +| Unknown ticker | 0.3 | + +The matrix is rebuilt only when tickers are added/removed (O(n²), but n < 50, so +negligible). + +## Random Shock Events + +On each tick, each ticker has a ~0.1% chance (`event_probability = 0.001`) of a +sudden jump: + +```python +if random.random() < self._event_prob: + shock = random.uniform(0.02, 0.05) * random.choice([-1, 1]) # ±2–5% + self._prices[ticker] *= (1 + shock) +``` + +With 10 tickers at 2 ticks/sec, expect an event roughly every ~50 seconds — +frequent enough for drama, rare enough to stay realistic. + +## Seed Data + +`seed_prices.py` holds the starting conditions: + +- **`SEED_PRICES`** — realistic opening prices (e.g. AAPL ~190, GOOGL ~175). + Unknown tickers get a random price in `[50, 300]`. +- **`TICKER_PARAMS`** — per-ticker `{"mu", "sigma"}`. Unknown tickers fall back to + `DEFAULT_PARAMS`. +- **`CORRELATION_GROUPS`** — `{"tech": {...}, "finance": {...}}` sector membership. +- Correlation constants: `INTRA_TECH_CORR`, `INTRA_FINANCE_CORR`, `TSLA_CORR`, + `CROSS_GROUP_CORR`. + +## Code Structure + +Two classes, separating the math from the lifecycle. + +### `GBMSimulator` — the math engine + +Pure, synchronous, no asyncio. Owns price state and the correlation matrix. + +```python +class GBMSimulator: + def __init__(self, tickers, dt=DEFAULT_DT, event_probability=0.001): ... + + def step(self) -> dict[str, float]: + """Advance all tickers one tick. Returns {ticker: new_price}. Hot path.""" + + def add_ticker(self, ticker: str) -> None: # seeds price + params, rebuilds Cholesky + def remove_ticker(self, ticker: str) -> None: # drops state, rebuilds Cholesky + def get_price(self, ticker: str) -> float | None: ... + def get_tickers(self) -> list[str]: ... +``` + +`step()` is the hot path: one vectorized `np.random.standard_normal(n)` draw, one +matrix multiply for correlation, then the GBM update per ticker, prices rounded +to cents. + +### `SimulatorDataSource` — the lifecycle wrapper + +Implements `MarketDataSource`. Runs the asyncio loop and writes to the cache. + +```python +class SimulatorDataSource(MarketDataSource): + def __init__(self, price_cache, update_interval=0.5, event_probability=0.001): ... + + async def start(self, tickers): + # build GBMSimulator, seed cache with initial prices, launch _run_loop + async def stop(self): # cancel task, await CancelledError + async def add_ticker(self, ticker): # add to sim, seed cache immediately + async def remove_ticker(self, ticker): # remove from sim and cache + def get_tickers(self): ... + + async def _run_loop(self): + while True: + prices = self._sim.step() + for ticker, price in prices.items(): + self._cache.update(ticker=ticker, price=price) + await asyncio.sleep(self._interval) +``` + +Key behaviors: +- `start()` seeds the cache with initial prices **before** the loop runs, so SSE + has data on the first client connection. +- `add_ticker()` seeds the new ticker's price into the cache immediately rather + than waiting up to 500ms for the next tick. +- The loop wraps `step()` in a `try/except` that logs and continues — one bad + step never kills the simulation. + +## Why GBM + Cholesky (not simpler noise) + +| Choice | Rationale | +|--------|-----------| +| GBM | Industry-standard price model; prices stay positive, moves are multiplicative | +| Annualized `mu`/`sigma` with tiny `dt` | Lets us reason in familiar terms; natural-looking sub-cent ticks | +| Cholesky-correlated draws | Sectors co-move → realistic heatmap and watchlist behavior | +| Rare ±2–5% shocks | Visual events for the terminal without dominating the signal | +| Split math vs. lifecycle | `GBMSimulator` is easily unit-tested in isolation from asyncio | diff --git a/planning/MASSIVE_API.md b/planning/MASSIVE_API.md new file mode 100644 index 00000000..b1f64021 --- /dev/null +++ b/planning/MASSIVE_API.md @@ -0,0 +1,174 @@ +# Massive API (formerly Polygon.io) — Reference + +Research notes and code examples for retrieving real-time and end-of-day (EOD) +stock prices for multiple tickers. This is the source-of-truth research that +drives the design in [MARKET_INTERFACE.md](MARKET_INTERFACE.md). + +## Background + +Polygon.io rebranded to **Massive** (massive.com) on 2025-10-30. Existing API +keys, accounts, and integrations continue to work unchanged. + +- New API base: `https://api.massive.com` +- Legacy base `https://api.polygon.io` is still supported for an extended period +- Official Python client: `massive` (GitHub: `massive-com/client-python`), + requires Python >= 3.9 +- The client is a drop-in successor to the old `polygon-api-client` package + +In this project the key is read from the `MASSIVE_API_KEY` environment variable. +If it is unset or empty, the project uses the built-in simulator instead (see +[MARKET_SIMULATOR.md](MARKET_SIMULATOR.md)). + +## Python Client + +```bash +uv add massive +``` + +```python +from massive import RESTClient + +client = RESTClient(api_key="") +``` + +The `RESTClient` is **synchronous** (blocking HTTP under the hood). In an async +app it must be called from a worker thread (`asyncio.to_thread(...)`) so it does +not block the event loop. This is exactly how the project's `MassiveDataSource` +uses it. + +## Rate Limits + +| Tier | Limit | Recommended poll interval | +|------|-------|---------------------------| +| Free | 5 requests/minute | 15 seconds | +| Starter / paid | higher | 2-5 seconds | + +The key design consequence: **fetch all watched tickers in a single request** +(the full snapshot endpoint), not one request per ticker. Ten tickers polled +individually would blow the free-tier budget instantly; one snapshot call covers +all of them. + +## Endpoints Used By This Project + +### 1. Full Market Snapshot — primary real-time source + +Fetch the latest snapshot for many tickers in one call. This is what the live +poller uses. + +REST: +``` +GET /v2/snapshot/locale/us/markets/stocks/tickers?tickers=AAPL,GOOGL,MSFT +``` + +Python client: +```python +from massive.rest.models import SnapshotMarketType + +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) +``` + +Key snapshot object attributes: + +| Attribute | Meaning | +|-----------|---------| +| `snap.ticker` | Ticker symbol | +| `snap.last_trade.price` | Most recent trade price | +| `snap.last_trade.timestamp` | Unix **milliseconds** (divide by 1000 for seconds) | +| `snap.last_quote` | Latest bid/ask | +| `snap.day` | Today's aggregate (OHLCV) | +| `snap.prev_day` | Previous trading day's aggregate (OHLCV) — used for daily % change | +| `snap.todays_change` / `snap.todays_change_percent` | Convenience fields | + +Notes: +- Snapshot data is cleared daily ~3:30 AM EST and repopulates from ~4:00 AM EST. +- Outside market hours `last_trade` reflects the last trade of the prior + session; `prev_day` is the reference for daily change. + +### 2. Single Ticker Snapshot + +```python +snap = client.get_snapshot_ticker(market_type=SnapshotMarketType.STOCKS, ticker="AAPL") +price = snap.last_trade.price +``` + +Useful for validating a ticker exists when the user adds one to the watchlist. + +### 3. Previous Close — simple EOD price + +```python +agg = client.get_previous_close_agg(ticker="AAPL") +# agg[0].close, agg[0].open, agg[0].high, agg[0].low, agg[0].volume +``` + +REST: `GET /v2/aggs/ticker/{ticker}/prev` + +### 4. Daily Open/Close — EOD for a specific date + +```python +daily = client.get_daily_open_close_agg(ticker="AAPL", date="2026-06-12") +# daily.open, daily.close, daily.high, daily.low, daily.volume +``` + +REST: `GET /v1/open-close/{ticker}/{date}` + +### 5. Aggregate Bars — historical series (charts / sparklines backfill) + +```python +bars = [] +for bar in client.list_aggs( + ticker="AAPL", + multiplier=1, + timespan="day", # "minute" | "hour" | "day" ... + from_="2026-01-01", + to="2026-06-13", + limit=50000, +): + bars.append(bar) # bar.open, bar.high, bar.low, bar.close, bar.volume, bar.timestamp +``` + +REST: `GET /v2/aggs/ticker/{ticker}/range/{multiplier}/{timespan}/{from}/{to}` + +The client auto-paginates iterator endpoints (`list_aggs`, `list_trades`, +`list_quotes`); `limit` is page size, not a total cap. + +## Error Handling + +The poller must not crash the app on transient API failures. Common cases: + +| Failure | Cause | Handling | +|---------|-------|----------| +| 401 | bad/expired key | log, keep retrying (config issue, surfaced in logs) | +| 429 | rate limit exceeded | log, retry on next interval (increase interval) | +| Network / timeout | connectivity | log, retry on next interval | +| Missing fields on a snapshot | thin/halted ticker | skip that ticker, process the rest | + +Strategy: wrap each poll in a broad `try/except`, log the error, and let the +loop retry on the next tick. Per-snapshot parsing is wrapped separately so one +malformed entry does not discard the whole batch. + +## Mapping To The Project's Cache + +Each successful snapshot becomes a `cache.update(ticker, price, timestamp)` call. +The `PriceCache` computes `previous_price`, `change`, and `direction`. Timestamps +are converted from Massive's milliseconds to seconds: + +```python +self._cache.update( + ticker=snap.ticker, + price=snap.last_trade.price, + timestamp=snap.last_trade.timestamp / 1000.0, +) +``` + +## Sources + +- [Massive Python client (GitHub)](https://github.com/massive-com/client-python) +- [Massive Stocks REST API overview](https://massive.com/docs/rest/stocks/overview) +- [Massive + Python guide](https://massive.com/blog/polygon-io-with-python-for-stock-market-data) +- [Massive home / rebrand](https://massive.com/) From 6120423b5c96b8ca0799cd4e4a74619d52bd8c58 Mon Sep 17 00:00:00 2001 From: Rahul Sahu Date: Sat, 13 Jun 2026 17:51:28 +0400 Subject: [PATCH 2/2] Address market data review findings - stream.py: create router inside factory (no module-global double-registration) - massive_client.py: pass ticker-list copy into worker thread (thread safety) - massive_client.py: exponential backoff on consecutive poll failures (429-safe) - massive_client.py: document add_ticker poll latency - tests: add SSE endpoint tests (stream.py coverage 33% -> 97%) 77 tests pass, lint clean, 97% coverage. Co-Authored-By: Claude Opus 4.8 --- backend/app/market/massive_client.py | 35 +++-- backend/app/market/stream.py | 7 +- backend/tests/market/test_stream.py | 74 ++++++++++ planning/MARKET_DATA_REVIEW.md | 201 +++++++++++++++++++++++++++ 4 files changed, 305 insertions(+), 12 deletions(-) create mode 100644 backend/tests/market/test_stream.py 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 00bc7b2a..95f08499 100644 --- a/backend/app/market/massive_client.py +++ b/backend/app/market/massive_client.py @@ -37,6 +37,7 @@ def __init__( self._tickers: list[str] = [] self._task: asyncio.Task | None = None self._client: RESTClient | None = None + self._consecutive_failures: int = 0 async def start(self, tickers: list[str]) -> None: self._client = RESTClient(api_key=self._api_key) @@ -64,6 +65,9 @@ async def stop(self) -> None: logger.info("Massive poller stopped") async def add_ticker(self, ticker: str) -> None: + # The new ticker has no price until the next poll cycle (up to + # poll_interval seconds later). We intentionally do not fetch it + # immediately to conserve the API rate-limit budget. ticker = ticker.upper().strip() if ticker not in self._tickers: self._tickers.append(ticker) @@ -81,9 +85,15 @@ def get_tickers(self) -> list[str]: # --- Internal --- async def _poll_loop(self) -> None: - """Poll on interval. First poll already happened in start().""" + """Poll on interval. First poll already happened in start(). + + On consecutive failures, back off exponentially (capped) so a sustained + error such as a 429 rate-limit does not keep hammering the API at the + base interval. + """ while True: - await asyncio.sleep(self._interval) + delay = self._interval * (2 ** min(self._consecutive_failures, 4)) + await asyncio.sleep(delay) await self._poll_once() async def _poll_once(self) -> None: @@ -93,8 +103,10 @@ async def _poll_once(self) -> None: try: # The Massive RESTClient is synchronous — run in a thread to - # avoid blocking the event loop. - snapshots = await asyncio.to_thread(self._fetch_snapshots) + # avoid blocking the event loop. Pass a copy of the ticker list so + # concurrent add/remove cannot mutate it mid-request. + snapshots = await asyncio.to_thread(self._fetch_snapshots, list(self._tickers)) + self._consecutive_failures = 0 processed = 0 for snap in snapshots: try: @@ -116,13 +128,18 @@ async def _poll_once(self) -> None: 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 will retry on the next interval. + self._consecutive_failures += 1 + logger.error("Massive poll failed (%d in a row): %s", self._consecutive_failures, e) + # Don't re-raise — the loop will retry with backoff on the next cycle. # 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.""" + def _fetch_snapshots(self, tickers: list[str]) -> list: + """Synchronous call to the Massive REST API. Runs in a thread. + + Receives a snapshot copy of the ticker list so concurrent + add/remove on the event loop cannot mutate it mid-request. + """ return self._client.get_snapshot_all( market_type=SnapshotMarketType.STOCKS, - tickers=self._tickers, + tickers=tickers, ) diff --git a/backend/app/market/stream.py b/backend/app/market/stream.py index 7fd974b7..cee981b6 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. + The router is created fresh on each call so the factory is self-contained: + no shared module-global state, and calling it twice never double-registers + the route. """ + router = APIRouter(prefix="/api/stream", tags=["streaming"]) @router.get("/prices") async def stream_prices(request: Request) -> StreamingResponse: diff --git a/backend/tests/market/test_stream.py b/backend/tests/market/test_stream.py new file mode 100644 index 00000000..26eef4b4 --- /dev/null +++ b/backend/tests/market/test_stream.py @@ -0,0 +1,74 @@ +"""Tests for the SSE streaming endpoint.""" + +import json + +from app.market.cache import PriceCache +from app.market.stream import _generate_events, create_stream_router + + +class FakeRequest: + """Minimal stand-in for a Starlette Request. + + is_disconnected() returns False for `alive_iterations` calls, then True, + so the generator loop terminates deterministically. + """ + + def __init__(self, alive_iterations: int = 1) -> None: + self._alive = alive_iterations + self.client = None + + async def is_disconnected(self) -> bool: + if self._alive > 0: + self._alive -= 1 + return False + return True + + +async def test_create_stream_router_fresh_each_call(): + """Each call returns a self-contained router with exactly one /prices route.""" + cache = PriceCache() + r1 = create_stream_router(cache) + r2 = create_stream_router(cache) + + assert r1 is not r2 + paths = [route.path for route in r1.routes] + assert paths.count("/api/stream/prices") == 1 + + +async def test_generate_events_emits_retry_then_prices(): + cache = PriceCache() + cache.update("AAPL", 190.0) + + gen = _generate_events(cache, FakeRequest(alive_iterations=1), interval=0.0) + + first = await anext(gen) + assert first == "retry: 1000\n\n" + + payload = await anext(gen) + assert payload.startswith("data: ") + data = json.loads(payload[len("data: ") :].strip()) + assert data["AAPL"]["price"] == 190.0 + assert data["AAPL"]["direction"] == "flat" + + +async def test_generate_events_only_sends_on_version_change(): + cache = PriceCache() + cache.update("AAPL", 190.0) + + # Allow several loop iterations but never change the cache after the first send. + gen = _generate_events(cache, FakeRequest(alive_iterations=3), interval=0.0) + + events = [event async for event in gen] + + # retry directive + exactly one data event (no duplicate sends for an unchanged cache) + data_events = [e for e in events if e.startswith("data: ")] + assert len(data_events) == 1 + + +async def test_generate_events_stops_on_disconnect(): + cache = PriceCache() + # Disconnected immediately: only the retry directive is yielded. + gen = _generate_events(cache, FakeRequest(alive_iterations=0), interval=0.0) + + events = [event async for event in gen] + assert events == ["retry: 1000\n\n"] diff --git a/planning/MARKET_DATA_REVIEW.md b/planning/MARKET_DATA_REVIEW.md new file mode 100644 index 00000000..95464f4a --- /dev/null +++ b/planning/MARKET_DATA_REVIEW.md @@ -0,0 +1,201 @@ +# Market Data Backend — Code Review + +**Date:** 2026-06-13 +**Scope:** `backend/app/market/` (8 modules) and `backend/tests/market/` (6 test modules) +**Reviewer:** Claude (Opus 4.8) + +## Resolution Status (2026-06-13) + +All findings below have been addressed on branch `fix/market-data-review-items`: + +1. **SSE endpoint tests** — added `tests/market/test_stream.py` (4 tests). `stream.py` coverage 33% → 97%. +2. **Module-global router** — router is now created inside `create_stream_router`; no shared state, no double-registration. +3. **Massive add_ticker latency** — documented in code (intentional, to conserve rate-limit budget). +4. **Cross-thread `_tickers`** — `_poll_once` now passes a list copy into the worker thread. +5. **429 backoff** — `_poll_loop` backs off exponentially (capped at 16x) on consecutive failures; resets on success. + +Result: **77 tests pass, lint clean, 97% coverage.** The original review follows. + +--- + +## Summary + +The market data subsystem is in **good shape**: clean Strategy-pattern design, +full test pass, lint clean, and high coverage on the core logic. The architecture +(sources push into a versioned `PriceCache`; consumers read from it) is sound and +well decoupled. + +A few low-severity issues are worth addressing, the most actionable being **no +test coverage for the SSE endpoint** (`stream.py` at 33%) and a **module-global +router** in `stream.py` that double-registers routes if the factory is called +more than once. + +**Verdict:** Ship-ready. The findings below are improvements, not blockers. + +## Verification Results + +### Tests — PASS + +``` +73 passed in 3.64s +``` + +All tests pass. Async tests run under `asyncio_mode = "auto"`. + +### Lint (ruff) — PASS + +``` +All checks passed! +``` + +Rules `E, F, I, N, W` with `E501` ignored. No issues. + +### Coverage — 91% overall + +``` +Name Stmts Miss Cover Missing +------------------------------------------------------------ +app/market/__init__.py 6 0 100% +app/market/cache.py 39 0 100% +app/market/factory.py 15 0 100% +app/market/interface.py 13 0 100% +app/market/massive_client.py 67 4 94% 85-87, 125 +app/market/models.py 26 0 100% +app/market/seed_prices.py 8 0 100% +app/market/simulator.py 139 3 98% 149, 268-269 +app/market/stream.py 36 24 33% 26-48, 62-87 +------------------------------------------------------------ +TOTAL 349 31 91% +``` + +The 91% is healthy, but it is unevenly distributed: core logic is at 94-100% +while the **SSE endpoint is effectively untested**. + +## Findings + +Severity: **High** = fix before relying on it · **Medium** = should fix · +**Low** = nice to have. + +### 1. [Medium] SSE endpoint has no tests (`stream.py`, 33% coverage) + +`_generate_events` and `create_stream_router` are the live wire to the frontend +and the one piece with no automated coverage. The disconnect path, the +version-change gate, and the payload shape are all unverified. + +**Recommendation:** Add a unit test using FastAPI's `TestClient` (SSE responses +can be read as a stream) or test `_generate_events` directly by driving the async +generator with a fake `Request` whose `is_disconnected()` flips to `True` after +one or two iterations, asserting the emitted `data:` payload matches the cache. + +```python +async def test_generate_events_emits_changed_prices(): + cache = PriceCache() + cache.update("AAPL", 190.0) + req = FakeRequest(disconnect_after=1) # is_disconnected() -> True on 2nd call + gen = _generate_events(cache, req, interval=0.0) + assert await anext(gen) == "retry: 1000\n\n" + payload = await anext(gen) + assert '"AAPL"' in payload and '"direction"' in payload +``` + +### 2. [Medium] Module-global router double-registers routes + +In `stream.py` the router is created at module import: + +```python +router = APIRouter(prefix="/api/stream", tags=["streaming"]) + +def create_stream_router(price_cache: PriceCache) -> APIRouter: + @router.get("/prices") # registers on the SHARED global router + async def stream_prices(...): ... + return router +``` + +Each call to `create_stream_router` adds another `/prices` route to the *same* +global router object. Calling the factory twice (e.g., in tests, or if app setup +is ever re-run) registers a duplicate route and leaks the first cache reference. + +**Recommendation:** Create the router inside the factory so each call is +self-contained: + +```python +def create_stream_router(price_cache: PriceCache) -> APIRouter: + router = APIRouter(prefix="/api/stream", tags=["streaming"]) + + @router.get("/prices") + async def stream_prices(request: Request) -> StreamingResponse: + ... + return router +``` + +### 3. [Low] Massive `add_ticker` doesn't seed the cache immediately + +`SimulatorDataSource.add_ticker` seeds the cache right away so a newly added +ticker has a price instantly. `MassiveDataSource.add_ticker` does not — the +ticker has no price until the next poll, up to `poll_interval` (15s default) +later. The frontend will show a blank cell in the meantime. + +**Recommendation:** Either document this as expected (the code comment already +says "will appear on next poll"), or trigger a one-off snapshot fetch for the new +ticker on add. Documenting is sufficient given the free-tier rate-limit +constraint; an immediate single-ticker fetch costs a request from the budget. + +### 4. [Low] `_tickers` mutated across threads in MassiveDataSource + +`_fetch_snapshots` runs in a worker thread (`asyncio.to_thread`) and reads +`self._tickers`, while `add_ticker`/`remove_ticker` mutate it from the event +loop. Individual list operations are GIL-atomic so this won't corrupt memory, but +a poll could capture a partially-updated set. The window is tiny and the next +poll self-heals, so impact is negligible. + +**Recommendation:** If hardening is wanted, snapshot the list under a lock (or +pass a copy into the thread). Low priority. + +### 5. [Low] No backoff on rate-limit (429) responses + +On any poll failure — including 429 — the loop logs and retries at the same +`poll_interval`. If the interval is set too aggressively for the tier, it will +keep hitting the limit at a steady rate rather than backing off. + +**Recommendation:** Optional. A simple exponential backoff on consecutive +failures would be more polite to the API, but the fixed-interval default (15s) +is already free-tier safe. + +## Things Verified as NOT Problems + +- **Cholesky positive-definiteness.** Concern: arbitrary user-added tickers all + get 0.3 cross-correlation and the block structure (tech 0.6, finance 0.5) could + produce a non-positive-definite matrix, making `np.linalg.cholesky` raise. I + tested the worst case (40 tech + 5 finance, 45 tickers): matrix stays PD with + min eigenvalue ~0.40. Since cross-correlation (0.3) is below intra-group + values, the nested-block matrix remains PD. **Not a risk** at any realistic + watchlist size. +- **Thread safety of `PriceCache`.** All mutating and reading methods hold the + lock; `get_all` returns a shallow copy. Correct for the one-writer / + many-reader model. +- **First-update semantics.** `cache.update` sets `previous_price == price` on + the first update so `direction` is `"flat"` and `change` is 0 — no spurious + flash on initial load. Correct. +- **Task lifecycle.** Both sources cancel the background task in `stop()` and + await the `CancelledError`; `stop()` is idempotent. Tested. +- **Timestamp conversion.** Massive ms → s conversion is correct and tested. + +## Coverage Gap Detail + +The uncovered lines are acceptable except for `stream.py`: + +| Module | Missing | Assessment | +|--------|---------|------------| +| `massive_client.py` | 85-87, 125 | `_poll_loop` sleep branch + `_fetch_snapshots` real call — only reachable against the live API; fine to leave (tested via mocked `_poll_once`) | +| `simulator.py` | 149, 268-269 | early-return guard + loop exception branch — minor | +| `stream.py` | 26-48, 62-87 | **the whole endpoint** — see Finding #1 | + +## Recommended Actions (priority order) + +1. Add SSE endpoint tests (Finding #1). +2. Move router creation inside the factory (Finding #2). +3. Document Massive `add_ticker` latency, or add an opt-in immediate fetch (#3). +4. (Optional) thread-snapshot `_tickers`; add 429 backoff (#4, #5). + +None of these block use of the subsystem. The design is clean, the core logic is +fully tested, and behavior is correct.