From 772545277aaa485f3897563de132048215548a3f Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 23:59:32 +0000 Subject: [PATCH] Add detailed market data backend design doc Documents the unified MarketDataSource interface, GBM simulator, Massive API client, price cache, and SSE streaming endpoint with real code from the existing implementation, plus forward-looking FastAPI lifecycle integration guidance for the not-yet-built app layer. --- planning/MARKET_DATA_DESIGN.md | 1252 ++++++++++++++++++++++++++++++++ 1 file changed, 1252 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 00000000..4811cfff --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,1252 @@ +# Market Data Backend — Detailed Design + +Implementation-ready design for the FinAlly market data subsystem: the unified +`MarketDataSource` interface, the in-memory `PriceCache`, the GBM simulator, the +Massive (Polygon.io) API client, the SSE streaming endpoint, and how the FastAPI +app wires them together. + +**Status:** The subsystem described here is fully implemented and tested at +`backend/app/market/` (see [§12](#12-testing-strategy) — 73 tests passing). The +FastAPI application shell (`app/main.py`), database layer, and REST routes that +consume this subsystem have **not** been built yet — [§10](#10-fastapi-lifecycle-integration) +and [§11](#11-watchlist-coordination) are forward-looking guidance for whoever +builds that layer next, written to match the interface that already exists. + +All code below is copied verbatim from the real source files (not aspirational +pseudocode), so it can be cross-referenced directly against the repo. + +--- + +## 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 & Correlation — `seed_prices.py`](#6-seed-prices--correlation) +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 Lifecycle Integration](#10-fastapi-lifecycle-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 │ (abstract interface) + │ start / stop / │ + │ add_ticker / │ + │ remove_ticker / │ + │ get_tickers │ + └───────────┬────────────┘ + │ implements + ┌────────────────┴─────────────────┐ + │ │ + ┌─────────▼──────────┐ ┌───────────▼───────────┐ + │ SimulatorDataSource │ │ MassiveDataSource │ + │ (GBM, ~500ms tick) │ │ (Polygon.io REST poll) │ + │ default, no API key │ │ used when │ + │ needed │ │ MASSIVE_API_KEY is set │ + └─────────┬──────────┘ └───────────┬───────────┘ + │ │ + └───────────────┬───────────────────┘ + │ writes + ┌────────▼─────────┐ + │ PriceCache │ thread-safe, in-memory + │ {ticker: latest} │ version counter for + └────────┬──────────┘ change detection + │ reads + ┌────────────────────┼────────────────────┐ + │ │ │ +┌───────▼────────┐ ┌────────▼─────────┐ ┌────────▼─────────┐ +│ SSE stream │ │ Portfolio │ │ Trade execution │ +│ /api/stream/ │ │ valuation │ │ (fills at cache │ +│ prices │ │ │ │ price) │ +└─────────────────┘ └──────────────────┘ └────────────────────┘ +``` + +`create_market_data_source()` selects between the two implementations at +startup based on the `MASSIVE_API_KEY` environment variable. Everything +downstream — SSE streaming, portfolio math, trade fills — talks only to +`PriceCache` and never needs to know which source is active. + +--- + +## 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 +``` + +Downstream code imports only from the package root: + +```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 — +SSE streaming, portfolio valuation, and trade execution all consume this type +exclusively. + +```python +"""Data models for market data.""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass(frozen=True, slots=True) +class PriceUpdate: + """Immutable snapshot of a single ticker's price at a point in time.""" + + ticker: str + price: float + previous_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: + """Absolute price change from previous update.""" + return round(self.price - self.previous_price, 4) + + @property + def change_percent(self) -> float: + """Percentage change from previous update.""" + if self.previous_price == 0: + return 0.0 + return round((self.price - self.previous_price) / self.previous_price * 100, 4) + + @property + def direction(self) -> str: + """'up', 'down', or 'flat'.""" + if self.price > self.previous_price: + return "up" + elif self.price < self.previous_price: + return "down" + return "flat" + + def to_dict(self) -> dict: + """Serialize for JSON / SSE transmission.""" + return { + "ticker": self.ticker, + "price": self.price, + "previous_price": self.previous_price, + "timestamp": self.timestamp, + "change": self.change, + "change_percent": self.change_percent, + "direction": self.direction, + } +``` + +### Design decisions + +- **`frozen=True`**: price updates are immutable value objects, safe to share across async tasks without copying. +- **`slots=True`**: memory optimization — many of these are created per second. +- **Computed properties** (`change`, `direction`, `change_percent`): derived from `price`/`previous_price` so they can never drift out of sync with a stale stored field. +- **`to_dict()`**: single serialization point shared by the SSE endpoint and any future REST responses. + +--- + +## 4. Price Cache + +**File: `backend/app/market/cache.py`** + +The central data hub. Data sources write to it; SSE streaming, portfolio +valuation, and trade execution read from it. + +```python +"""Thread-safe in-memory price cache.""" + +from __future__ import annotations + +import time +from threading import Lock + +from .models import PriceUpdate + + +class PriceCache: + """Thread-safe in-memory cache of the latest price for each ticker. + + Writers: SimulatorDataSource or MassiveDataSource (one at a time). + Readers: SSE streaming endpoint, portfolio valuation, trade execution. + """ + + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._lock = Lock() + self._version: int = 0 # Monotonically increasing; bumped on every update + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + """Record a new price for a ticker. Returns the created PriceUpdate. + + Automatically computes direction and change from the previous price. + If this is the first update for the ticker, previous_price == price (direction='flat'). + """ + with self._lock: + ts = timestamp or time.time() + prev = self._prices.get(ticker) + previous_price = prev.price if prev else price + + update = PriceUpdate( + ticker=ticker, + price=round(price, 2), + previous_price=round(previous_price, 2), + timestamp=ts, + ) + self._prices[ticker] = update + self._version += 1 + return update + + def get(self, ticker: str) -> PriceUpdate | None: + """Get the latest price for a single ticker, or None if unknown.""" + with self._lock: + return self._prices.get(ticker) + + def get_all(self) -> dict[str, PriceUpdate]: + """Snapshot of all current prices. Returns a shallow copy.""" + with self._lock: + return dict(self._prices) + + def get_price(self, ticker: str) -> float | None: + """Convenience: get just the price float, or None.""" + update = self.get(ticker) + return update.price if update else None + + def remove(self, ticker: str) -> None: + """Remove a ticker from the cache (e.g., when removed from watchlist).""" + with self._lock: + self._prices.pop(ticker, None) + + @property + def version(self) -> int: + """Current version counter. Useful for SSE change detection.""" + return self._version + + def __len__(self) -> int: + with self._lock: + return len(self._prices) + + def __contains__(self, ticker: str) -> bool: + with self._lock: + return ticker in self._prices +``` + +### Why a version counter? + +The SSE loop polls the cache every ~500ms. Without a version counter it would +re-serialize and resend every price on every tick even when nothing changed +(e.g., the Massive source only updates every 15s). The counter lets the SSE +loop skip a send when nothing is new: + +```python +last_version = -1 +while True: + if price_cache.version != last_version: + last_version = price_cache.version + yield format_sse(price_cache.get_all()) + await asyncio.sleep(0.5) +``` + +### Thread safety rationale + +`threading.Lock` is used instead of `asyncio.Lock` because the Massive +client's synchronous call runs via `asyncio.to_thread()` — a real OS thread +that `asyncio.Lock` would not protect against. `threading.Lock` works +correctly from both a sync thread and the async event loop, which is the +combination this cache actually has to survive. + +--- + +## 5. Abstract Interface + +**File: `backend/app/market/interface.py`** + +```python +"""Abstract interface for market data sources.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class MarketDataSource(ABC): + """Contract for market data providers. + + Implementations push price updates into a shared PriceCache on their own + schedule. Downstream code never calls the data source directly for prices — + it reads from the cache. + + Lifecycle: + source = create_market_data_source(cache) + await source.start(["AAPL", "GOOGL", ...]) + # ... app runs ... + await source.add_ticker("TSLA") + await source.remove_ticker("GOOGL") + # ... app shutting down ... + await source.stop() + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing price updates for the given tickers. + + Starts a background task that periodically writes to the PriceCache. + Must be called exactly once. Calling start() twice is undefined behavior. + """ + + @abstractmethod + async def stop(self) -> None: + """Stop the background task and release resources. + + Safe to call multiple times. After stop(), the source will not write + to the cache again. + """ + + @abstractmethod + async def add_ticker(self, ticker: str) -> None: + """Add a ticker to the active set. No-op if already present. + + The next update cycle will include this ticker. + """ + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Remove a ticker from the active set. No-op if not present. + + Also removes the ticker from the PriceCache. + """ + + @abstractmethod + def get_tickers(self) -> list[str]: + """Return the current list of actively tracked tickers.""" +``` + +### Why the source writes to the cache instead of returning prices + +This push model decouples timing. The simulator ticks every 500ms, Massive +polls every 15s, but SSE always reads from the cache at its own 500ms +cadence — it never needs to know which data source is active or how often +it updates. + +--- + +## 6. Seed Prices & Correlation + +**File: `backend/app/market/seed_prices.py`** + +Constants only — no logic, no imports beyond stdlib. Shared by the simulator +(initial prices, GBM parameters, correlation structure) and reusable as +fallback prices for a newly-added ticker before any real data has arrived. + +```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 +``` + +A dynamically-added ticker not in `SEED_PRICES`/`TICKER_PARAMS` falls back to +a random seed price in `$50–$300` and `DEFAULT_PARAMS` (see +[§7.4](#74-dynamically-added-tickers)). + +--- + +## 7. GBM Simulator + +**File: `backend/app/market/simulator.py`** + +Two classes live here: +- `GBMSimulator` — pure math engine, stateful, holds current prices and steps them forward. +- `SimulatorDataSource` — the `MarketDataSource` implementation that wraps it in an async loop and writes to `PriceCache`. + +### 7.1 The Math + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +Where `S(t)` is the current price, `mu` is annualized drift, `sigma` is +annualized volatility, `dt` is the time step as a fraction of a trading year, +and `Z` is a correlated standard-normal draw. `dt` is tiny (~8.48e-8 for a +500ms tick over a 252-day, 6.5-hour trading year), so each tick produces a +sub-cent move that accumulates naturally over many ticks — this is what +makes the price series look organic rather than jumpy. + +### 7.2 `GBMSimulator` + +```python +class GBMSimulator: + """Geometric Brownian Motion simulator for correlated stock prices.""" + + # 500ms expressed as a fraction of a trading year + # 252 trading days * 6.5 hours/day * 3600 seconds/hour = 5,896,800 seconds + TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 + 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() + + def step(self) -> dict[str, float]: + """Advance all tickers by one time step. Returns {ticker: new_price}. + + This is the hot path — called every 500ms. Keep it fast. + """ + n = len(self._tickers) + if n == 0: + return {} + + z_independent = np.random.standard_normal(n) + z_correlated = self._cholesky @ z_independent if self._cholesky is not None else z_independent + + result: dict[str, float] = {} + for i, ticker in enumerate(self._tickers): + params = self._params[ticker] + mu, sigma = params["mu"], params["sigma"] + + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z_correlated[i] + self._prices[ticker] *= math.exp(drift + diffusion) + + # Random event: ~0.1% chance per tick per ticker (~every 50s + # across 10 tickers at 2 ticks/sec) — a 2-5% shock for visual drama + 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 + + 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) + + 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: + """O(n^2), called on add/remove. Fine for n < 50 tickers.""" + 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: + """Same tech sector: 0.6. Same finance sector: 0.5. TSLA or + cross-sector or unknown: 0.3.""" + 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 +``` + +Correlated draws are produced by multiplying a vector of independent +standard-normal samples by the Cholesky factor of the correlation matrix +(`L @ z` where `corr = L @ L.T`) — the standard technique for generating +jointly-correlated normals from independent ones. + +### 7.3 `SimulatorDataSource` — Async Wrapper + +```python +class SimulatorDataSource(MarketDataSource): + """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 immediately so SSE has data before the first tick + 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") + + 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 + + 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) +``` + +Key behaviors: +- **Immediate seeding** — `start()` and `add_ticker()` both write a price to the cache synchronously, before the loop's first tick, so there's never a blank-screen gap. +- **Graceful cancellation** — `stop()` cancels the task and awaits it, swallowing `CancelledError`, so FastAPI lifespan teardown is clean. +- **Exception resilience** — the loop catches exceptions per-step so one bad tick can't kill the whole feed. + +### 7.4 Dynamically Added Tickers + +A ticker not present in `SEED_PRICES`/`TICKER_PARAMS` (e.g. the user or the +AI assistant adds `"PYPL"` to the watchlist) gets a random seed price in +`$50–$300` and `DEFAULT_PARAMS` (`sigma=0.25, mu=0.05`), and is treated as +uncorrelated with everything else (`CROSS_GROUP_CORR = 0.3`, the same +fallback used for cross-sector pairs). + +--- + +## 8. Massive API Client + +**File: `backend/app/market/massive_client.py`** + +Polls the Massive (Polygon.io) REST snapshot endpoint on a configurable +interval. The underlying `massive` client is synchronous, so calls run via +`asyncio.to_thread()` to avoid blocking the event loop. + +```python +"""Massive (Polygon.io) API client for real market data.""" + +from __future__ import annotations + +import asyncio +import logging + +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +from .cache import PriceCache +from .interface import MarketDataSource + +logger = logging.getLogger(__name__) + + +class MassiveDataSource(MarketDataSource): + """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) + + # 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") + + 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 + + async def add_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + if ticker not in self._tickers: + self._tickers.append(ticker) # appears on the next poll + + 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) + + 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: + 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 — retried 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, + ) +``` + +### Error handling philosophy + +| 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 the next cycle. | +| **Malformed snapshot** | That ticker is skipped with a warning; other tickers in the same response still get processed. | +| **All tickers fail** | Cache retains last-known prices; SSE keeps streaming stale-but-present data — better than nothing. | + +`massive` is a declared core dependency of the backend (`pyproject.toml`), so +the import at the top of the module is unconditional — there is no lazy +import to avoid a hard dependency, since every install already includes it. + +--- + +## 9. Factory + +**File: `backend/app/market/factory.py`** + +```python +"""Factory for creating market data sources.""" + +from __future__ import annotations + +import logging +import os + +from .cache import PriceCache +from .interface import MarketDataSource +from .massive_client import MassiveDataSource +from .simulator import SimulatorDataSource + +logger = logging.getLogger(__name__) + + +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """Create the appropriate market data source based on environment variables. + + - MASSIVE_API_KEY set and non-empty → MassiveDataSource (real market data) + - Otherwise → SimulatorDataSource (GBM simulation) + + Returns an unstarted source. Caller must await source.start(tickers). + """ + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + + if api_key: + logger.info("Market data source: Massive API (real data)") + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + else: + logger.info("Market data source: GBM Simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +Usage at app startup: + +```python +price_cache = PriceCache() +source = create_market_data_source(price_cache) +await source.start(initial_tickers) # e.g., ["AAPL", "GOOGL", ...] +``` + +--- + +## 9.5 SSE Streaming Endpoint + +**File: `backend/app/market/stream.py`** + +A FastAPI route that holds open a long-lived `text/event-stream` connection +and pushes price updates to the client. + +```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 + +router = APIRouter(prefix="/api/stream", tags=["streaming"]) + + +def create_stream_router(price_cache: PriceCache) -> APIRouter: + """Factory pattern injects the PriceCache without module-level 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]: + # Tell the browser to retry after 1 second if the connection drops + yield "retry: 1000\n\n" + + last_version = -1 + try: + while True: + if await request.is_disconnected(): + 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: + pass +``` + +### Wire format + +``` +data: {"AAPL":{"ticker":"AAPL","price":190.50,"previous_price":190.42,"timestamp":1707580800.5,"change":0.08,"change_percent":0.042,"direction":"up"},"GOOGL":{...}} +``` + +Frontend consumption: + +```javascript +const eventSource = new EventSource('/api/stream/prices'); +eventSource.onmessage = (event) => { + const prices = JSON.parse(event.data); + // prices is { "AAPL": { ticker, price, previous_price, ... }, ... } +}; +``` + +### Why poll-and-push instead of event-driven? + +The endpoint polls the cache on a fixed interval rather than being notified +by the data source. This produces predictable, evenly-spaced updates, which +matters because the frontend accumulates these events into sparkline charts +— even spacing keeps that visualization clean regardless of which data +source (500ms simulator vs. 15s Massive poll) is active underneath. + +--- + +## 10. FastAPI Lifecycle Integration + +> **Not yet built.** `app/main.py` does not exist in the repo yet — the +> FastAPI app shell, database, and REST routes are the next phase of work. +> This section shows how that layer should wire up the market data +> subsystem, matching the public interface that already exists and is +> already tested. + +The market data system should start and stop with the FastAPI app via the +`lifespan` context manager: + +```python +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from app.market import PriceCache, create_market_data_source, create_stream_router + + +@asynccontextmanager +async def lifespan(app: FastAPI): + # --- STARTUP --- + price_cache = PriceCache() + app.state.price_cache = price_cache + + source = create_market_data_source(price_cache) + app.state.market_source = source + + initial_tickers = await load_watchlist_tickers() # reads from SQLite + await source.start(initial_tickers) + + app.include_router(create_stream_router(price_cache)) + + 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 (trade execution, watchlist management) should access the cache +and 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 + +> Also forward-looking — describes how the not-yet-built watchlist routes +> should interact with the market data source. + +### Adding a ticker + +``` +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 + +``` +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 has an open position + +If the user removes a ticker from the watchlist while still holding shares, +the data source must keep tracking it so portfolio valuation stays accurate. +The watchlist route should check for an open position before calling +`remove_ticker()`: + +```python +@router.delete("/watchlist/{ticker}") +async def remove_from_watchlist( + ticker: str, + source: MarketDataSource = Depends(get_market_source), +): + await db.delete_watchlist_entry(ticker) + + position = await db.get_position(ticker) + if position is None or position.quantity == 0: + await source.remove_ticker(ticker) + + return {"status": "ok"} +``` + +--- + +## 12. Testing Strategy + +**73 tests, all passing** (`uv run --extra dev pytest -v` from `backend/`), +across 6 modules in `backend/tests/market/`: + +| Module | Focus | +|--------|-------| +| `test_models.py` | `PriceUpdate` computed properties (`change`, `change_percent`, `direction`), `to_dict()` serialization | +| `test_cache.py` | update/get/get_all/remove, version counter increments, first-update-is-flat behavior | +| `test_simulator.py` | GBM math: prices stay positive, drift over many steps, add/remove ticker rebuilds Cholesky, unknown tickers get random seed prices, empty simulator returns `{}` | +| `test_simulator_source.py` | `SimulatorDataSource` integration: cache seeded before first tick, prices evolve over time, clean double-`stop()`, add/remove ticker propagates to cache | +| `test_factory.py` | Env-var driven selection between `MassiveDataSource` and `SimulatorDataSource` | +| `test_massive.py` | Mocked `RESTClient` — successful poll updates cache, malformed snapshot skipped without killing the batch, API exception doesn't crash the poll loop | + +### Representative patterns + +**Simulator math invariants:** + +```python +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 # 1 ticker, no correlation matrix needed + sim.add_ticker("GOOGL") + assert sim._cholesky is not None +``` + +**Cache change detection:** + +```python +def test_version_increments(self): + cache = PriceCache() + v0 = cache.version + cache.update("AAPL", 190.00) + assert cache.version == v0 + 1 +``` + +**Massive client resilience (mocked, no network calls):** + +```python +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 # good ticker still processed + assert cache.get_price("BAD") is None # bad one skipped, no crash + + +async def test_api_error_does_not_crash(self): + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + + with patch.object(source, "_fetch_snapshots", side_effect=Exception("network error")): + await source._poll_once() # must not raise + + assert cache.get_price("AAPL") is None +``` + +### Coverage + +Overall: 84%. `models.py`, `cache.py`, and `factory.py` at 100%; `simulator.py` +at 98%; `massive_client.py` at 56% (expected — the actual HTTP call is mocked +out, so only the request-construction and response-parsing paths are +exercised, not the `massive` package's own internals). + +### Manual verification: terminal demo + +```bash +cd backend +uv run market_data_demo.py +``` + +Runs the simulator standalone for 60 seconds (or until Ctrl+C), rendering a +live Rich dashboard with all 10 default tickers, per-ticker sparklines, +color-coded up/down arrows, and an event log for random shock events — a +fast way to eyeball that the GBM math and correlation structure look right +without booting the full FastAPI app. + +--- + +## 13. Error Handling & Edge Cases + +### 13.1 Startup with an empty watchlist + +If the database has no watchlist rows (e.g. user deleted everything), +`start([])` is handled gracefully by both sources — 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), the trade route should fail fast with a clear error: + +```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 case entirely by seeding the cache synchronously +in both `start()` and `add_ticker()`. The Massive poller has an unavoidable +gap between "ticker added" and "first successful poll" — the 400 with a +clear message is the correct behavior for that window. + +### 13.3 Invalid Massive API key + +If `MASSIVE_API_KEY` is set but wrong, the first poll fails with 401. The +poller logs the error and keeps retrying every `poll_interval` — it does not +crash or fall back to the simulator. The SSE endpoint keeps streaming empty +data; the user sees a "connected" status (SSE itself works fine) but no +prices. Fix: correct the key and restart the container. + +### 13.4 Thread safety under load + +`PriceCache`'s `threading.Lock` protects a tiny critical section (dict +lookup + assignment). At the project's actual scale — 10 tickers, ~2 +updates/sec, one SSE reader per browser tab — contention is negligible. If +this ever became a bottleneck (hundreds of tickers, many concurrent +readers), a read-write lock would be the fix, but that's not warranted here. + +### 13.5 Numerical stability of the simulator + +GBM with a tiny `dt` produces very small per-tick moves, but this isn't a +precision concern: prices are rounded to 2 decimals in `GBMSimulator.step()`, +the `exp()` formulation is numerically stable, and prices are structurally +guaranteed positive (an exponential is never negative or zero). + +--- + +## 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 (5 req/min free tier) | +| `event_probability` | `GBMSimulator.__init__` | `0.001` | Chance of a random 2–5% shock per ticker per tick | +| `dt` | `GBMSimulator.__init__` | `~8.5e-8` | GBM time step, fraction of a trading year | +| SSE push interval | `stream._generate_events()` | `0.5` s | Time between cache checks / pushes to the client | +| SSE retry directive | `stream._generate_events()` | `1000` ms | Browser `EventSource` reconnection delay | + +### Public package API + +**File: `backend/app/market/__init__.py`** + +```python +from .cache import PriceCache +from .factory import create_market_data_source +from .interface import MarketDataSource +from .models import PriceUpdate +from .stream import create_stream_router + +__all__ = [ + "PriceUpdate", + "PriceCache", + "MarketDataSource", + "create_market_data_source", + "create_stream_router", +] +```