diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json new file mode 100644 index 00000000..9f3c1b8a --- /dev/null +++ b/.claude-plugin/marketplace.json @@ -0,0 +1,18 @@ +{ + "name": "rick-tools", + "owner": { + "name": "rickhill65", + "email": "rick@eastmidlandsconsulting.com" + }, + "plugins": [ + { + "name": "independent-reviewer", + "source": "./independent-reviewer", + "description": "carry out an independent review of all changes since last commit", + "version": "1.0.0", + "author": { + "name": "rickhill65" + } + } + ] +} \ No newline at end of file diff --git a/.claude/agents/reviewer.md b/.claude/agents/reviewer.md new file mode 100644 index 00000000..3e5feed4 --- /dev/null +++ b/.claude/agents/reviewer.md @@ -0,0 +1,6 @@ +--- +name: reviewer +description: carry out a comprehensive review when requested +--- + +You review the file planning/PLAN.md and write your feedback to planning/REVIEWER.md \ No newline at end of file diff --git a/.claude/commands/doc-review.md b/.claude/commands/doc-review.md new file mode 100644 index 00000000..fefb880f --- /dev/null +++ b/.claude/commands/doc-review.md @@ -0,0 +1 @@ +Review the documentation file in the planning folder called $ARGUMENTS and add questions, clarifications or feedback to a new section at the end, along with any opportunities to simplify diff --git a/.claude/settings.json b/.claude/settings.json index aa06f43d..6aee70d3 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -1,7 +1,6 @@ { "enabledPlugins": { - "frontend-design@claude-plugins-official": true, - "context7@claude-plugins-official": true, + "independent-reviewer@rick-tools": true, "playwright@claude-plugins-official": true } } diff --git a/.gitignore b/.gitignore index b7faf403..45a94f9e 100644 --- a/.gitignore +++ b/.gitignore @@ -135,7 +135,8 @@ celerybeat.pid *.sage.py # Environments -.env +.env* +!.env.example .envrc .venv env/ diff --git a/README.md b/README.md index 3f2582ae..1a84a757 100644 --- a/README.md +++ b/README.md @@ -2,18 +2,20 @@ A visually stunning AI-powered trading workstation that streams live market data, simulates portfolio trading, and integrates an LLM chat assistant that can analyze positions and execute trades via natural language. -Built entirely by coding agents as a capstone project for an agentic AI coding course. +Built entirely by coding agents as a capstone project for an agentic AI coding course. See [`planning/PLAN.md`](planning/PLAN.md) for the full specification. -## Features +## Status -- **Live price streaming** via SSE with green/red flash animations -- **Simulated portfolio** — $10k virtual cash, market orders, instant fills -- **Portfolio visualizations** — heatmap (treemap), P&L chart, positions table -- **AI chat assistant** — analyzes holdings, suggests and auto-executes trades -- **Watchlist management** — track tickers manually or via AI -- **Dark terminal aesthetic** — Bloomberg-inspired, data-dense layout +**In progress.** The market data subsystem is complete; the rest of the platform (portfolio, LLM chat, frontend, Docker packaging) is still to be built. -## Architecture +- ✅ Market data backend — GBM simulator + Massive (Polygon.io) client behind a shared interface, SSE price streaming, 73 tests passing. See [`planning/MARKET_DATA_SUMMARY.md`](planning/MARKET_DATA_SUMMARY.md). +- ⬜ Portfolio & trading (positions, trades, P&L) +- ⬜ AI chat assistant (LiteLLM → OpenRouter, Cerebras inference) +- ⬜ Frontend (Next.js trading terminal UI) +- ⬜ Docker packaging & start/stop scripts +- ⬜ E2E tests + +## Architecture (target) Single Docker container serving everything on port 8000: @@ -23,40 +25,36 @@ Single Docker container serving everything on port 8000: - **AI**: LiteLLM → OpenRouter (Cerebras inference) with structured outputs - **Market data**: Built-in GBM simulator (default) or Massive API (optional) -## Quick Start +## Backend (available now) ```bash -# Clone and configure -cp .env.example .env -# Add your OPENROUTER_API_KEY to .env - -# Run with Docker -docker build -t finally . -docker run -v finally-data:/app/db -p 8000:8000 --env-file .env finally - -# Open http://localhost:8000 +cd backend +uv sync --extra dev +uv run --extra dev pytest -v # run tests +uv run market_data_demo.py # live terminal dashboard of simulated prices ``` +See [`backend/README.md`](backend/README.md) and [`backend/CLAUDE.md`](backend/CLAUDE.md) for details on the market data API. + ## Environment Variables | Variable | Required | Description | |---|---|---| -| `OPENROUTER_API_KEY` | Yes | OpenRouter API key for AI chat | -| `MASSIVE_API_KEY` | No | Massive (Polygon.io) key for real market data; omit to use simulator | +| `OPENROUTER_API_KEY` | For AI chat | OpenRouter API key | +| `MASSIVE_API_KEY` | No | Massive (Polygon.io) key for real market data; omit to use the simulator | | `LLM_MOCK` | No | Set `true` for deterministic mock LLM responses (testing) | ## Project Structure ``` finally/ -├── frontend/ # Next.js static export -├── backend/ # FastAPI uv project +├── backend/ # FastAPI uv project (market data complete; portfolio/AI/API pending) ├── planning/ # Project documentation and agent contracts -├── test/ # Playwright E2E tests -├── db/ # SQLite volume mount (runtime) -└── scripts/ # Start/stop helpers +└── db/ # SQLite volume mount (runtime) ``` +`frontend/`, `scripts/`, and `test/` are not yet created — see `planning/PLAN.md` for their intended layout. + ## License See [LICENSE](LICENSE). diff --git a/backend/app/market/massive_client.py b/backend/app/market/massive_client.py index 00bc7b2a..82abd04a 100644 --- a/backend/app/market/massive_client.py +++ b/backend/app/market/massive_client.py @@ -99,8 +99,8 @@ async def _poll_once(self) -> None: for snap in snapshots: try: price = snap.last_trade.price - # Massive timestamps are Unix milliseconds → convert to seconds - timestamp = snap.last_trade.timestamp / 1000.0 + # Massive trade timestamps (sip_timestamp) are Unix nanoseconds → convert to seconds + timestamp = snap.last_trade.sip_timestamp / 1_000_000_000 self._cache.update( ticker=snap.ticker, price=price, diff --git a/backend/tests/market/test_massive.py b/backend/tests/market/test_massive.py index cdd7dbd2..bb595ad2 100644 --- a/backend/tests/market/test_massive.py +++ b/backend/tests/market/test_massive.py @@ -3,18 +3,25 @@ from unittest.mock import MagicMock, patch import pytest +from massive.rest.models.trades import LastTrade from app.market.cache import PriceCache from app.market.massive_client import MassiveDataSource -def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock: - """Create a mock Massive snapshot object.""" +def _make_snapshot(ticker: str, price: float, sip_timestamp_ns: int) -> MagicMock: + """Create a mock Massive snapshot object. + + last_trade is spec'd against the real LastTrade model so that setting or + reading an attribute that doesn't actually exist on the API response + (e.g. `.timestamp`, which was mistakenly assumed in earlier code) raises + AttributeError here too, instead of silently mocking a fake field. + """ snap = MagicMock() snap.ticker = ticker - snap.last_trade = MagicMock() + snap.last_trade = MagicMock(spec_set=LastTrade) snap.last_trade.price = price - snap.last_trade.timestamp = timestamp_ms + snap.last_trade.sip_timestamp = sip_timestamp_ns return snap @@ -34,8 +41,8 @@ async def test_poll_updates_cache(self): source._client = MagicMock() # Satisfy the _poll_once guard mock_snapshots = [ - _make_snapshot("AAPL", 190.50, 1707580800000), - _make_snapshot("GOOGL", 175.25, 1707580800000), + _make_snapshot("AAPL", 190.50, 1707580800000000000), + _make_snapshot("GOOGL", 175.25, 1707580800000000000), ] with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): @@ -55,7 +62,7 @@ async def test_malformed_snapshot_skipped(self): source._tickers = ["AAPL", "BAD"] source._client = MagicMock() # Satisfy the _poll_once guard - good_snap = _make_snapshot("AAPL", 190.50, 1707580800000) + good_snap = _make_snapshot("AAPL", 190.50, 1707580800000000000) bad_snap = MagicMock() bad_snap.ticker = "BAD" bad_snap.last_trade = None # Will cause AttributeError @@ -84,7 +91,7 @@ async def test_api_error_does_not_crash(self): assert cache.get_price("AAPL") is None # No update happened async def test_timestamp_conversion(self): - """Test that timestamps are converted from milliseconds to seconds.""" + """Test that sip_timestamp is converted from nanoseconds to seconds.""" cache = PriceCache() source = MassiveDataSource( api_key="test-key", @@ -94,7 +101,7 @@ async def test_timestamp_conversion(self): source._tickers = ["AAPL"] source._client = MagicMock() # Satisfy the _poll_once guard - mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)] + mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000000000)] with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): await source._poll_once() @@ -189,7 +196,7 @@ async def test_start_immediate_poll(self): cache = PriceCache() source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) - mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)] + mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000000000)] with patch("app.market.massive_client.RESTClient"): with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): diff --git a/independent-reviewer/hooks/hooks.json b/independent-reviewer/hooks/hooks.json new file mode 100644 index 00000000..c71269be --- /dev/null +++ b/independent-reviewer/hooks/hooks.json @@ -0,0 +1,21 @@ +{ + "enabledPlugins": { + "frontend-design@claude-plugins-official": true, + "context7@claude-plugins-official": true, + "playwright@claude-plugins-official": true + }, + "hooks": { + "Stop": [ + { + "hooks": [ + { + "type": "command", + "command": "echo \"$(date '+%Y-%m-%d %H:%M:%S') hook fired\" >> /tmp/claude-stop-hook-debug.log; set -a; source /Users/rickhill65/Cursor_Projects/finally/.env 2>/dev/null; set +a; /Users/rickhill65/.hermes/node/bin/codex exec \"Review changes since last commit and write results to a file named planning/REVIEWHOOK.md\" < /dev/null 2>/dev/null || true", + "async": true, + "timeout": 300 + } + ] + } + ] + } +} \ No newline at end of file diff --git a/independent-reviewer/plugin.json b/independent-reviewer/plugin.json new file mode 100644 index 00000000..55f25453 --- /dev/null +++ b/independent-reviewer/plugin.json @@ -0,0 +1,6 @@ +{ + "name": "independent-reviewer", + "description": "carry out an independent review of all changes since last commit", + "version": "1.0.0" + +} \ No newline at end of file diff --git a/planning/MARKET_DATA_DESIGN.md b/planning/MARKET_DATA_DESIGN.md new file mode 100644 index 00000000..d1ea84f2 --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,1447 @@ +# Market Data Backend — Detailed Design + +Implementation-ready design for the FinAlly market data subsystem: the unified interface, in-memory price cache, GBM simulator, Massive API client, SSE streaming endpoint, and FastAPI lifecycle integration. + +**Status:** This describes the subsystem **as implemented** in `backend/app/market/` (verified against source, 2026-07-03). It consolidates and supersedes the three standalone docs `planning/MARKET_INTERFACE.md`, `planning/MARKET_SIMULATOR.md`, and `planning/MASSIVE_API.md` into one implementation-ready reference; those documents remain available for narrower deep-dives (interface design rationale, GBM math derivation, and Massive REST API reference respectively). It also supersedes `planning/archive/MARKET_DATA_DESIGN.md`, a pre-implementation sketch that contained bugs fixed during implementation — see [Appendix: Corrections vs. the Archived Sketch](#appendix-corrections-vs-the-archived-sketch). + +Everything below lives under `backend/app/market/`, backed by 73 passing tests in `backend/tests/market/` (84% coverage). + +--- + +## Table of Contents + +1. [Architecture Overview](#1-architecture-overview) +2. [File Structure](#2-file-structure) +3. [Data Model — `models.py`](#3-data-model--modelspy) +4. [Price Cache — `cache.py`](#4-price-cache--cachepy) +5. [Abstract Interface — `interface.py`](#5-abstract-interface--interfacepy) +6. [Seed Prices & Ticker Parameters — `seed_prices.py`](#6-seed-prices--ticker-parameters--seed_pricespy) +7. [GBM Simulator — `simulator.py`](#7-gbm-simulator--simulatorpy) +8. [Massive API Client — `massive_client.py`](#8-massive-api-client--massive_clientpy) +9. [Factory — `factory.py`](#9-factory--factorypy) +10. [SSE Streaming Endpoint — `stream.py`](#10-sse-streaming-endpoint--streampy) +11. [Public Package API — `__init__.py`](#11-public-package-api--__init__py) +12. [FastAPI Lifecycle Integration](#12-fastapi-lifecycle-integration) +13. [Watchlist Coordination](#13-watchlist-coordination) +14. [Testing Strategy](#14-testing-strategy) +15. [Error Handling & Edge Cases](#15-error-handling--edge-cases) +16. [Configuration Summary](#16-configuration-summary) +17. [Appendix: Corrections vs. the Archived Sketch](#appendix-corrections-vs-the-archived-sketch) + +--- + +## 1. Architecture Overview + +``` + MarketDataSource (ABC) + start / stop / add_ticker / remove_ticker / get_tickers + │ + ┌──────────┴──────────┐ + │ │ +SimulatorDataSource MassiveDataSource + (GBM, default, (Polygon.io REST poller, + no API key needed) used when MASSIVE_API_KEY set) + │ │ + └──────────┬──────────┘ + ▼ + PriceCache + (thread-safe, in-memory, + single point of truth) + │ + ┌─────────┼─────────────┐ + ▼ ▼ ▼ + SSE stream Portfolio Trade execution + /api/stream valuation (reads current price) + /prices +``` + +**Strategy pattern**: both data sources implement the same `MarketDataSource` ABC. Everything downstream — SSE streaming, portfolio valuation, trade execution — is source-agnostic; it only ever talks to the `PriceCache`. + +**Push model, not pull**: data sources write to the cache on their own schedule (500ms for the simulator, 15s for Massive free tier). The SSE layer polls the cache on its own fixed ~500ms cadence, independent of how often the underlying source actually updates. This decouples timing entirely — the SSE loop never needs to know which source is active. + +--- + +## 2. File Structure + +``` +backend/ + app/ + market/ + __init__.py # Public API re-exports + 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 constants + simulator.py # GBMSimulator (math) + SimulatorDataSource (async wrapper) + massive_client.py # MassiveDataSource + factory.py # create_market_data_source() + stream.py # create_stream_router() — SSE endpoint factory + tests/ + market/ + test_models.py # 11 tests, 100% coverage + test_cache.py # 13 tests, 100% coverage + test_simulator.py # 17 tests, 98% coverage + test_simulator_source.py # 10 tests (integration) + test_factory.py # 7 tests, 100% coverage + test_massive.py # 13 tests, 56% coverage (API methods mocked) + market_data_demo.py # Rich terminal live-dashboard demo +``` + +`simulator.py` holds both `GBMSimulator` (pure math, no I/O) and `SimulatorDataSource` (the async `MarketDataSource` wrapper) in one file — the wrapper is thin enough that it has no independent reason to live elsewhere. `seed_prices.py` holds only constant dictionaries/sets, no logic. + +--- + +## 3. Data Model — `models.py` + +`PriceUpdate` is the **only** data structure that leaves the market data layer. Every downstream consumer — SSE streaming, portfolio valuation, trade execution — works exclusively with this type (or its `.to_dict()` form). + +```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 defensive copying. +- **`slots=True`** — minor memory optimization; many of these are allocated per second. +- **Computed properties**, not stored fields — `change`, `change_percent`, and `direction` are derived from `price`/`previous_price` on access, so they can never drift out of sync with the underlying values. +- **`to_dict()`** — the single serialization point used by both the SSE endpoint and any future REST responses. + +--- + +## 4. Price Cache — `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 streaming loop polls the cache every ~500ms. Without a version counter it would re-serialize and re-send all prices every tick even when nothing changed (e.g., while using Massive, which only updates every 15s). The counter lets the SSE loop skip redundant sends: + +```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) +``` + +**Why `threading.Lock`, not `asyncio.Lock`?** The Massive client's synchronous `get_snapshot_all()` call runs via `asyncio.to_thread()` — a real OS thread, which `asyncio.Lock` would not protect against. `threading.Lock` works correctly from both a sync thread and the async event loop, so it's the only choice that covers both writers. + +**Note on `version`:** the property reads `self._version` without acquiring the lock. On CPython (GIL), a single `int` read is atomic, so this is safe today. It would need to move under the lock if this project ever ran on a no-GIL build (PEP 703 / Python 3.13t+) — not a concern for the current target of Python 3.12. + +--- + +## 5. Abstract Interface — `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.""" +``` + +The interface deliberately does **not** return prices from any method — it only pushes updates into a shared cache on its own timing. This is what lets a 500ms simulator and a 15s Massive poller sit behind the identical contract. + +--- + +## 6. Seed Prices & Ticker Parameters — `seed_prices.py` + +Constants only — no logic, no imports beyond stdlib. Shared by the simulator for initial prices, GBM parameters, and correlation structure. + +```python +"""Seed prices and per-ticker parameters for the market simulator.""" + +# Realistic starting prices for the default watchlist (as of project creation) +SEED_PRICES: dict[str, float] = { + "AAPL": 190.00, + "GOOGL": 175.00, + "MSFT": 420.00, + "AMZN": 185.00, + "TSLA": 250.00, + "NVDA": 800.00, + "META": 500.00, + "JPM": 195.00, + "V": 280.00, + "NFLX": 600.00, +} + +# Per-ticker GBM parameters +# sigma: annualized volatility (higher = more price movement) +# mu: annualized drift / expected return +TICKER_PARAMS: dict[str, dict[str, float]] = { + "AAPL": {"sigma": 0.22, "mu": 0.05}, + "GOOGL": {"sigma": 0.25, "mu": 0.05}, + "MSFT": {"sigma": 0.20, "mu": 0.05}, + "AMZN": {"sigma": 0.28, "mu": 0.05}, + "TSLA": {"sigma": 0.50, "mu": 0.03}, # High volatility + "NVDA": {"sigma": 0.40, "mu": 0.08}, # High volatility, strong drift + "META": {"sigma": 0.30, "mu": 0.05}, + "JPM": {"sigma": 0.18, "mu": 0.04}, # Low volatility (bank) + "V": {"sigma": 0.17, "mu": 0.04}, # Low volatility (payments) + "NFLX": {"sigma": 0.35, "mu": 0.05}, +} + +# Default parameters for tickers not in the list above (dynamically added) +DEFAULT_PARAMS: dict[str, float] = {"sigma": 0.25, "mu": 0.05} + +# Correlation groups for the simulator's Cholesky decomposition +# Tickers in the same group have higher intra-group correlation +CORRELATION_GROUPS: dict[str, set[str]] = { + "tech": {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"}, + "finance": {"JPM", "V"}, +} + +# Correlation coefficients +INTRA_TECH_CORR = 0.6 # Tech stocks move together +INTRA_FINANCE_CORR = 0.5 # Finance stocks move together +CROSS_GROUP_CORR = 0.3 # Between sectors / unknown tickers +TSLA_CORR = 0.3 # TSLA does its own thing +``` + +Tickers added dynamically that aren't in `SEED_PRICES` start at `random.uniform(50.0, 300.0)` and use `DEFAULT_PARAMS` — see `GBMSimulator._add_ticker_internal` below. + +--- + +## 7. GBM Simulator — `simulator.py` + +Two classes live here: `GBMSimulator` (pure math engine, synchronous, no asyncio/I/O) and `SimulatorDataSource` (the `MarketDataSource` implementation that drives it on a timer and writes to the cache). + +### 7.1 The Math + +Prices evolve under **Geometric Brownian Motion** — the standard lognormal model underlying Black-Scholes: + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +- `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 random draw + +`dt` is derived from a trading-calendar assumption, not hardcoded: + +```python +TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 (252 trading days, 6.5h/day) +DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8, for 500ms ticks +``` + +This tiny `dt` produces small, sub-cent moves per tick that accumulate naturally into realistic intraday ranges — and since GBM is multiplicative (`exp()` is always positive), prices can never go negative. + +### 7.2 Correlated Moves via Cholesky Decomposition + +Real stocks don't move independently. Given a correlation matrix `C`, compute `L = cholesky(C)`; then for independent standard normals `Z_independent`, `Z_correlated = L @ Z_independent` produces draws with the desired pairwise correlations. + +Correlation lookup checks TSLA **before** sector membership — this ordering matters: + +```python +@staticmethod +def _pairwise_correlation(t1: str, t2: str) -> float: + """Determine correlation between two tickers based on sector grouping. + + Correlation structure: + - Same tech sector: 0.6 + - Same finance sector: 0.5 + - TSLA with anything: 0.3 (it does its own thing) + - Cross-sector: 0.3 + - Unknown tickers: 0.3 + """ + tech = CORRELATION_GROUPS["tech"] + finance = CORRELATION_GROUPS["finance"] + + # TSLA is in tech set but behaves independently + if t1 == "TSLA" or t2 == "TSLA": + return TSLA_CORR + + if t1 in tech and t2 in tech: + return INTRA_TECH_CORR + if t1 in finance and t2 in finance: + return INTRA_FINANCE_CORR + + return CROSS_GROUP_CORR +``` + +`TSLA` is a member of `CORRELATION_GROUPS["tech"]` (so it still participates in the tech set for other purposes), but the design intent — "TSLA does its own thing, correlation ~0.3 with everything" — requires this TSLA check to run **before** the tech-membership check. If tech/finance membership were checked first, every TSLA/other-tech pair would match `t1 in tech and t2 in tech` and return `0.6`, making the TSLA branch unreachable dead code. + +### 7.3 Random Shock Events + +Every step, each ticker has a small probability of a sudden 2-5% move, for visual drama: + +```python +if random.random() < self._event_prob: # default 0.001 + shock_magnitude = random.uniform(0.02, 0.05) + shock_sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock_magnitude * shock_sign +``` + +At 0.1% chance per tick per ticker, with 10 tickers at 2 ticks/sec, expect a visible event roughly every 50 seconds. + +### 7.4 Full Implementation + +```python +"""GBM-based market simulator.""" + +from __future__ import annotations + +import asyncio +import logging +import math +import random + +import numpy as np + +from .cache import PriceCache +from .interface import MarketDataSource +from .seed_prices import ( + CORRELATION_GROUPS, + CROSS_GROUP_CORR, + DEFAULT_PARAMS, + INTRA_FINANCE_CORR, + INTRA_TECH_CORR, + SEED_PRICES, + TICKER_PARAMS, + TSLA_CORR, +) + +logger = logging.getLogger(__name__) + + +class GBMSimulator: + """Geometric Brownian Motion simulator for correlated stock prices.""" + + 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) # no Cholesky rebuild per-ticker + self._rebuild_cholesky() # ...one rebuild after batch init + + def step(self) -> dict[str, float]: + """Advance all tickers by one time step. Hot path — called every 500ms.""" + 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) + + 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: + """Rebuilds the Cholesky matrix — O(n^2), but n stays small (<50 tickers).""" + 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: + """Adds price/params state without rebuilding Cholesky — used for batch init.""" + 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: + 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 +``` + +`__init__` batches ticker setup via the private `_add_ticker_internal` (no Cholesky rebuild per call), followed by a single `_rebuild_cholesky()` — avoiding `O(n)` redundant `O(n^2)` rebuilds when constructing with the full default 10-ticker watchlist. The public `add_ticker()` rebuilds on every call since tickers are added one at a time after startup, and `n` stays small enough (<50) that this is not a performance concern. + +`step()` is kept allocation-light for its hot-path role: one `np.random.standard_normal(n)` call, one matrix-vector multiply, then a per-ticker loop with no further NumPy calls. + +### 7.5 `SimulatorDataSource` — the Async Wrapper + +`GBMSimulator` itself has no knowledge of `PriceCache` or asyncio. `SimulatorDataSource` owns the background task and writes results into the cache: + +```python +class SimulatorDataSource(MarketDataSource): + """MarketDataSource backed by the GBM simulator. + + Runs a background asyncio task that calls GBMSimulator.step() every + `update_interval` seconds and writes results to the PriceCache. + """ + + def __init__( + self, + price_cache: PriceCache, + update_interval: float = 0.5, + event_probability: float = 0.001, + ) -> None: + self._cache = price_cache + self._interval = update_interval + self._event_prob = event_probability + self._sim: GBMSimulator | None = None + self._task: asyncio.Task | None = None + + async def start(self, tickers: list[str]) -> None: + self._sim = GBMSimulator(tickers=tickers, event_probability=self._event_prob) + # Seed the cache with initial prices so SSE has data immediately + for ticker in tickers: + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + logger.info("Simulator started with %d tickers", len(tickers)) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("Simulator stopped") + + async def add_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.add_ticker(ticker) + # Seed cache immediately so the ticker has a price right away + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + logger.info("Simulator: added ticker %s", ticker) + + async def remove_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.remove_ticker(ticker) + self._cache.remove(ticker) + logger.info("Simulator: removed ticker %s", ticker) + + def get_tickers(self) -> list[str]: + return self._sim.get_tickers() if self._sim else [] + + async def _run_loop(self) -> None: + """Core loop: step the simulation, write to cache, sleep.""" + while True: + try: + if self._sim: + prices = self._sim.step() + for ticker, price in prices.items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") # loop survives a bad step + await asyncio.sleep(self._interval) +``` + +**Key behaviors:** + +- **Immediate seeding** — both `start()` and `add_ticker()` write an initial price into the cache synchronously, rather than waiting for the next scheduled tick. This matters for the frontend: a newly-added ticker shows a price immediately instead of a blank cell for up to 500ms. +- **Graceful cancellation** — `stop()` cancels the task and awaits it, swallowing `CancelledError`, for clean shutdown during FastAPI lifespan teardown. Safe to call more than once. +- **Exception resilience** — `_run_loop` catches exceptions per-tick so one bad step doesn't kill the background task; it logs and continues on the next interval. +- **`get_tickers()` delegates to `GBMSimulator.get_tickers()`** — a public method, not a reach into a private attribute (`self._sim._tickers`), keeping the class boundary clean. + +--- + +## 8. Massive API Client — `massive_client.py` + +Polls the Massive (formerly Polygon.io) REST API snapshot endpoint on a timer. `massive` is a core dependency (`pyproject.toml`), imported at module level — no lazy-import indirection. + +### 8.1 Endpoint Used + +**Snapshot — All Tickers**: `GET /v2/snapshot/locale/us/markets/stocks/tickers?tickers=AAPL,GOOGL,MSFT` — returns current data for multiple tickers in a **single API call**, which is essential for staying within the free tier's 5 req/min limit. + +```python +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +client = RESTClient(api_key="your_key_here") # or reads MASSIVE_API_KEY from env + +snapshots = client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "GOOGL", "MSFT"], +) +for snap in snapshots: + print(f"{snap.ticker}: ${snap.last_trade.price}") +``` + +Relevant `TickerSnapshot` fields (from `massive/rest/models/snapshot.py`): `ticker`, `last_trade` (has `.price`, `.sip_timestamp` in **nanoseconds**), `day` / `prev_day` (OHLCV `Agg` objects), `last_quote`, `todays_change_percent`. + +**Two gotchas verified against the real client:** +- There is no `day.previous_close` field — previous close lives on the sibling `prev_day.close`. +- `LastTrade` has **no `.timestamp` attribute** — only `sip_timestamp` / `participant_timestamp` / `trf_timestamp`, all in nanoseconds, not milliseconds. Getting this wrong silently breaks the price feed (see §8.4). + +### 8.2 Full Implementation + +```python +"""Massive (Polygon.io) API client for real market data.""" + +from __future__ import annotations + +import asyncio +import logging + +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +from .cache import PriceCache +from .interface import MarketDataSource + +logger = logging.getLogger(__name__) + + +class MassiveDataSource(MarketDataSource): + """MarketDataSource backed by the Massive (Polygon.io) REST API. + + Polls GET /v2/snapshot/locale/us/markets/stocks/tickers for all watched + tickers in a single API call, then writes results to the PriceCache. + + Rate limits: + - Free tier: 5 req/min → poll every 15s (default) + - Paid tiers: higher limits → poll every 2-5s + """ + + def __init__( + self, + api_key: str, + price_cache: PriceCache, + poll_interval: float = 15.0, + ) -> None: + self._api_key = api_key + self._cache = price_cache + self._interval = poll_interval + self._tickers: list[str] = [] + self._task: asyncio.Task | None = None + self._client: RESTClient | None = None + + async def start(self, tickers: list[str]) -> None: + self._client = RESTClient(api_key=self._api_key) + self._tickers = list(tickers) + + # Do an immediate first poll so the cache has data right away + await self._poll_once() + + self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") + logger.info( + "Massive poller started: %d tickers, %.1fs interval", + len(tickers), + self._interval, + ) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + self._client = None + logger.info("Massive poller stopped") + + async def add_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + if ticker not in self._tickers: + self._tickers.append(ticker) + logger.info("Massive: added ticker %s (will appear on next poll)", ticker) + + async def remove_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + self._tickers = [t for t in self._tickers if t != ticker] + self._cache.remove(ticker) + logger.info("Massive: removed ticker %s", ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + async def _poll_loop(self) -> None: + """Poll on interval. First poll already happened in start().""" + while True: + await asyncio.sleep(self._interval) + await self._poll_once() + + async def _poll_once(self) -> None: + """Execute one poll cycle: fetch snapshots, update cache.""" + if not self._tickers or not self._client: + return + + try: + # The Massive RESTClient is synchronous — run in a thread to + # avoid blocking the event loop. + snapshots = await asyncio.to_thread(self._fetch_snapshots) + processed = 0 + for snap in snapshots: + try: + price = snap.last_trade.price + # Massive trade timestamps (sip_timestamp) are Unix nanoseconds → convert to seconds + timestamp = snap.last_trade.sip_timestamp / 1_000_000_000 + self._cache.update(ticker=snap.ticker, price=price, timestamp=timestamp) + processed += 1 + except (AttributeError, TypeError) as e: + logger.warning( + "Skipping snapshot for %s: %s", getattr(snap, "ticker", "???"), e + ) + logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers)) + + except Exception as e: + logger.error("Massive poll failed: %s", e) + # Don't re-raise — the loop will retry 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, + ) +``` + +### 8.3 Error Handling Philosophy + +| Error | Behavior | +|-------|----------| +| **401 Unauthorized** (bad key) | Logged as error via the outer `except Exception`. Poller keeps running; fixing `.env` requires a 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 for one ticker** | Caught by the inner `except (AttributeError, TypeError)`; that ticker is skipped with a warning, others still processed. | +| **All tickers fail** | Cache retains last-known prices. SSE keeps streaming stale data — better than no data. | + +`_poll_once` deliberately nests two exception scopes: the inner one isolates a single bad snapshot so a malformed field on one ticker doesn't discard the whole batch; the outer one isolates the entire poll cycle (network failure, auth failure) so the background task survives indefinitely and simply retries. + +### 8.4 The Timestamp Bug (Resolved) — Why the Inner `except` Is Load-Bearing + +An earlier version of `_poll_once` read `snap.last_trade.timestamp / 1000.0` (assuming milliseconds). The real `LastTrade` model has no `.timestamp` attribute at all — only `sip_timestamp` (nanoseconds). Against the live API this raised `AttributeError` **inside the per-snapshot try/except**, which logged a warning and silently skipped the cache update for every ticker, on every poll. With a real `MASSIVE_API_KEY` configured, the price cache would never populate and the SSE stream would sit frozen — with no crash, no obvious error, just silence. + +This shipped because `test_massive.py`'s mock built `last_trade` as a bare `MagicMock()`, which happily accepts a `.timestamp` attribute that doesn't exist on the real class — the test validated behavior against a shape the live API never returns. + +**Fix applied**: read `snap.last_trade.sip_timestamp / 1_000_000_000` (nanoseconds → seconds). The test helper now builds `last_trade` as `MagicMock(spec_set=LastTrade)` using the real model class from `massive.rest.models.trades`: + +```python +from massive.rest.models.trades import LastTrade + +def _make_snapshot(ticker: str, price: float, sip_timestamp_ns: int) -> MagicMock: + snap = MagicMock() + snap.ticker = ticker + snap.last_trade = MagicMock(spec_set=LastTrade) + snap.last_trade.price = price + snap.last_trade.sip_timestamp = sip_timestamp_ns + return snap +``` + +`spec_set` rejects reading *or* setting any attribute not on the real class, so a future regression back to `.timestamp` fails the test suite immediately instead of silently mocking a field that doesn't exist. This is the general lesson for any test that mocks a third-party API response shape: spec the mock against the real model class, not a bare `MagicMock()`. + +--- + +## 9. Factory — `factory.py` + +Selects the data source at startup based on environment, with no lazy-import indirection (both implementations are imported at module level since `massive` is a core dependency). + +```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", ...] +``` + +--- + +## 10. SSE Streaming Endpoint — `stream.py` + +A FastAPI route that holds open a long-lived HTTP connection and pushes price updates as `text/event-stream`. + +```python +"""SSE streaming endpoint for live price updates.""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import AsyncGenerator + +from fastapi import APIRouter, Request +from fastapi.responses import StreamingResponse + +from .cache import PriceCache + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/stream", tags=["streaming"]) + + +def create_stream_router(price_cache: PriceCache) -> APIRouter: + """Create the SSE streaming router with a reference to the price cache. + + This factory pattern lets us inject the PriceCache without globals. + """ + + @router.get("/prices") + async def stream_prices(request: Request) -> StreamingResponse: + """SSE endpoint for live price updates. + + Streams all tracked ticker prices every ~500ms. The client connects + with EventSource and receives events in the format: + + data: {"AAPL": {"ticker": "AAPL", "price": 190.50, ...}, ...} + + Includes a retry directive so the browser auto-reconnects on + disconnection (EventSource built-in behavior). + """ + return StreamingResponse( + _generate_events(price_cache, request), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", # Disable nginx buffering if proxied + }, + ) + + return router + + +async def _generate_events( + price_cache: PriceCache, + request: Request, + interval: float = 0.5, +) -> AsyncGenerator[str, None]: + """Async generator that yields SSE-formatted price events. + + Sends all prices every `interval` seconds. Stops when the client + disconnects (detected via request.is_disconnected()). + """ + yield "retry: 1000\n\n" # Tell the browser to retry after 1s if the connection drops + + last_version = -1 + client_ip = request.client.host if request.client else "unknown" + logger.info("SSE client connected: %s", client_ip) + + try: + while True: + if await request.is_disconnected(): + logger.info("SSE client disconnected: %s", client_ip) + break + + current_version = price_cache.version + if current_version != last_version: + last_version = current_version + prices = price_cache.get_all() + if prices: + data = {ticker: update.to_dict() for ticker, update in prices.items()} + yield f"data: {json.dumps(data)}\n\n" + + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("SSE stream cancelled for: %s", client_ip) +``` + +### Wire Format + +``` +data: {"AAPL":{"ticker":"AAPL","price":190.50,"previous_price":190.42,"timestamp":1707580800.5,"change":0.08,"change_percent":0.042,"direction":"up"},"GOOGL":{...}} + +``` + +Client side: + +```javascript +const eventSource = new EventSource('/api/stream/prices'); +eventSource.onmessage = (event) => { + const prices = JSON.parse(event.data); + // prices is { "AAPL": { ticker, price, previous_price, ... }, ... } +}; +``` + +### Design Notes + +- **Poll-and-push, not event-driven.** The generator 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 into sparkline charts — regular spacing keeps the visualization clean. +- **Version-gated emission vs. `planning/PLAN.md` §6.** The spec calls for the server to push at a fixed ~500ms cadence "regardless of the market data source," repeating the last-known price between Massive updates so the connection and frontend state stay consistent even when the underlying price hasn't changed for up to 15s. The implementation above is a deliberate refinement: it only emits an SSE `data:` event when `price_cache.version` has actually changed, rather than resending an identical payload every tick. The connection itself stays alive regardless (via the `while True` / `is_disconnected()` loop running every `interval`), and `EventSource`'s own connection handling doesn't care about event frequency — so the spec's intent (connection resilience, consistent frontend state) is preserved while avoiding wasted bandwidth on genuinely-unchanged prices. The tradeoff: a client relying on receiving an event every single tick (rather than every genuine change) would need to be aware prices only arrive when they move. +- **`request.is_disconnected()`** is checked every loop iteration so a closed client connection tears down the generator (and its logging) promptly rather than looping forever into the void. +- **`X-Accel-Buffering: no`** disables response buffering if this ever sits behind an nginx reverse proxy — otherwise nginx could buffer the stream and defeat real-time delivery. + +--- + +## 11. Public Package API — `__init__.py` + +```python +"""Market data subsystem for FinAlly. + +Public API: + PriceUpdate - Immutable price snapshot dataclass + PriceCache - Thread-safe in-memory price store + MarketDataSource - Abstract interface for data providers + create_market_data_source - Factory that selects simulator or Massive + create_stream_router - FastAPI router factory for SSE endpoint +""" + +from .cache import PriceCache +from .factory import create_market_data_source +from .interface import MarketDataSource +from .models import PriceUpdate +from .stream import create_stream_router + +__all__ = [ + "PriceUpdate", + "PriceCache", + "MarketDataSource", + "create_market_data_source", + "create_stream_router", +] +``` + +The rest of the backend imports exclusively from `app.market`, never reaching into submodules: + +```python +from app.market import PriceCache, PriceUpdate, MarketDataSource, create_market_data_source, create_stream_router +``` + +--- + +## 12. FastAPI Lifecycle Integration + +The market data system does not yet have a `main.py` to integrate into — the rest of the backend (portfolio, watchlist, chat routes, `main.py`) is still to be built per `planning/PLAN.md`. This section is prescriptive guidance for whoever wires it up, using the `lifespan` context manager pattern. + +```python +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from app.market import PriceCache, MarketDataSource, create_market_data_source, create_stream_router + + +@asynccontextmanager +async def lifespan(app: FastAPI): + """Manage startup and shutdown of background services.""" + + # --- STARTUP --- + + # 1. Create the shared price cache + price_cache = PriceCache() + app.state.price_cache = price_cache + + # 2. Create and start the market data source + source = create_market_data_source(price_cache) + app.state.market_source = source + + # 3. Load initial tickers from the database watchlist (lazy-init DB first) + initial_tickers = await load_watchlist_tickers() # reads from SQLite + await source.start(initial_tickers) + + # 4. Register the SSE streaming router + app.include_router(create_stream_router(price_cache)) + + yield # App is running + + # --- SHUTDOWN --- + await source.stop() + + +app = FastAPI(title="FinAlly", lifespan=lifespan) + + +def get_price_cache() -> PriceCache: + return app.state.price_cache + + +def get_market_source() -> MarketDataSource: + return app.state.market_source +``` + +Other routes access the cache and source via FastAPI 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), +): + # ... remove from watchlist table ... + await source.remove_ticker(ticker) +``` + +--- + +## 13. Watchlist Coordination + +When the watchlist changes (via REST API or LLM chat tool call), the market data source must be told so it tracks the right ticker set. + +**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 15s later) + → 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 still 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: + +```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) # only stop tracking if no open position + + return {"status": "ok"} +``` + +--- + +## 14. Testing Strategy + +**73 tests, all passing, 84% overall coverage.** Six modules under `backend/tests/market/`: + +| Module | Tests | Coverage | What it exercises | +|--------|-------|----------|--------------------| +| `test_models.py` | 11 | 100% | `PriceUpdate` computed properties, `to_dict()` | +| `test_cache.py` | 13 | 100% | `update`/`get`/`get_all`/`remove`, version counter, first-update flat direction | +| `test_simulator.py` | 17 | 98% | GBM math, correlation lookup, add/remove ticker, Cholesky rebuilds | +| `test_simulator_source.py` | 10 | (integration) | `SimulatorDataSource` lifecycle: seeding, start/stop idempotency, add/remove | +| `test_factory.py` | 7 | 100% | Env-var-driven source selection | +| `test_massive.py` | 13 | 56% (expected) | Polling, malformed-snapshot skip, error resilience, timestamp conversion | + +`massive_client.py`'s 56% coverage is expected, not a gap to close blindly — its API-calling methods are exercised against mocks (`_fetch_snapshots` patched), not the real Massive service, so the actual HTTP call path stays untested by design. + +### 14.1 `GBMSimulator` — representative tests + +```python +class TestGBMSimulator: + def test_step_returns_all_tickers(self): + sim = GBMSimulator(tickers=["AAPL", "GOOGL"]) + result = sim.step() + assert set(result.keys()) == {"AAPL", "GOOGL"} + + 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): + assert sim.step()["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 # 2 tickers, matrix now exists +``` + +### 14.2 `PriceCache` — representative tests + +```python +class TestPriceCache: + def test_first_update_is_flat(self): + cache = PriceCache() + update = cache.update("AAPL", 190.50) + assert update.direction == "flat" + assert update.previous_price == 190.50 + + def test_version_increments(self): + cache = PriceCache() + v0 = cache.version + cache.update("AAPL", 190.00) + assert cache.version == v0 + 1 +``` + +### 14.3 `SimulatorDataSource` — integration tests + +```python +@pytest.mark.asyncio +class TestSimulatorDataSource: + async def test_start_populates_cache(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL", "GOOGL"]) + # Cache has seed prices immediately — before the first loop tick + assert cache.get("AAPL") is not None + await source.stop() + + async def test_add_and_remove_ticker(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + await source.add_ticker("TSLA") + assert "TSLA" in source.get_tickers() + assert cache.get("TSLA") is not None + await source.remove_ticker("TSLA") + assert cache.get("TSLA") is None + await source.stop() +``` + +### 14.4 `MassiveDataSource` — mocked tests + +The mock discipline here is the load-bearing detail — see §8.4 for why: + +```python +from massive.rest.models.trades import LastTrade + +def _make_snapshot(ticker: str, price: float, sip_timestamp_ns: int) -> MagicMock: + """last_trade is spec'd against the real LastTrade model so that reading or + setting a nonexistent attribute (e.g. `.timestamp`) raises AttributeError + here too, instead of silently mocking a fake field.""" + snap = MagicMock() + snap.ticker = ticker + snap.last_trade = MagicMock(spec_set=LastTrade) + snap.last_trade.price = price + snap.last_trade.sip_timestamp = sip_timestamp_ns + return snap + + +@pytest.mark.asyncio +class TestMassiveDataSource: + async def test_poll_updates_cache(self): + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL", "GOOGL"] + source._client = MagicMock() # satisfy the _poll_once guard + + mock_snapshots = [ + _make_snapshot("AAPL", 190.50, 1707580800000000000), + _make_snapshot("GOOGL", 175.25, 1707580800000000000), + ] + with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): + await source._poll_once() + + assert cache.get_price("AAPL") == 190.50 + + async def test_timestamp_conversion(self): + """sip_timestamp (nanoseconds) is converted to seconds.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + with patch.object( + source, "_fetch_snapshots", + return_value=[_make_snapshot("AAPL", 190.50, 1707580800000000000)], + ): + await source._poll_once() + + assert cache.get("AAPL").timestamp == 1707580800.0 + + 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"] + source._client = MagicMock() + + good = _make_snapshot("AAPL", 190.50, 1707580800000000000) + bad = MagicMock() + bad.ticker = "BAD" + bad.last_trade = None # triggers AttributeError, caught per-snapshot + + with patch.object(source, "_fetch_snapshots", return_value=[good, bad]): + await source._poll_once() + + assert cache.get_price("AAPL") == 190.50 + assert cache.get_price("BAD") is None + + 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"] + source._client = MagicMock() + + 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 +``` + +Note `_client = MagicMock()` is set directly on the instance in each test rather than patching `RESTClient` at module import time — since `RESTClient` is now a top-level import (not lazy), this keeps tests independent of whether the `massive` package needs live network access to construct a client. + +### 14.5 Running the Suite + +```bash +cd backend +uv run --extra dev pytest -v # All tests +uv run --extra dev pytest --cov=app # With coverage +uv run --extra dev ruff check app/ tests/ # Lint +``` + +### 14.6 Demo + +A Rich terminal demo exercises the full stack end-to-end (simulator → cache → live dashboard), useful as a manual sanity check outside of pytest: + +```bash +cd backend +uv run market_data_demo.py +``` + +Displays all 10 default tickers with sparklines, color-coded direction arrows, and an event log for notable price moves; runs 60 seconds or until Ctrl+C. + +--- + +## 15. Error Handling & Edge Cases + +### 15.1 Startup with an Empty Watchlist + +If the database has no watchlist entries, `start()` receives an empty list. Both data sources handle this gracefully — `GBMSimulator.step()` returns `{}` for zero tickers, and `MassiveDataSource._poll_once()` returns immediately when `self._tickers` is empty. The SSE endpoint sends no events (`if prices:` guard). When a ticker is later added, the source begins tracking it immediately. + +### 15.2 Price Cache Miss During Trade + +If a user (or the LLM) tries to trade a ticker with no cached price yet (just added, Massive hasn't polled): + +```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 in practice by seeding the cache synchronously in `add_ticker()`. The Massive client may have a real gap of up to `poll_interval` seconds — the 400 with a clear message is the correct response rather than trading at a stale or fabricated price. + +### 15.3 Massive API Key Invalid + +If the key is set but wrong, the first poll fails with 401, caught by `_poll_once`'s outer `except Exception`. The poller logs the error and keeps retrying every `poll_interval` seconds — it does not stop or crash. The SSE endpoint keeps streaming (connected, but empty/stale data). The user sees a green connection indicator with no prices; the fix is correcting `MASSIVE_API_KEY` and restarting. + +### 15.4 Thread Safety Under Load + +`PriceCache` uses `threading.Lock`, a mutex — one thread at a time. Under normal load (10 tickers, 2 updates/sec from the simulator, or one `asyncio.to_thread` write per 15s from Massive), contention is negligible; the critical section is a dict lookup plus assignment. A `ReadWriteLock` would only matter at a scale (hundreds of tickers, many concurrent SSE readers) well beyond this project's scope. + +### 15.5 Simulator Numerical Behavior + +- Prices are rounded to 2 decimals in `GBMSimulator.step()`; the exponential formulation (`exp(drift + diffusion)`) is numerically stable and always positive. +- The correlation matrix is guaranteed positive semi-definite for `np.linalg.cholesky` to succeed, since all pairwise correlations are fixed positive constants (0.3–0.6) with a diagonal of 1.0 — comfortably within the valid PSD range for any `n`. +- Adding/removing a ticker mid-session triggers an `O(n^2)` Cholesky rebuild — not a concern while `n` stays under ~50 tickers. + +--- + +## 16. Configuration Summary + +| Parameter | Location | Default | Description | +|-----------|----------|---------|--------------| +| `MASSIVE_API_KEY` | Environment variable | `""` (empty) | If set and non-empty, use Massive API; otherwise use the simulator | +| `update_interval` | `SimulatorDataSource.__init__` | `0.5` (seconds) | Time between simulator ticks | +| `event_probability` | `GBMSimulator.__init__` | `0.001` | Chance of a random shock event per ticker per tick | +| `dt` | `GBMSimulator.__init__` | `~8.48e-8` | GBM time step (fraction of a trading year) | +| `poll_interval` | `MassiveDataSource.__init__` | `15.0` (seconds) | Time between Massive API polls (free-tier safe) | +| SSE push interval | `_generate_events()` | `0.5` (seconds) | Time between cache-version checks / pushes | +| SSE retry directive | `_generate_events()` | `1000` (ms) | Browser `EventSource` reconnection delay | + +--- + +## Appendix: Corrections vs. the Archived Sketch + +`planning/archive/MARKET_DATA_DESIGN.md` was written before implementation began. A code review (`planning/archive/MARKET_DATA_REVIEW.md`) caught the following, all fixed in the shipped code reflected throughout this document: + +1. **`pyproject.toml` build config** — the sketch never specified `[tool.hatch.build.targets.wheel] packages = ["app"]`; without it `uv sync` fails with "Unable to determine which files to ship inside the wheel." Added. +2. **Lazy imports of `massive` removed** — the sketch imported `from massive import RESTClient` inside `start()` to make the dependency optional. Since `massive` is a core `pyproject.toml` dependency regardless, this added indirection with no benefit and made tests fragile (patching a name that doesn't exist at module level without `RESTClient` already installed). Shipped code imports at module level. +3. **`_generate_events` return type** — the sketch annotated `-> None` on an async generator; shipped code uses `-> AsyncGenerator[str, None]`. +4. **`GBMSimulator.get_tickers()`** — the sketch had `SimulatorDataSource.get_tickers()` reach into the private `self._sim._tickers`. Shipped code adds a public `GBMSimulator.get_tickers()` method. +5. **`DEFAULT_CORR` removed** — the sketch defined an unused `DEFAULT_CORR = 0.3` alongside `CROSS_GROUP_CORR = 0.3`, which `_pairwise_correlation` actually returns for the fallback case. Same value, confusingly named twice; consolidated into `CROSS_GROUP_CORR` only. +6. **Test hygiene** — unused `pytest`/`math`/`asyncio` imports removed from four test files. +7. **The timestamp bug** — see §8.4 above; this was the one substantive logic bug, not just a sketch-vs-shipped discrepancy, and it's the main reason `test_massive.py` now specs its mocks against the real `LastTrade` class. + +All 73 tests pass with these fixes applied; see `planning/MARKET_DATA_SUMMARY.md` for the full review and fix history. diff --git a/planning/MARKET_INTERFACE.md b/planning/MARKET_INTERFACE.md new file mode 100644 index 00000000..1d617e3f --- /dev/null +++ b/planning/MARKET_INTERFACE.md @@ -0,0 +1,264 @@ +# Market Data Interface Design + +Unified Python interface for market data in FinAlly. Two implementations (simulator and Massive API) behind one abstract interface. All downstream code — SSE streaming, price cache, portfolio valuation — is source-agnostic. + +This document describes the interface **as implemented** in `backend/app/market/`. It supersedes `planning/archive/MARKET_INTERFACE.md`, which was a pre-implementation sketch; a few details below (the `PriceUpdate` shape, cache versioning, immediate-first-poll behavior) evolved during implementation and are captured here. + +## Core Data Model + +`backend/app/market/models.py`: + +```python +from dataclasses import dataclass, field +import time + +@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: + return round(self.price - self.previous_price, 4) + + @property + def change_percent(self) -> float: + 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: + 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.""" + ... +``` + +`change`, `change_percent`, and `direction` are **computed properties**, not stored fields — derived once from `price`/`previous_price` rather than duplicated as separate cache-mutable state. This is the only data structure that leaves the market data layer; everything downstream works with `PriceUpdate` objects (or their `.to_dict()` form over SSE). + +## Abstract Interface + +`backend/app/market/interface.py`: + +```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 never calls the data source directly for prices — + it reads from the cache. + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing price updates for the given tickers. + Must be called exactly once.""" + + @abstractmethod + async def stop(self) -> None: + """Stop the background task and release resources. Safe to call multiple times.""" + + @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 from the active set. Also removes it from the PriceCache.""" + + @abstractmethod + def get_tickers(self) -> list[str]: + """Return the current list of actively tracked tickers.""" +``` + +Both implementations write to a shared `PriceCache`. The interface does **not** return prices directly — it pushes updates into the cache on its own schedule. + +## Price Cache + +`backend/app/market/cache.py`. Thread-safe (guarded by a `threading.Lock`, not an asyncio lock — both `SimulatorDataSource`'s event-loop task and any future non-async writer can share it safely) in-memory store that data sources write to and the SSE streamer reads from. + +```python +class PriceCache: + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._lock = Lock() + self._version: int = 0 # bumped on every update + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + """Record a new price. Computes previous_price from the prior entry + (or == price on first write, so direction is 'flat' initially).""" + ... + + def get(self, ticker: str) -> PriceUpdate | None: ... + def get_all(self) -> dict[str, PriceUpdate]: ... # shallow copy snapshot + def get_price(self, ticker: str) -> float | None: ... # convenience accessor + def remove(self, ticker: str) -> None: ... + + @property + def version(self) -> int: + """Monotonically increasing counter, bumped on every update() call.""" +``` + +The `version` counter exists specifically so the SSE endpoint can cheaply detect "did anything change since I last checked" without diffing dictionaries — see Integration with SSE below. This is a refinement over the original sketch, which had no change-detection mechanism. + +## Factory Function + +`backend/app/market/factory.py`. Selects the data source at startup based on environment: + +```python +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """MASSIVE_API_KEY set and non-empty -> MassiveDataSource. Otherwise -> SimulatorDataSource. + Returns an unstarted source; caller must await source.start(tickers).""" + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + + if api_key: + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + else: + return SimulatorDataSource(price_cache=price_cache) +``` + +## Massive Implementation + +`backend/app/market/massive_client.py` — `MassiveDataSource`. Polls `get_snapshot_all()` on a timer (default 15s, matching the free-tier rate limit; see `planning/MASSIVE_API.md`). + +Key implementation details not obvious from the interface alone: +- `start()` does an **immediate first poll** before starting the background loop, so the cache has data right away instead of waiting a full interval +- The synchronous `RESTClient` call runs via `asyncio.to_thread` so it doesn't block the event loop +- `_poll_once` wraps each snapshot's field extraction in `try/except (AttributeError, TypeError)` per-ticker, and the whole poll in a broader `try/except Exception` — a single bad snapshot or a transient API failure doesn't kill the poller +- The timestamp conversion reads `snap.last_trade.sip_timestamp / 1_000_000_000` (nanoseconds → seconds). This was previously `snap.last_trade.timestamp / 1000.0`, referencing a field that doesn't exist on the real Massive `LastTrade` model — see "Fixed Issue" in `planning/MASSIVE_API.md` for the history; it's now fixed and covered by a `spec_set`-mocked test. + +```python +async def _poll_once(self) -> None: + if not self._tickers or not self._client: + return + try: + 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.sip_timestamp / 1_000_000_000, # ns -> seconds + ) + except (AttributeError, TypeError) as e: + logger.warning("Skipping snapshot for %s: %s", getattr(snap, "ticker", "???"), e) + except Exception as e: + logger.error("Massive poll failed: %s", e) # retried on next interval +``` + +## Simulator Implementation + +`backend/app/market/simulator.py` — `SimulatorDataSource` wraps `GBMSimulator` (see `planning/MARKET_SIMULATOR.md` for the math). Runs a background asyncio task calling `.step()` every `update_interval` seconds (default 0.5s). + +```python +class SimulatorDataSource(MarketDataSource): + 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 add_ticker(self, ticker: str) -> None: + self._sim.add_ticker(ticker) + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) # seed immediately, don't wait for next tick + + async def _run_loop(self) -> None: + while True: + try: + prices = self._sim.step() + for ticker, price in prices.items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") # loop survives a bad step + await asyncio.sleep(self._interval) +``` + +Both `start()` and `add_ticker()` write an initial price into the cache synchronously rather than waiting for the next scheduled tick — this matters for the frontend, since a newly-added ticker should show a price immediately rather than a blank cell for up to 500ms. + +## Integration with SSE + +`backend/app/market/stream.py` — `create_stream_router(price_cache)` returns a FastAPI router exposing `GET /api/stream/prices`. The generator polls every 500ms but only emits an event when `price_cache.version` has changed since the last check, avoiding redundant identical payloads: + +```python +async def _generate_events(price_cache: PriceCache, request: Request, interval: float = 0.5): + yield "retry: 1000\n\n" + last_version = -1 + 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: + yield f"data: {json.dumps({t: u.to_dict() for t, u in prices.items()})}\n\n" + await asyncio.sleep(interval) +``` + +Per `planning/PLAN.md` §6, the server is meant to push at a fixed ~500ms cadence "regardless of the market data source," repeating the last-known price between Massive updates so the connection and frontend state stay consistent even when the underlying data hasn't changed for up to 15s. The version-gated implementation above is a deliberate refinement: it only emits a new SSE `data:` event when the price *actually* changed, rather than resending an identical payload every 500ms. This still satisfies the spec's intent (SSE connection stays alive via `is_disconnected()` polling every interval; the frontend's `EventSource` keeps its connection open regardless of event frequency) while avoiding wasted bandwidth on genuinely unchanged prices. It does mean the client won't receive a redundant heartbeat event every single tick — worth knowing if a future frontend change relies on tick-frequency events rather than change-frequency events. + +## File Structure + +``` +backend/ + app/ + market/ + __init__.py # Public API: PriceCache, PriceUpdate, MarketDataSource, + # create_market_data_source, create_stream_router + models.py # PriceUpdate dataclass + cache.py # PriceCache + interface.py # MarketDataSource ABC + factory.py # create_market_data_source() + massive_client.py # MassiveDataSource + simulator.py # GBMSimulator + SimulatorDataSource + seed_prices.py # SEED_PRICES, TICKER_PARAMS, correlation constants + stream.py # create_stream_router() — SSE endpoint factory +``` + +## Lifecycle + +1. **App startup**: Create `PriceCache`, call `create_market_data_source(price_cache)`, then `await source.start(initial_tickers)` +2. **Watchlist changes**: Call `source.add_ticker()` or `source.remove_ticker()` +3. **SSE streaming**: `create_stream_router(cache)` mounted once at startup; each client connection reads from `PriceCache` independently +4. **Trade execution**: Reads current price via `PriceCache.get_price(ticker)` +5. **App shutdown**: `await source.stop()` + +## Usage for Downstream Code + +```python +from app.market import PriceCache, create_market_data_source + +cache = PriceCache() +source = create_market_data_source(cache) # reads MASSIVE_API_KEY +await source.start(["AAPL", "GOOGL", "MSFT", ...]) + +update = cache.get("AAPL") # PriceUpdate or None +price = cache.get_price("AAPL") # float or None +all_prices = cache.get_all() # dict[str, PriceUpdate] + +await source.add_ticker("TSLA") +await source.remove_ticker("GOOGL") + +await source.stop() +``` + +## Test Coverage + +73 tests across 6 modules in `backend/tests/market/`, 84% overall coverage. `massive_client.py` sits at 56% — expected, since its API-calling methods are exercised against mocks rather than the real Massive service. That gap is what had let the timestamp bug described above ship undetected; `test_massive.py` now mocks `last_trade` with `spec_set=LastTrade` (the real model class) instead of a bare `MagicMock()`, so a future field-name mismatch like this one fails the test suite instead of passing silently. See `planning/MASSIVE_API.md` for the fix history. diff --git a/planning/MARKET_SIMULATOR.md b/planning/MARKET_SIMULATOR.md new file mode 100644 index 00000000..194e5850 --- /dev/null +++ b/planning/MARKET_SIMULATOR.md @@ -0,0 +1,270 @@ +# Market Simulator Design + +Approach and code structure for simulating realistic stock prices when no `MASSIVE_API_KEY` is configured. + +This document describes the simulator **as implemented** in `backend/app/market/simulator.py` and `seed_prices.py`. It supersedes `planning/archive/MARKET_SIMULATOR.md`, which was a pre-implementation sketch; one correction is called out below where the shipped code fixes a logic bug present in the original sketch. + +## Overview + +The simulator uses **Geometric Brownian Motion (GBM)** to generate realistic stock price paths — the standard model underlying Black-Scholes option pricing. Prices evolve continuously with random noise, can't go negative, and exhibit the lognormal distribution seen in real markets. + +Updates run at ~500ms intervals (`SimulatorDataSource`'s default `update_interval`), producing a continuous stream of price changes that feel alive. + +## GBM Math + +At each time step, a stock price evolves as: + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +Where: +- `S(t)` = current price +- `mu` = annualized drift (expected return), e.g. 0.05 (5%) +- `sigma` = annualized volatility, e.g. 0.20 (20%) +- `dt` = time step as a fraction of a trading year +- `Z` = correlated standard normal random variable + +`dt` is derived, not hardcoded, from the actual trading-calendar assumption: + +```python +TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 (252 trading days, 6.5h/day) +DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8, for 500ms ticks +``` + +This tiny `dt` produces small, sub-cent moves per tick that accumulate naturally over time into realistic intraday ranges. + +## Correlated Moves + +Real stocks don't move independently — tech stocks tend to move together, etc. The simulator uses a **Cholesky decomposition** of a correlation matrix to turn independent random draws into correlated ones. + +Given a correlation matrix `C`, compute `L = cholesky(C)`. Then for independent standard normals `Z_independent`: + +``` +Z_correlated = L @ Z_independent +``` + +Correlation groups, defined in `seed_prices.py`: + +```python +CORRELATION_GROUPS: dict[str, set[str]] = { + "tech": {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"}, + "finance": {"JPM", "V"}, +} + +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 +``` + +### Pairwise correlation lookup — and a bug the shipped code fixes + +`GBMSimulator._pairwise_correlation` checks TSLA **before** sector membership: + +```python +@staticmethod +def _pairwise_correlation(t1: str, t2: str) -> float: + tech = CORRELATION_GROUPS["tech"] + finance = CORRELATION_GROUPS["finance"] + + # TSLA is in the tech set but behaves independently — checked FIRST + 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 +``` + +`TSLA` is a member of the `"tech"` set (so it still participates in the default cross-group baseline), but the design intent — "TSLA does its own thing, correlation ~0.3 with everything" — requires the TSLA check to run *before* the tech-membership check. The archived sketch (`planning/archive/MARKET_SIMULATOR.md`) checked tech/finance membership first and TSLA last, which meant the TSLA branch was unreachable dead code: for any TSLA/other-tech pair, `t1_tech and t2_tech` would already be `True` (since TSLA ∈ tech) and return `0.6` before the TSLA-specific check was ever reached. The shipped implementation orders the checks correctly. + +## Random Events + +Every step, each ticker has a small probability (`event_probability`, default `0.001`) of a random event — a sudden 2-5% move — for visual drama: + +```python +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 +``` + +At 0.1% chance per tick per ticker, with 10 tickers at 2 ticks/sec, expect an event roughly every 50 seconds — frequent enough to keep the dashboard interesting without feeling chaotic. + +## Seed Prices & Per-Ticker Parameters + +`backend/app/market/seed_prices.py`: + +```python +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, +} + +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_PARAMS: dict[str, float] = {"sigma": 0.25, "mu": 0.05} +``` + +Tickers added dynamically (not in `SEED_PRICES`) start at `random.uniform(50.0, 300.0)` and use `DEFAULT_PARAMS`. + +## Implementation + +`GBMSimulator` (pure, synchronous — no asyncio, no I/O): + +```python +class GBMSimulator: + TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 + DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR + + def __init__(self, tickers, dt=DEFAULT_DT, event_probability=0.001): + self._dt = dt + self._event_prob = event_probability + self._tickers: list[str] = [] + self._prices: dict[str, float] = {} + self._params: dict[str, dict] = {} + self._cholesky: np.ndarray | None = None + for ticker in tickers: + self._add_ticker_internal(ticker) # no Cholesky rebuild per-ticker + self._rebuild_cholesky() # ...one rebuild after batch init + + def step(self) -> dict[str, float]: + """Advance all tickers by one time step. Hot path — called every 500ms.""" + n = len(self._tickers) + if n == 0: + return {} + + z_independent = np.random.standard_normal(n) + z = self._cholesky @ z_independent if self._cholesky is not None else z_independent + + result = {} + for i, ticker in enumerate(self._tickers): + mu, sigma = self._params[ticker]["mu"], self._params[ticker]["sigma"] + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z[i] + self._prices[ticker] *= math.exp(drift + diffusion) + + if random.random() < self._event_prob: + shock = random.uniform(0.02, 0.05) * random.choice([-1, 1]) + self._prices[ticker] *= (1 + shock) + + result[ticker] = round(self._prices[ticker], 2) + return result + + def add_ticker(self, ticker: str) -> None: + """Rebuilds the Cholesky matrix — O(n^2), but n stays small (<50 tickers).""" + 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: + """Adds price/params state without rebuilding Cholesky — used for batch init.""" + 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: + 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) +``` + +`__init__` batches ticker setup via the private `_add_ticker_internal` (no Cholesky rebuild) followed by a single `_rebuild_cholesky()` call — avoiding `O(n)` redundant `O(n^2)` rebuilds when constructing with the full default watchlist. The public `add_ticker()` rebuilds on every call since tickers are added one at a time after startup. + +## `SimulatorDataSource` — the async wrapper + +`GBMSimulator` itself is synchronous and has no knowledge of the `PriceCache` or asyncio. `SimulatorDataSource` (also in `simulator.py`) is the `MarketDataSource` implementation that owns the background task and writes results into the cache — see `planning/MARKET_INTERFACE.md` for its full lifecycle (including the immediate-seed-on-start/add-ticker behavior). In short: + +```python +class SimulatorDataSource(MarketDataSource): + def __init__(self, price_cache, 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) + for ticker in tickers: # seed cache before first tick + 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 _run_loop(self) -> None: + while True: + try: + for ticker, price in self._sim.step().items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") # loop survives a bad step + await asyncio.sleep(self._interval) +``` + +## File Structure + +``` +backend/ + app/ + market/ + simulator.py # GBMSimulator (pure math) + SimulatorDataSource (async wrapper) + seed_prices.py # SEED_PRICES, TICKER_PARAMS, DEFAULT_PARAMS, + # CORRELATION_GROUPS, INTRA_TECH_CORR, INTRA_FINANCE_CORR, + # CROSS_GROUP_CORR, TSLA_CORR +``` + +`seed_prices.py` holds only constant dictionaries/sets — no logic. `simulator.py` holds both the GBM math (`GBMSimulator`) and the `MarketDataSource` implementation that drives it on a timer (`SimulatorDataSource`); they're kept in one module because `SimulatorDataSource` is a thin wrapper with no independent reason to exist in its own file. + +## Behavior Notes + +- Prices never go negative — GBM is multiplicative (`exp()` is always positive) +- The tiny `dt` produces sub-cent moves per tick, which accumulate naturally over time into realistic-looking intraday ranges +- With `sigma=0.50` (TSLA), a simulated trading day produces roughly the right intraday range +- The correlation matrix must be positive semi-definite for `np.linalg.cholesky` to succeed — guaranteed here since all pairwise correlations are fixed positive constants (0.3-0.6) and the diagonal is 1.0, which keeps the matrix well within the valid PSD range for any `n` +- Random events happen ~0.1% of steps ≈ once every ~500 seconds per ticker; with 10 tickers, expect a visible event roughly every 50 seconds +- Adding/removing a ticker mid-session rebuilds the Cholesky matrix — `O(n^2)`, but `n` stays small (<50 tickers), so this is not a performance concern +- `GBMSimulator.step()` is the hot path (called every 500ms by `SimulatorDataSource._run_loop`) and is kept allocation-light: one `np.random.standard_normal(n)` call, one matrix-vector multiply, then a per-ticker loop with no further NumPy calls + +## Test Coverage + +`test_simulator.py` (17 tests, 98% coverage of `simulator.py`) and `test_simulator_source.py` (10 integration tests) in `backend/tests/market/` — see `planning/MARKET_DATA_SUMMARY.md` for the full test inventory. diff --git a/planning/MASSIVE_API.md b/planning/MASSIVE_API.md new file mode 100644 index 00000000..f9691bab --- /dev/null +++ b/planning/MASSIVE_API.md @@ -0,0 +1,276 @@ +# Massive API Reference + +Reference documentation for the Massive (formerly Polygon.io) REST API as used in FinAlly. Verified against `massive.com/docs` and the `massive-com/client-python` source in July 2026 — this supersedes `planning/archive/MASSIVE_API.md`, which predates verification and contains two inaccuracies called out below. + +## Overview + +- **Base URL**: `https://api.massive.com` (legacy `https://api.polygon.io` still works — full API compatibility, identical JSON schemas) +- **Python package**: `massive` on PyPI (`pip install massive` / `uv add massive`), formerly `polygon-api-client` +- **Min Python version**: 3.9 (tested through 3.14) +- **Auth**: API key via `MASSIVE_API_KEY` env var (read automatically by `RESTClient()`) or passed explicitly as `RESTClient(api_key=...)`. The legacy `POLYGON_API_KEY` still works during the transition period. +- **License**: MIT + +Polygon.io rebranded to Massive on 2025-10-30. Existing API keys and accounts continued to work unchanged through the rename; only the package name, import path, and base URL changed. + +## Rate Limits + +| Tier | Limit | +|------|-------| +| Free (Stocks Starter) | 5 requests/minute | +| Paid tiers | Unlimited (recommend staying under 100 req/s) | + +FinAlly polls on a timer. Free tier: poll every 15s. Paid: poll every 2-5s. This matches `MassiveDataSource`'s `poll_interval` default of `15.0` in `backend/app/market/massive_client.py`. + +## Client Initialization + +```python +from massive import RESTClient + +# Reads MASSIVE_API_KEY (or legacy POLYGON_API_KEY) from environment automatically +client = RESTClient() + +# Or pass explicitly +client = RESTClient(api_key="your_key_here") +``` + +## Endpoints Used in FinAlly + +### 1. Snapshot — All Tickers (Primary Endpoint) + +Gets current prices for multiple tickers in a **single API call**. This is the endpoint FinAlly's poller uses. + +**REST**: `GET /v2/snapshot/locale/us/markets/stocks/tickers?tickers=AAPL,GOOGL,MSFT` + +**Python client method**: +```python +def get_snapshot_all( + self, + market_type: Union[str, SnapshotMarketType], + tickers: Optional[Union[str, List[str]]] = None, + include_otc: Optional[bool] = False, + ... +) -> Union[List[TickerSnapshot], HTTPResponse] +``` + +```python +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +client = RESTClient() + +snapshots = client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"], +) + +for snap in snapshots: + print(f"{snap.ticker}: ${snap.last_trade.price}") + print(f" Day change: {snap.todays_change_percent}%") + print(f" Day OHLC: O={snap.day.open} H={snap.day.high} L={snap.day.low} C={snap.day.close}") + print(f" Prev close: {snap.prev_day.close}") +``` + +**`TickerSnapshot` fields** (Python model, `massive/rest/models/snapshot.py`): + +| Field | Type | Notes | +|---|---|---| +| `ticker` | `str` | Symbol | +| `day` | `Agg` | Today's OHLCV bar — `open`, `high`, `low`, `close`, `volume`, `vwap` | +| `prev_day` | `Agg` | **Previous day's** OHLCV bar, same shape as `day` | +| `last_trade` | `LastTrade` | Most recent trade (see below) | +| `last_quote` | `LastQuote` | Most recent NBBO quote | +| `min` | `MinuteSnapshot` | Most recent minute bar | +| `todays_change` | `float` | Absolute change vs. previous close | +| `todays_change_percent` | `float` | Percent change vs. previous close | +| `updated` | `int` | Last-updated timestamp (ns) | +| `fair_market_value` | `float` \| `None` | Business plan tiers only | + +**Correction vs. the archived doc**: there is no `day.previous_close` field. Previous close lives on the sibling `prev_day.close` (raw JSON: `prevDay.c`). Any code (or doc) that reads `snap.day.previous_close` will raise `AttributeError`. + +**`LastTrade` fields** (`massive/rest/models/trades.py`, reused inside snapshots): + +| Field | Type | Notes | +|---|---|---| +| `price` | `float` | Trade price | +| `size` | `float` | Trade size | +| `sip_timestamp` | `int` | **Nanosecond**-accuracy Unix timestamp — when the SIP received the trade | +| `participant_timestamp` | `int` | Nanosecond-accuracy timestamp — when the exchange generated the trade | +| `trf_timestamp` | `int` \| `None` | Trade Reporting Facility timestamp, nanoseconds | +| `exchange`, `conditions`, `id`, `tape`, `correction` | — | Trade metadata | + +**Correction vs. the archived doc, and a real bug in the current codebase**: there is **no `.timestamp` attribute** on `LastTrade` — only `sip_timestamp` / `participant_timestamp` / `trf_timestamp`, all in **nanoseconds**, not milliseconds. See "Known Issue" below. + +### 2. Single Ticker Snapshot + +For detailed data on one ticker (e.g. the detail view when a user clicks a ticker). + +**REST**: `GET /v2/snapshot/locale/us/markets/stocks/tickers/{stocksTicker}` + +```python +snapshot = client.get_snapshot_ticker( + market_type=SnapshotMarketType.STOCKS, + ticker="AAPL", +) + +print(f"Price: ${snapshot.last_trade.price}") +print(f"Bid/Ask: ${snapshot.last_quote.bid_price} / ${snapshot.last_quote.ask_price}") +print(f"Day range: ${snapshot.day.low} - ${snapshot.day.high}") +``` + +### 3. Previous Close + +Previous trading day's OHLC for a ticker. Useful for seed prices. + +**REST**: `GET /v2/aggs/ticker/{ticker}/prev` + +```python +def get_previous_close_agg( + self, + ticker: str, + adjusted: Optional[bool] = None, + ... +) -> Union[PreviousCloseAgg, HTTPResponse] +``` + +```python +prev = client.get_previous_close_agg(ticker="AAPL") + +for agg in prev: + print(f"Previous close: ${agg.close}") + print(f"OHLC: O={agg.open} H={agg.high} L={agg.low} C={agg.close}") + print(f"Volume: {agg.volume}") +``` + +### 4. Aggregates (Bars) + +Historical OHLCV bars over a date range. Not needed for live polling but useful if historical charts are added later. + +**REST**: `GET /v2/aggs/ticker/{ticker}/range/{multiplier}/{timespan}/{from}/{to}` + +```python +def list_aggs( + self, + ticker: str, + multiplier: int, + timespan: str, + from_: Union[str, int, datetime, date], + to: Union[str, int, datetime, date], + adjusted: Optional[bool] = None, + sort: Optional[Union[str, Sort]] = None, + limit: Optional[int] = None, + ... +) -> Union[Iterator[Agg], HTTPResponse] +``` + +```python +aggs = [] +for a in client.list_aggs( + ticker="AAPL", + multiplier=1, + timespan="day", + from_="2026-01-01", + to="2026-01-31", + limit=50000, +): + aggs.append(a) + +for a in aggs: + print(f"O={a.open} H={a.high} L={a.low} C={a.close} V={a.volume}") +``` + +Pagination is enabled by default (`pagination=True`) — `list_aggs` automatically follows `next_url` across pages. + +### 5. Last Trade / Last Quote + +Individual endpoints for the most recent trade or NBBO quote on one ticker. + +```python +trade = client.get_last_trade(ticker="AAPL") +print(f"Last trade: ${trade.price} x {trade.size}") + +quote = client.get_last_quote(ticker="AAPL") +print(f"Bid: ${quote.bid_price} x {quote.bid_size}") +print(f"Ask: ${quote.ask_price} x {quote.ask_size}") +``` + +## How FinAlly Uses the API + +The Massive poller runs as a background asyncio task (`MassiveDataSource` in `backend/app/market/massive_client.py`): + +1. Collects all tickers from the watchlist +2. Calls `get_snapshot_all()` with those tickers (one API call, run via `asyncio.to_thread` since the client is synchronous) +3. Extracts `last_trade.price` from each snapshot and writes it to the shared `PriceCache` +4. Sleeps for the poll interval, then repeats + +```python +import asyncio +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +async def poll_massive(api_key: str, get_tickers, price_cache, interval: float = 15.0): + """Poll Massive API and update the price cache.""" + client = RESTClient(api_key=api_key) + + while True: + tickers = get_tickers() + if tickers: + snapshots = await asyncio.to_thread( + client.get_snapshot_all, + market_type=SnapshotMarketType.STOCKS, + tickers=tickers, + ) + for snap in snapshots: + price_cache.update( + ticker=snap.ticker, + price=snap.last_trade.price, + timestamp=snap.last_trade.sip_timestamp / 1_000_000_000, # ns -> seconds + ) + + await asyncio.sleep(interval) +``` + +## Fixed Issue — timestamp handling (resolved 2026-07-02) + +`backend/app/market/massive_client.py::_poll_once` previously did: + +```python +price = snap.last_trade.price +timestamp = snap.last_trade.timestamp / 1000.0 # ms -> seconds [WRONG] +``` + +`LastTrade` has no `.timestamp` field — real attribute names are `sip_timestamp` / `participant_timestamp`, and they're **nanoseconds**, not milliseconds. Against the real API this raised `AttributeError` inside the per-snapshot `try/except (AttributeError, TypeError)` block, which logged a warning and **skipped the cache update for every ticker, every poll**. With a real `MASSIVE_API_KEY` set, the price cache would never have been populated from Massive data and the SSE stream would sit frozen. + +This wasn't caught by `test_massive.py` because the test's mock manually stamped a `.timestamp` attribute onto a plain `MagicMock()` — an attribute the real `LastTrade` class doesn't have — so the test validated behavior against a shape the live API never returns. + +**Fix applied**: `massive_client.py` now reads `snap.last_trade.sip_timestamp / 1_000_000_000`. `test_massive.py`'s `_make_snapshot` helper now builds `last_trade` as `MagicMock(spec_set=LastTrade)` (the real model class from `massive.rest.models.trades`) rather than a bare `MagicMock()` — `spec_set` rejects both reading *and* setting any attribute not on the real class, so a future regression back to `.timestamp` would fail the test suite immediately instead of silently mocking a field that doesn't exist. All 73 tests in `backend/tests/market/` pass with the fix. + +## Error Handling + +The client raises exceptions for HTTP errors: +- **401**: Invalid API key +- **403**: Insufficient permissions (plan doesn't include the endpoint) +- **429**: Rate limit exceeded (free tier: 5 req/min) +- **5xx**: Server errors (client has built-in retry, 3 retries by default) + +`MassiveDataSource._poll_once` wraps the whole poll in a broad `except Exception` and logs+continues rather than crashing the background task — appropriate for 401/429/network errors, and also what had masked the timestamp bug above (see "Fixed Issue") until the test mock was tightened to `spec_set`. + +## Notes + +- The snapshot endpoint returns data for **all requested tickers in one call** — critical for staying within the free tier's rate limit +- All Massive timestamp fields are Unix **nanoseconds**, not milliseconds — this differs from many other market data APIs and is easy to get wrong (see above) +- During market-closed hours, `last_trade.price` reflects the last traded price (may include after-hours) +- The `day` object resets at market open; during pre-market, its values may still reflect the previous session +- A newer, multi-asset `GET /v3/snapshot` "Unified Snapshot" endpoint exists (stocks/options/forex/crypto in one call, up to 250 tickers via `ticker.any_of`). FinAlly doesn't need it — it's stocks-only and the per-market-type `get_snapshot_all` is simpler and already covers the watchlist use case — but it's worth knowing about if a future feature needs cross-asset data. + +## Sources + +- [API Docs | Massive](https://massive.com/docs) +- [Full Market Snapshot | Stocks REST API - Massive](https://massive.com/docs/rest/stocks/snapshots/full-market-snapshot) +- [Single Ticker Snapshot | Stocks REST API - Massive](https://massive.com/docs/rest/stocks/snapshots/single-ticker-snapshot) +- [Unified Snapshot | Stocks REST API - Massive](https://massive.com/docs/rest/stocks/snapshots/unified-snapshot) +- [Daily Ticker Summary (OHLC) | Stocks REST API - Massive](https://massive.com/docs/rest/stocks/aggregates/daily-ticker-summary) +- [What is the request limit for Massive's RESTful APIs? | Massive](https://massive.com/knowledge-base/article/what-is-the-request-limit-for-massives-restful-apis) +- [Polygon.io is Now Massive | Massive](https://massive.com/blog/polygon-is-now-massive) +- [massive-com/client-python (GitHub)](https://github.com/massive-com/client-python) +- [massive (PyPI)](https://pypi.org/project/massive/) +- [Migration Guide | massive-com/client-python | DeepWiki](https://deepwiki.com/massive-com/client-python/8-migration-guide) diff --git a/planning/PLAN.md b/planning/PLAN.md index bc1811b3..003b85dc 100644 --- a/planning/PLAN.md +++ b/planning/PLAN.md @@ -23,10 +23,10 @@ The user runs a single Docker command (or a provided start script). A browser op - **Watch prices stream** — prices flash green (uptick) or red (downtick) with subtle CSS animations that fade - **View sparkline mini-charts** — price action beside each ticker in the watchlist, accumulated on the frontend from the SSE stream since page load (sparklines fill in progressively) -- **Click a ticker** to see a larger detailed chart in the main chart area -- **Buy and sell shares** — market orders only, instant fill at current price, no fees, no confirmation dialog -- **Monitor their portfolio** — a heatmap (treemap) showing positions sized by weight and colored by P&L, plus a P&L chart tracking total portfolio value over time -- **View a positions table** — ticker, quantity, average cost, current price, unrealized P&L, % change +- **Click a ticker** to see a larger detailed chart in the main chart area — also accumulated from SSE since page load, starting empty and filling in progressively; no historical price API exists +- **Buy and sell shares** — market orders only, instant fill at current price, no fees, no confirmation dialog; quantity must be a whole number in the trade bar (fractional shares are supported via the AI assistant only) +- **Monitor their portfolio** — a heatmap (treemap) showing positions sized by portfolio weight and colored by unrealized P&L % (green = profit, red = loss), plus a P&L chart tracking total portfolio value over time +- **View a positions table** — ticker, quantity, average cost, current price, unrealized P&L ($), unrealized P&L % - **Chat with the AI assistant** — ask about their portfolio, get analysis, and have the AI execute trades and manage the watchlist through natural language - **Manage the watchlist** — add/remove tickers manually or via the AI chat @@ -39,7 +39,7 @@ The user runs a single Docker command (or a provided start script). A browser op - **Responsive but desktop-first**: optimized for wide screens, functional on tablet ### Color Scheme -- Accent Yellow: `#ecad0a` +- Accent Yellow: `#ecad0a` — active/selected states (e.g., selected ticker highlight, active tab underline, key metric labels) - Blue Primary: `#209dd7` - Purple Secondary: `#753991` (submit buttons) @@ -75,7 +75,7 @@ The user runs a single Docker command (or a provided start script). A browser op |---|---| | SSE over WebSockets | One-way push is all we need; simpler, no bidirectional complexity, universal browser support | | Static Next.js export | Single origin, no CORS issues, one port, one container, simple deployment | -| SQLite over Postgres | No auth = no multi-user = no need for a database server; self-contained, zero config | +| SQLite over Postgres | Single-user for now; no need for a database server. Schema includes `user_id` columns so multi-user can be added later without migration. | | Single Docker container | Students run one command; no docker-compose for production, no service orchestration | | uv for Python | Fast, modern Python project management; reproducible lockfile; what students should learn | | Market orders only | Eliminates order book, limit order logic, partial fills — dramatically simpler portfolio math | @@ -88,7 +88,7 @@ The user runs a single Docker command (or a provided start script). A browser op finally/ ├── frontend/ # Next.js TypeScript project (static export) ├── backend/ # FastAPI uv project (Python) -│ └── db/ # Schema definitions, seed data, migration logic +│ └── schema/ # SQL schema definitions and seed data (NOT the runtime DB) ├── planning/ # Project-wide documentation for agents │ ├── PLAN.md # This document │ └── ... # Additional agent reference docs @@ -98,11 +98,12 @@ finally/ │ ├── start_windows.ps1 # Launch Docker container (Windows PowerShell) │ └── stop_windows.ps1 # Stop Docker container (Windows PowerShell) ├── test/ # Playwright E2E tests + docker-compose.test.yml -├── db/ # Volume mount target (SQLite file lives here at runtime) +├── db/ # Volume mount target (runtime SQLite file lives here) │ └── .gitkeep # Directory exists in repo; finally.db is gitignored ├── Dockerfile # Multi-stage build (Node → Python) ├── docker-compose.yml # Optional convenience wrapper -├── .env # Environment variables (gitignored, .env.example committed) +├── .env # Environment variables (gitignored) +├── .env.example # Committed template showing all required variables └── .gitignore ``` @@ -110,7 +111,7 @@ finally/ - **`frontend/`** is a self-contained Next.js project. It knows nothing about Python. It talks to the backend via `/api/*` endpoints and `/api/stream/*` SSE endpoints. Internal structure is up to the Frontend Engineer agent. - **`backend/`** is a self-contained uv project with its own `pyproject.toml`. It owns all server logic including database initialization, schema, seed data, API routes, SSE streaming, market data, and LLM integration. Internal structure is up to the Backend/Market Data agents. -- **`backend/db/`** contains schema SQL definitions and seed logic. The backend lazily initializes the database on first request — creating tables and seeding default data if the SQLite file doesn't exist or is empty. +- **`backend/schema/`** contains SQL schema definitions and seed logic. The backend lazily initializes the database on first request — creating tables and seeding default data if the SQLite file doesn't exist or is empty. This directory holds source files only; the runtime SQLite file is never written here. - **`db/`** at the top level is the runtime volume mount point. The SQLite file (`db/finally.db`) is created here by the backend and persists across container restarts via Docker volume. - **`planning/`** contains project-wide documentation, including this plan. All agents reference files here as the shared contract. - **`test/`** contains Playwright E2E tests and supporting infrastructure (e.g., `docker-compose.test.yml`). Unit tests live within `frontend/` and `backend/` respectively, following each framework's conventions. @@ -137,6 +138,7 @@ LLM_MOCK=false - If `MASSIVE_API_KEY` is set and non-empty → backend uses Massive REST API for market data - If `MASSIVE_API_KEY` is absent or empty → backend uses the built-in market simulator - If `LLM_MOCK=true` → backend returns deterministic mock LLM responses (for E2E tests) +- If `OPENROUTER_API_KEY` is absent or empty → backend starts normally (market data and portfolio work unaffected), but `POST /api/chat` returns HTTP 503 with a clear error message: `"OPENROUTER_API_KEY is not configured"` - The backend reads `.env` from the project root (mounted into the container or read via docker `--env-file`) --- @@ -175,7 +177,7 @@ Both the simulator and the Massive client implement the same abstract interface. - Endpoint: `GET /api/stream/prices` - Long-lived SSE connection; client uses native `EventSource` API -- Server pushes price updates for all tickers known to the system at a regular cadence (~500ms) — in the single-user model this is equivalent to the user's watchlist +- Server always pushes at a fixed ~500ms cadence regardless of the market data source. When using the Massive API (which polls every 15 seconds), the SSE server repeats the last-known price between Massive updates — the price value is unchanged but the event still fires, keeping the connection alive and the frontend state consistent. The price flash animation only triggers when the price value actually changes. - Each SSE event contains ticker, price, previous price, timestamp, and change direction - Client handles reconnection automatically (EventSource has built-in retry) @@ -201,11 +203,10 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - `created_at` TEXT (ISO timestamp) **watchlist** — Tickers the user is watching -- `id` TEXT PRIMARY KEY (UUID) - `user_id` TEXT (default: `"default"`) - `ticker` TEXT - `added_at` TEXT (ISO timestamp) -- UNIQUE constraint on `(user_id, ticker)` +- PRIMARY KEY `(user_id, ticker)` — the composite key is the natural identifier; no UUID needed **positions** — Current holdings (one row per ticker per user) - `id` TEXT PRIMARY KEY (UUID) @@ -244,6 +245,10 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - One user profile: `id="default"`, `cash_balance=10000.0` - Ten watchlist entries: AAPL, GOOGL, MSFT, AMZN, TSLA, NVDA, META, JPM, V, NFLX +### Table Growth + +`portfolio_snapshots` and `chat_messages` grow indefinitely (no TTL or row limit). For a demo this is acceptable — overnight a single session will accumulate at most a few thousand rows. Do **not** add automatic cleanup logic; it would erase the P&L chart history the user sees. + --- ## 8. API Endpoints @@ -270,6 +275,7 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod ### Chat | Method | Path | Description | |--------|------|-------------| +| GET | `/api/chat/history` | Load prior conversation (last 50 messages) for display on page load/refresh | | POST | `/api/chat` | Send a message, receive complete JSON response (message + executed actions) | ### System @@ -290,7 +296,7 @@ There is an OPENROUTER_API_KEY in the .env file in the project root. When the user sends a chat message, the backend: 1. Loads the user's current portfolio context (cash, positions with P&L, watchlist with live prices, total portfolio value) -2. Loads recent conversation history from the `chat_messages` table +2. Loads the most recent 20 messages from the `chat_messages` table (10 user + 10 assistant pairs; sufficient context without risking overflow) 3. Constructs a prompt with a system message, portfolio context, conversation history, and the user's new message 4. Calls the LLM via LiteLLM → OpenRouter, requesting structured output, using the cerebras-inference skill 5. Parses the complete structured JSON response @@ -309,14 +315,15 @@ The LLM is instructed to respond with JSON matching this schema: {"ticker": "AAPL", "side": "buy", "quantity": 10} ], "watchlist_changes": [ - {"ticker": "PYPL", "action": "add"} + {"ticker": "PYPL", "action": "add"}, + {"ticker": "NFLX", "action": "remove"} ] } ``` - `message` (required): The conversational text shown to the user -- `trades` (optional): Array of trades to auto-execute. Each trade goes through the same validation as manual trades (sufficient cash for buys, sufficient shares for sells) -- `watchlist_changes` (optional): Array of watchlist modifications +- `trades` (optional): Array of trades to auto-execute. Each trade goes through the same validation as manual trades (sufficient cash for buys, sufficient shares for sells). `quantity` must be a positive number; fractional shares are allowed. +- `watchlist_changes` (optional): Array of watchlist modifications. `action` is either `"add"` or `"remove"`. ### Auto-Execution @@ -344,6 +351,18 @@ When `LLM_MOCK=true`, the backend returns deterministic mock responses instead o - Development without an API key - CI/CD pipelines +The canonical mock response (used by all agents and E2E tests) is: + +```json +{ + "message": "I've analyzed your portfolio. You have $10,000 in cash and no open positions. I recommend starting with a diversified position — shall I buy 5 shares of AAPL?", + "trades": [], + "watchlist_changes": [] +} +``` + +This response is intentionally action-free so E2E tests control trade execution separately (by sending explicit buy/sell instructions and verifying the next mock response includes a trade). + --- ## 10. Frontend Design @@ -352,19 +371,19 @@ When `LLM_MOCK=true`, the backend returns deterministic mock responses instead o The frontend is a single-page application with a dense, terminal-inspired layout. The specific component architecture and layout system is up to the Frontend Engineer, but the UI should include these elements: -- **Watchlist panel** — grid/table of watched tickers with: ticker symbol, current price (flashing green/red on change), daily change %, and a sparkline mini-chart (accumulated from SSE since page load) -- **Main chart area** — larger chart for the currently selected ticker, with at minimum price over time. Clicking a ticker in the watchlist selects it here. -- **Portfolio heatmap** — treemap visualization where each rectangle is a position, sized by portfolio weight, colored by P&L (green = profit, red = loss) +- **Watchlist panel** — grid/table of watched tickers with: ticker symbol, current price (flashing green/red on change), % chg since session open (first SSE price received for that ticker since page load), and a sparkline mini-chart (accumulated from SSE since page load) +- **Main chart area** — larger chart for the currently selected ticker, showing price over time accumulated from SSE since page load (starts empty, fills progressively). Clicking a ticker in the watchlist selects it here. +- **Portfolio heatmap** — treemap visualization where each rectangle is a position, sized by portfolio weight, colored by unrealized P&L % (green = profit, red = loss) - **P&L chart** — line chart showing total portfolio value over time, using data from `portfolio_snapshots` - **Positions table** — tabular view of all positions: ticker, quantity, avg cost, current price, unrealized P&L, % change -- **Trade bar** — simple input area: ticker field, quantity field, buy button, sell button. Market orders, instant fill. +- **Trade bar** — simple input area: ticker field, whole-number quantity field, buy button, sell button. Market orders, instant fill. Fractional shares are only available via the AI chat. - **AI chat panel** — docked/collapsible sidebar. Message input, scrolling conversation history, loading indicator while waiting for LLM response. Trade executions and watchlist changes shown inline as confirmations. - **Header** — portfolio total value (updating live), connection status indicator, cash balance ### Technical Notes - Use `EventSource` for SSE connection to `/api/stream/prices` -- Canvas-based charting library preferred (Lightweight Charts or Recharts) for performance +- TradingView Lightweight Charts preferred for performance (canvas-based); Recharts is an acceptable fallback (SVG-based, easier API) if Lightweight Charts integration proves problematic - Price flash effect: on receiving a new price, briefly apply a CSS class with background color transition, then remove it - All API calls go to the same origin (`/api/*`) — no CORS configuration needed - Tailwind CSS for styling with a custom dark theme @@ -454,3 +473,4 @@ The container is designed to deploy to AWS App Runner, Render, or any container - Portfolio visualization: heatmap renders with correct colors, P&L chart has data points - AI chat (mocked): send a message, receive a response, trade execution appears inline - SSE resilience: disconnect and verify reconnection +