From 4692f2f76a322a24288f8e730251cf54eeaacb9c Mon Sep 17 00:00:00 2001 From: Hendro Date: Fri, 31 Jul 2026 19:23:19 +0700 Subject: [PATCH 1/2] initial finally --- .claude/agents/change-reviewer.md | 6 + .claude/agents/codex-reviewer.md | 6 + .claude/agents/reviewer.md | 6 + .claude/commands/doc-review.md | 1 + .claude/settings.json | 12 ++ README.md | 48 +++--- planning/MARKET_INTERFACE.md | 272 ++++++++++++++++++++++++++++++ planning/MARKET_SIMULATOR.md | 230 +++++++++++++++++++++++++ planning/MASSIVE_API.md | 238 ++++++++++++++++++++++++++ planning/review.md | 243 ++++++++++++++++++++++++++ 10 files changed, 1042 insertions(+), 20 deletions(-) create mode 100644 .claude/agents/change-reviewer.md create mode 100644 .claude/agents/codex-reviewer.md create mode 100644 .claude/agents/reviewer.md create mode 100644 .claude/commands/doc-review.md create mode 100644 planning/MARKET_INTERFACE.md create mode 100644 planning/MARKET_SIMULATOR.md create mode 100644 planning/MASSIVE_API.md create mode 100644 planning/review.md diff --git a/.claude/agents/change-reviewer.md b/.claude/agents/change-reviewer.md new file mode 100644 index 000000000..687118e83 --- /dev/null +++ b/.claude/agents/change-reviewer.md @@ -0,0 +1,6 @@ +--- +name: change-reviewer +description: carry out a comprehensive review of all changes since the last commit +--- + +You review the file planning/Plan.md and write your feedback to planning/review.md. \ No newline at end of file diff --git a/.claude/agents/codex-reviewer.md b/.claude/agents/codex-reviewer.md new file mode 100644 index 000000000..772082329 --- /dev/null +++ b/.claude/agents/codex-reviewer.md @@ -0,0 +1,6 @@ +--- +name: codex-reviewer +description: carry out a comprehensive review of Plan.md when requested using codex +--- +You are using a different ai agent to carry out a review of the document planning/Plan.md. you must execute the following shell commend to carry out the review - do not review yourself: 'codex exec "please review the file planning/plan.md and write your feedback to planning/review.md"' +This will run the review process and save the result. Do not review yourself. \ No newline at end of file diff --git a/.claude/agents/reviewer.md b/.claude/agents/reviewer.md new file mode 100644 index 000000000..fc5cfaf87 --- /dev/null +++ b/.claude/agents/reviewer.md @@ -0,0 +1,6 @@ +--- +name: Reviewer +description: This custom agent reviews code and provides feedback on improvements, best practices, and potential issues. +--- + +You review the file planning/Plan.md and write your feedback to planning/review.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 000000000..3938ccf2a --- /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 toa new section at the end, along with any opportunities to simplify \ No newline at end of file diff --git a/.claude/settings.json b/.claude/settings.json index aa06f43dc..861d0a8fb 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -3,5 +3,17 @@ "frontend-design@claude-plugins-official": true, "context7@claude-plugins-official": true, "playwright@claude-plugins-official": true + }, + "hooks": { + "Stop": [ + { + "hooks": [ + { + "type": "command", + "command": "if [ -z \"$FINALLY_STOP_HOOK_ACTIVE\" ]; then FINALLY_STOP_HOOK_ACTIVE=1 claude -p 'Use the change-reviewer agent to review all changes since the last commit and write the result to planning/review.md'; fi" + } + ] + } + ] } } diff --git a/README.md b/README.md index 3f2582ae2..f6b493c17 100644 --- a/README.md +++ b/README.md @@ -2,9 +2,22 @@ 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 spec. -## Features +## Status + +๐Ÿšง **Under active development.** Only the market data backend is built so far. The API routes, portfolio/trading logic, AI chat, frontend, and Docker packaging described below are planned but not yet implemented. + +**Done:** +- โœ… Market data subsystem (`backend/app/market/`) โ€” GBM simulator with correlated moves, Massive (Polygon.io) client, thread-safe price cache, SSE stream endpoint factory. 73 tests passing. See [planning/MARKET_DATA_SUMMARY.md](planning/MARKET_DATA_SUMMARY.md). + +**Not yet started:** +- Database schema, portfolio/trade endpoints, watchlist endpoints +- AI chat integration (LiteLLM โ†’ OpenRouter via Cerebras) +- Frontend (Next.js terminal UI) +- Dockerfile, start/stop scripts, E2E tests + +## Planned Features - **Live price streaming** via SSE with green/red flash animations - **Simulated portfolio** โ€” $10k virtual cash, market orders, instant fills @@ -13,7 +26,7 @@ Built entirely by coding agents as a capstone project for an agentic AI coding c - **Watchlist management** โ€” track tickers manually or via AI - **Dark terminal aesthetic** โ€” Bloomberg-inspired, data-dense layout -## Architecture +## Planned Architecture Single Docker container serving everything on port 8000: @@ -21,27 +34,25 @@ Single Docker container serving everything on port 8000: - **Backend**: FastAPI (Python/uv) with SSE streaming - **Database**: SQLite with lazy initialization - **AI**: LiteLLM โ†’ OpenRouter (Cerebras inference) with structured outputs -- **Market data**: Built-in GBM simulator (default) or Massive API (optional) +- **Market data**: Built-in GBM simulator (default) or Massive API (optional) โ€” โœ… implemented -## Quick Start +## Try the Market Data Demo -```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 +The only runnable piece right now is a terminal demo of the market data simulator: -# Open http://localhost:8000 +```bash +cd backend +uv sync +uv run market_data_demo.py ``` +Displays a live-updating dashboard with all 10 tickers, sparklines, and an event log. Runs 60 seconds or until Ctrl+C. + ## Environment Variables | Variable | Required | Description | |---|---|---| -| `OPENROUTER_API_KEY` | Yes | OpenRouter API key for AI chat | +| `OPENROUTER_API_KEY` | Yes (once chat is built) | OpenRouter API key for AI chat | | `MASSIVE_API_KEY` | No | Massive (Polygon.io) key for real market data; omit to use simulator | | `LLM_MOCK` | No | Set `true` for deterministic mock LLM responses (testing) | @@ -49,12 +60,9 @@ docker run -v finally-data:/app/db -p 8000:8000 --env-file .env finally ``` finally/ -โ”œโ”€โ”€ frontend/ # Next.js static export -โ”œโ”€โ”€ backend/ # FastAPI uv project +โ”œโ”€โ”€ backend/ # FastAPI uv project (market data subsystem built; API/DB/chat pending) โ”œโ”€โ”€ planning/ # Project documentation and agent contracts -โ”œโ”€โ”€ test/ # Playwright E2E tests -โ”œโ”€โ”€ db/ # SQLite volume mount (runtime) -โ””โ”€โ”€ scripts/ # Start/stop helpers +โ””โ”€โ”€ (planned) frontend/, test/, db/, scripts/ ``` ## License diff --git a/planning/MARKET_INTERFACE.md b/planning/MARKET_INTERFACE.md new file mode 100644 index 000000000..4c7507225 --- /dev/null +++ b/planning/MARKET_INTERFACE.md @@ -0,0 +1,272 @@ +# Unified Market Data Interface + +Design for `backend/app/market/` โ€” the abstraction that lets the rest of FinAlly (SSE stream, +portfolio valuation, trade execution) read live prices without knowing whether they came from the +Massive API or the built-in simulator. Backed by the research in `MASSIVE_API.md`; simulator +internals are in `MARKET_SIMULATOR.md`. + +## 1. Goals + +- One interface, two implementations, selected purely by whether `MASSIVE_API_KEY` is set + (`PLAN.md` ยง5) +- Downstream code (SSE stream, portfolio math) never branches on data source +- Survives Massive rate limits / outages without ever serving "no price" to the frontend +- Cheap to extend with a third source later (e.g. a different vendor) without touching callers + +## 2. Shape of the Data: `PriceUpdate` + +A single immutable record represents "the latest known state of one ticker": + +```python +# models.py +from dataclasses import dataclass +from datetime import datetime +from enum import Enum + +class Direction(str, Enum): + UP = "up" + DOWN = "down" + FLAT = "flat" + +@dataclass(frozen=True) +class PriceUpdate: + ticker: str + price: float + previous_price: float + timestamp: datetime + direction: Direction + + @property + def change(self) -> float: + return self.price - self.previous_price + + @property + def change_percent(self) -> float: + if self.previous_price == 0: + return 0.0 + return (self.change / self.previous_price) * 100 +``` + +Both the simulator and the Massive client produce this same type โ€” it's the only thing that +crosses the boundary out of `app/market/`. + +## 3. The Abstract Interface + +```python +# interface.py +from abc import ABC, abstractmethod + +class MarketDataSource(ABC): + """A background process that keeps a PriceCache updated for a set of tickers.""" + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing updates for the given tickers (writes into the shared cache).""" + + @abstractmethod + async def stop(self) -> None: + """Stop background work cleanly (cancel tasks, close HTTP/WS clients).""" + + @abstractmethod + async def add_ticker(self, ticker: str) -> None: + """Start tracking a new ticker without restarting the whole source.""" + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Stop tracking a ticker (e.g. removed from the watchlist).""" + + @abstractmethod + def get_tickers(self) -> list[str]: + """Currently tracked tickers.""" +``` + +Both `SimulatorDataSource` and `MassiveDataSource` implement this. Neither exposes anything else +publicly โ€” no HTTP client, no simulation state โ€” so callers can't accidentally couple to one +implementation's internals. + +## 4. The Shared Price Cache + +A single in-memory, thread/async-safe store sits between the data source and every consumer +(SSE endpoint, portfolio valuation, trade execution). Producers write, everyone else reads: + +```python +# cache.py +import asyncio + +class PriceCache: + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._version = 0 + self._lock = asyncio.Lock() + + async def set(self, update: PriceUpdate) -> None: + async with self._lock: + self._prices[update.ticker] = update + self._version += 1 + + def get(self, ticker: str) -> PriceUpdate | None: + return self._prices.get(ticker) + + def get_price(self, ticker: str) -> float | None: + u = self.get(ticker) + return u.price if u else None + + def get_all(self) -> dict[str, PriceUpdate]: + return dict(self._prices) + + @property + def version(self) -> int: + return self._version +``` + +The `version` counter lets the SSE endpoint cheaply detect "has anything changed since I last +looked" without diffing the whole dict on every tick โ€” see ยง7. + +This design point matters regardless of which source is active: **the cache always holds the last +known value for every ticker.** Nothing ever gets deleted except by an explicit `remove_ticker`. +That's what makes Massive rate limits and transient outages invisible to the frontend โ€” see ยง6. + +## 5. Selecting an Implementation: the Factory + +```python +# factory.py +import os + +def create_market_data_source(cache: PriceCache) -> MarketDataSource: + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + if api_key: + return MassiveDataSource(cache, api_key=api_key) + return SimulatorDataSource(cache) +``` + +This is the only place that reads the environment variable. Everything else โ€” app startup, the +SSE router, tests โ€” takes a `MarketDataSource` and doesn't care which concrete class it got. + +Startup wiring (e.g. FastAPI lifespan): + +```python +cache = PriceCache() +source = create_market_data_source(cache) +watchlist = load_watchlist_tickers() # from SQLite +await source.start(watchlist) +``` + +## 6. `MassiveDataSource`: REST Polling Implementation + +Per `MASSIVE_API.md` ยง3, the free Massive tier allows 5 requests/minute, so this implementation: + +- Polls the **batched multi-ticker snapshot** endpoint (`GET /v2/snapshot/.../tickers?tickers=...`) + โ€” one HTTP call covers the entire watchlist regardless of size +- Uses a poll interval read from an env var (default 15s, matching the free-tier budget of 5/min + with headroom); paid-tier users can lower it via `MASSIVE_POLL_INTERVAL_SECONDS` +- Runs as a single `asyncio` background task with a `while running: await asyncio.sleep(interval)` + loop โ€” no separate task per ticker +- On each poll: parses the batch response, and for each ticker builds a `PriceUpdate` using the + *previous* cached price (or `prevDay.c` on the very first poll) as `previous_price`, so direction + and change are always computed tick-over-tick rather than against a stale baseline +- On a per-ticker `NOT_FOUND`/error entry in the batch response: skip that ticker this round, + leave its last cached value untouched, log a warning +- On a request failure (timeout, `429`, 5xx): catch it, log, skip this poll cycle entirely, leave + the whole cache untouched, retry on the next scheduled poll โ€” **never propagate the failure to + callers or blank out prices** +- `add_ticker`/`remove_ticker` just mutate the tracked ticker set consulted on the *next* poll + (no need to restart the task or make an extra request) + +```python +# massive_client.py (sketch) +class MassiveDataSource(MarketDataSource): + def __init__(self, cache: PriceCache, api_key: str, + poll_interval: float = 15.0) -> None: + self._cache = cache + self._client = RESTClient(api_key=api_key) + self._interval = poll_interval + self._tickers: set[str] = set() + self._task: asyncio.Task | None = None + self._running = False + + async def start(self, tickers: list[str]) -> None: + self._tickers = set(tickers) + self._running = True + self._task = asyncio.create_task(self._poll_loop()) + + async def _poll_loop(self) -> None: + while self._running: + try: + await self._poll_once() + except Exception: + log.warning("Massive poll failed; keeping last known prices", exc_info=True) + await asyncio.sleep(self._interval) + + async def _poll_once(self) -> None: + if not self._tickers: + return + snapshot = await asyncio.to_thread( + self._client.get_snapshot_all, "stocks", tickers=list(self._tickers) + ) + for entry in snapshot: + if getattr(entry, "error", None): + continue + prev = self._cache.get_price(entry.ticker) + baseline = prev if prev is not None else entry.prev_day.close + price = entry.day.close or entry.last_trade.price + await self._cache.set(PriceUpdate( + ticker=entry.ticker, + price=price, + previous_price=baseline, + timestamp=datetime.now(timezone.utc), + direction=_direction(price, baseline), + )) +``` + +The blocking `massive` client call is offloaded via `asyncio.to_thread` since it's a synchronous +`requests`-based SDK, keeping the event loop free. + +## 7. Consumers of the Cache + +### SSE stream (`GET /api/stream/prices`) + +```python +async def _generate_events(cache: PriceCache) -> AsyncGenerator[str, None]: + last_seen_version = -1 + while True: + if cache.version != last_seen_version: + last_seen_version = cache.version + for update in cache.get_all().values(): + yield f"data: {json.dumps(asdict(update), default=str)}\n\n" + await asyncio.sleep(0.5) +``` + +Polling the cache's `version` at ~500ms gives the frontend the smooth, frequent cadence described +in `PLAN.md` ยง6/ยง10, decoupled from however slowly the *upstream* Massive poll actually refreshes +data โ€” between real updates the cache simply reports the same values again, which is harmless +(the frontend's flash animation only fires on an actual price change). + +### Portfolio valuation / trade execution + +Both call `cache.get_price(ticker)` synchronously โ€” a plain dict lookup, no I/O, so trade execution +is never blocked on network calls to Massive. + +## 8. Testing Strategy + +- `MarketDataSource` is an ABC โ€” a lightweight `FakeDataSource` (or just `SimulatorDataSource` + with a fixed seed) can stand in for `MassiveDataSource` in tests that need a data source but + aren't testing Massive-specific parsing +- `MassiveDataSource` tests mock the `massive.RESTClient` methods (or the underlying HTTP call) + and assert: successful batch parsing, per-ticker `NOT_FOUND` handling, and that a raised + exception during polling leaves the cache untouched rather than clearing it +- `factory.py` tests assert env-var presence/absence selects the right class, using + `monkeypatch.setenv`/`delenv` + +## 9. File Layout + +``` +backend/app/market/ +โ”œโ”€โ”€ __init__.py # re-exports PriceCache, MarketDataSource, create_market_data_source +โ”œโ”€โ”€ models.py # PriceUpdate, Direction +โ”œโ”€โ”€ interface.py # MarketDataSource ABC +โ”œโ”€โ”€ cache.py # PriceCache +โ”œโ”€โ”€ factory.py # create_market_data_source() +โ”œโ”€โ”€ simulator.py # SimulatorDataSource โ€” see MARKET_SIMULATOR.md +โ”œโ”€โ”€ massive_client.py # MassiveDataSource โ€” this document, ยง6 +โ””โ”€โ”€ stream.py # SSE router factory consuming a PriceCache โ€” ยง7 +``` diff --git a/planning/MARKET_SIMULATOR.md b/planning/MARKET_SIMULATOR.md new file mode 100644 index 000000000..427bc1781 --- /dev/null +++ b/planning/MARKET_SIMULATOR.md @@ -0,0 +1,230 @@ +# Market Simulator Design + +Design for `SimulatorDataSource` in `backend/app/market/simulator.py` โ€” the default price feed +used whenever `MASSIVE_API_KEY` is not set (`PLAN.md` ยง6). It implements the same +`MarketDataSource` interface described in `MARKET_INTERFACE.md`, so nothing downstream needs to +know a simulator is running instead of live Massive data. + +## 1. Goals + +- Realistic-*looking* price action: continuous small moves, occasional visible jumps, correlated + sector behavior โ€” not just uncorrelated random noise +- Zero external dependencies โ€” runs entirely in-process, no network calls, works offline +- Deterministic enough to test (seedable RNG), lively enough to demo well +- Updates at ~500ms per `PLAN.md` ยง6, matching the cadence the SSE stream pushes to the frontend + +## 2. Model: Geometric Brownian Motion (GBM) + +Each ticker's price follows discrete-time GBM, the standard model for simulating a stock price +path: + +``` +S(t + dt) = S(t) * exp((ฮผ - ฯƒยฒ/2) * dt + ฯƒ * sqrt(dt) * Z) +``` + +where: +- `S(t)` โ€” current price +- `ฮผ` (mu) โ€” annualized drift (expected return) +- `ฯƒ` (sigma) โ€” annualized volatility +- `dt` โ€” time step, expressed in *years* (so a 500ms tick is a very small `dt`) +- `Z` โ€” a standard normal random draw (`N(0, 1)`) + +GBM is used because it's the textbook model for equity prices: returns are log-normally +distributed, prices never go negative, and `ฮผ`/`ฯƒ` map directly onto real-world "this stock trends +up slowly but is choppy" (high ฯƒ) vs. "steady mover" (low ฯƒ) intuitions โ€” easy to tune per ticker. + +### Converting the update cadence into `dt` + +With ticks every 500ms and 252 trading days/year ร— 6.5 trading hours/day of "market time" as the +reference frame (a simplification โ€” the sim runs continuously, not just market hours, since it's a +demo): + +```python +TICKS_PER_YEAR = 252 * 6.5 * 3600 / 0.5 # โ‰ˆ 11,793,600 ticks/year +dt = 1 / TICKS_PER_YEAR +``` + +This keeps per-tick moves small (fractions of a percent) so 500ms updates look like continuous +streaming price action rather than jumpy random walks, while still compounding into plausible +daily/weekly ranges over a longer demo session. + +## 3. Per-Ticker Parameters and Seed Prices + +Each ticker gets a starting price and its own `(ฮผ, ฯƒ)` pair, tuned to feel roughly true to the real +stock's character (not intended to be predictive โ€” just recognizable): + +```python +# seed_prices.py +SEED_PRICES: dict[str, float] = { + "AAPL": 190.0, "GOOGL": 175.0, "MSFT": 420.0, "AMZN": 185.0, "TSLA": 250.0, + "NVDA": 120.0, "META": 500.0, "JPM": 200.0, "V": 275.0, "NFLX": 650.0, +} + +GBM_PARAMS: dict[str, tuple[float, float]] = { + # ticker: (annual drift, annual volatility) + "AAPL": (0.10, 0.25), + "GOOGL": (0.10, 0.28), + "MSFT": (0.12, 0.22), + "AMZN": (0.12, 0.30), + "TSLA": (0.05, 0.55), # high volatility, near-zero net drift โ€” famously choppy + "NVDA": (0.20, 0.45), # high growth, high volatility + "META": (0.10, 0.32), + "JPM": (0.08, 0.20), # financials: steadier + "V": (0.09, 0.18), + "NFLX": (0.11, 0.35), +} +``` + +## 4. Correlated Moves Across Tickers + +Real markets don't move ticker-by-ticker independently โ€” sectors move together (tech stocks rally +or sell off as a group; financials react to rate news together). Independent random draws per +ticker look obviously fake once you watch the demo for more than a few seconds. + +**Approach: Cholesky decomposition of a sector correlation matrix.** + +1. Assign each ticker a sector group: + ```python + SECTOR: dict[str, str] = { + "AAPL": "tech", "GOOGL": "tech", "MSFT": "tech", "NVDA": "tech", "META": "tech", + "AMZN": "consumer", "TSLA": "consumer", "NFLX": "consumer", + "JPM": "finance", "V": "finance", + } + ``` +2. Build a target correlation matrix `ฮฃ` from a few constants: + ```python + SAME_SECTOR_CORR = 0.6 # e.g. AAPL vs MSFT + CROSS_GROUP_CORR = 0.3 # e.g. AAPL vs JPM + FINANCE_CORR = 0.5 # JPM vs V, slightly tighter than the general cross-sector figure + ``` + with 1.0 on the diagonal. +3. Compute the Cholesky factor `L` such that `L @ L.T == ฮฃ` once at startup (it's fixed for the + life of the process, since sector assignments don't change). +4. On every tick, draw one vector of independent standard normals `Z_indep` (length = number of + tickers), then correlate them: `Z_correlated = L @ Z_indep`. Feed each ticker's entry from + `Z_correlated` into its own GBM step as `Z` in the formula in ยง2. + +```python +# simulator.py (sketch) +import numpy as np + +class GBMSimulator: + def __init__(self, tickers: list[str], seed: int | None = None) -> None: + self._tickers = tickers + self._rng = np.random.default_rng(seed) + self._prices = {t: SEED_PRICES[t] for t in tickers} + self._corr = _build_correlation_matrix(tickers) # ฮฃ, from SECTOR groups + self._chol = np.linalg.cholesky(self._corr) + + def step(self, dt: float) -> dict[str, float]: + z_indep = self._rng.standard_normal(len(self._tickers)) + z_corr = self._chol @ z_indep + new_prices = {} + for i, ticker in enumerate(self._tickers): + mu, sigma = GBM_PARAMS[ticker] + s = self._prices[ticker] + z = z_corr[i] + new_prices[ticker] = s * math.exp((mu - sigma**2 / 2) * dt + sigma * math.sqrt(dt) * z) + new_prices = self._maybe_apply_shocks(new_prices) + self._prices = new_prices + return new_prices + + def get_tickers(self) -> list[str]: + return list(self._tickers) +``` + +Adding a ticker at runtime (via watchlist) appends a row/column to the correlation matrix (default +it to `CROSS_GROUP_CORR` against everything unless it matches a known `SECTOR` entry) and +recomputes the Cholesky factor โ€” cheap at this scale (โ‰ค a few dozen tickers). + +## 5. Random Shock Events + +Continuous GBM alone looks smooth and a little boring for a demo. Per `PLAN.md` ยง6, the simulator +adds occasional sudden moves: + +- Each tick, each ticker independently has a small probability (~0.1%) of a "shock" +- A shock is a one-off 2โ€“5% move (uniformly sampled magnitude, random sign) applied on top of the + normal GBM step for that tick only โ€” it does not alter `ฮผ`/`ฯƒ` going forward +- At ~2 ticks/second this yields roughly one shock every few minutes across a 10-ticker watchlist, + often enough to be noticeable in a live demo without dominating the price action + +```python +SHOCK_PROBABILITY = 0.001 +SHOCK_MAGNITUDE_RANGE = (0.02, 0.05) + +def _maybe_apply_shocks(self, prices: dict[str, float]) -> dict[str, float]: + for ticker in prices: + if self._rng.random() < SHOCK_PROBABILITY: + magnitude = self._rng.uniform(*SHOCK_MAGNITUDE_RANGE) + sign = self._rng.choice([-1, 1]) + prices[ticker] *= 1 + sign * magnitude + return prices +``` + +## 6. `SimulatorDataSource`: Wiring into the `MarketDataSource` Interface + +```python +# simulator.py (sketch, continued) +class SimulatorDataSource(MarketDataSource): + def __init__(self, cache: PriceCache, tick_interval: float = 0.5, + seed: int | None = None) -> None: + self._cache = cache + self._interval = tick_interval + self._seed = seed + self._sim: GBMSimulator | None = None + self._task: asyncio.Task | None = None + self._running = False + + async def start(self, tickers: list[str]) -> None: + self._sim = GBMSimulator(tickers, seed=self._seed) + self._running = True + self._task = asyncio.create_task(self._tick_loop()) + + async def _tick_loop(self) -> None: + dt = 1 / TICKS_PER_YEAR + while self._running: + new_prices = self._sim.step(dt) + for ticker, price in new_prices.items(): + prev = self._cache.get_price(ticker) + baseline = prev if prev is not None else price + await self._cache.set(PriceUpdate( + ticker=ticker, price=price, previous_price=baseline, + timestamp=datetime.now(timezone.utc), + direction=_direction(price, baseline), + )) + await asyncio.sleep(self._interval) + + async def stop(self) -> None: + self._running = False + if self._task: + self._task.cancel() + + async def add_ticker(self, ticker: str) -> None: + self._sim.add_ticker(ticker) # seeds price, extends correlation matrix + + async def remove_ticker(self, ticker: str) -> None: + self._sim.remove_ticker(ticker) +``` + +This is a direct structural mirror of `MassiveDataSource` from `MARKET_INTERFACE.md` ยง6 โ€” same +task-loop shape, same "write into a shared `PriceCache`" contract โ€” which is what makes the two +implementations swappable via the factory without touching any consumer. + +## 7. Testing Strategy + +- **GBM math**: with a fixed seed, assert prices stay positive over a long run, and that + aggregate drift/volatility over many ticks roughly matches the configured `ฮผ`/`ฯƒ` (statistical + assertions with generous tolerance, not exact-value checks) +- **Correlation**: with a fixed seed and many ticks, compute the empirical correlation between two + same-sector tickers' log returns and assert it's closer to `SAME_SECTOR_CORR` than to + `CROSS_GROUP_CORR` +- **Shocks**: force `SHOCK_PROBABILITY = 1.0` in a test to assert a shock is applied and its + magnitude falls within `SHOCK_MAGNITUDE_RANGE` +- **Interface conformance**: the same test suite structure used for `MassiveDataSource` (start/ + stop/add_ticker/remove_ticker/get_tickers behave per the ABC contract) runs against + `SimulatorDataSource` too, since both implement `MarketDataSource` + +## 8. Dependencies + +Only `numpy` is needed beyond the standard library (for the RNG and Cholesky decomposition) โ€” +no external services, no API keys, works fully offline. diff --git a/planning/MASSIVE_API.md b/planning/MASSIVE_API.md new file mode 100644 index 000000000..a4c0a85e5 --- /dev/null +++ b/planning/MASSIVE_API.md @@ -0,0 +1,238 @@ +# Massive API Research (formerly Polygon.io) + +Research notes on the [Massive](https://massive.com) market data API โ€” the optional real-data +source for FinAlly, used when `MASSIVE_API_KEY` is set (see `PLAN.md` ยง5, ยง6). + +## 1. Background + +Polygon.io rebranded to **Massive** in late 2025. Existing Polygon.io API keys, endpoints, and +the `polygon-api-client` Python package continue to work unchanged โ€” Massive is the same company, +same data, same infrastructure, new name. Key facts: + +- Base URL: `https://api.massive.com` (the old `https://api.polygon.io` still works โ€” same backend) +- Official Python client package is now **`massive`** (was `polygon-api-client`) +- Coverage: US equities, options, indices, forex, crypto, and futures; equities data goes back to 2003 +- Products relevant to FinAlly: **Stocks REST API** (snapshots, aggregates) โ€” we do not need + options, forex, or the WebSocket product for this project (see ยง6 for why) + +Docs root: `https://massive.com/docs` โ€” the stocks section has a machine-readable dump at +`https://massive.com/docs/rest/stocks/llms-full.txt` which is the fastest way to get exact +endpoint/field names. + +## 2. Authentication + +Every REST request needs an API key, supplied one of two ways: + +**Query parameter** (raw HTTP): +``` +GET https://api.massive.com/v2/snapshot/locale/us/markets/stocks/tickers/AAPL?apiKey=YOUR_API_KEY +``` + +**Authorization header** (what the official client sends under the hood): +``` +Authorization: Bearer YOUR_API_KEY +``` + +The Python client takes the key as a constructor argument โ€” it does **not** read an environment +variable itself, so FinAlly's backend must read `MASSIVE_API_KEY` from `.env` and pass it in +explicitly: + +```python +import os +from massive import RESTClient + +client = RESTClient(api_key=os.environ["MASSIVE_API_KEY"]) +``` + +## 3. Rate Limits + +This is the single most important constraint for the project's design: + +| Tier | Limit | +|---|---| +| Free | **5 requests/minute** | +| Paid (Stocks Starter and up) | Much higher / effectively unlimited, but Massive asks clients to stay under ~100 req/sec | + +FinAlly targets students running with the **free tier**, so the Massive-backed data source must +poll infrequently and batch every ticker into a single request rather than one request per ticker. +This directly shapes the interface in `MARKET_INTERFACE.md` โ€” see ยง6 below for the polling cadence +this implies. + +## 4. Installing the Python Client + +```bash +pip install -U massive +# or, in this project: uv add massive +``` + +```python +from massive import RESTClient # REST polling client +from massive import WebSocketClient # real-time streaming client (not used by FinAlly, see ยง6) +``` + +## 5. Endpoints FinAlly Needs + +### 5.1 Real-time / latest price โ€” multi-ticker snapshot (batch) + +The **Full Market Snapshot** endpoint takes a comma-separated ticker list and returns the latest +trade, quote, and day/prev-day bar for each in a single call. This is the endpoint FinAlly's +poller uses โ€” one call covers the whole watchlist regardless of size (up to 250 tickers). + +``` +GET /v2/snapshot/locale/us/markets/stocks/tickers?tickers=AAPL,GOOGL,MSFT&apiKey=YOUR_API_KEY +``` + +Raw HTTP example: + +```python +import requests + +resp = requests.get( + "https://api.massive.com/v2/snapshot/locale/us/markets/stocks/tickers", + params={"tickers": "AAPL,GOOGL,MSFT", "apiKey": API_KEY}, + timeout=10, +) +resp.raise_for_status() +data = resp.json() +``` + +Example response shape (one entry per ticker, plus per-ticker errors for bad symbols): + +```json +{ + "status": "OK", + "tickers": [ + { + "ticker": "AAPL", + "todaysChange": 0.98, + "todaysChangePerc": 0.82, + "updated": 1605195918306274000, + "day": { "o": 119.62, "h": 120.53, "l": 118.81, "c": 120.4229, "v": 28727868, "vw": 119.725 }, + "prevDay": { "o": 117.19, "h": 119.63, "l": 116.44, "c": 119.49, "v": 110597265, "vw": 118.4998 }, + "lastTrade": { "p": 120.47, "s": 236, "t": 1605195918306274000 }, + "lastQuote": { "p": 120.46, "P": 120.47, "s": 8, "S": 4, "t": 1605195918507251700 }, + "min": { "o": 120.435, "h": 120.468, "l": 120.37, "c": 120.4201, "v": 270796, "t": 1684428720000 } + }, + { "ticker": "BADSYM", "error": "NOT_FOUND", "message": "Ticker not found." } + ] +} +``` + +Fields FinAlly's `MassiveDataSource` cares about per ticker: + +| Field | Meaning | Maps to | +|---|---|---| +| `ticker` | Symbol | `PriceUpdate.ticker` | +| `day.c` (fallback: `lastTrade.p`) | Latest/current price | `PriceUpdate.price` | +| `prevDay.c` | Previous close | Baseline for computing `previous_price` on the first poll | +| `updated` | Nanosecond timestamp of last update | `PriceUpdate.timestamp` | +| `todaysChangePerc` | % change since prev close | Available for display, though FinAlly computes its own tick-over-tick change | + +Using the official client instead of raw `requests` (client method for this endpoint is the +v2 "snapshot all" call โ€” verify the exact method name against the `massive` version pinned in +`pyproject.toml`, since the client has been migrating callers toward `list_universal_snapshots()`, +the v3 cross-asset equivalent described next): + +```python +from massive import RESTClient + +client = RESTClient(api_key=API_KEY) +snapshot = client.get_snapshot_all("stocks", tickers=["AAPL", "GOOGL", "MSFT"]) +for t in snapshot: + print(t.ticker, t.day.close, t.prev_day.close) +``` + +### 5.2 Alternative: unified/universal snapshot (v3) + +A newer, cross-asset-class endpoint that also accepts a batched ticker list +(`ticker.any_of=AAPL,MSFT`, up to 250) and is what Massive now recommends for new integrations. +Response shape differs slightly (nested `session` instead of `day`/`prevDay`). Either endpoint +works for FinAlly; the v2 multi-ticker snapshot above is simpler and its field names map more +directly onto our `PriceUpdate` model, so that's the one documented in `MARKET_INTERFACE.md`. + +``` +GET /v3/snapshot?ticker.any_of=AAPL,GOOGL,MSFT&apiKey=YOUR_API_KEY +``` + +### 5.3 End-of-day (EOD) โ€” all tickers in one call + +The **Grouped Daily** endpoint returns OHLCV for *every* US stock ticker for one trading date in +a single response โ€” useful for EOD backfill/seeding without per-ticker requests: + +``` +GET /v2/aggs/grouped/locale/us/market/stocks/{date}?adjusted=true&apiKey=YOUR_API_KEY +``` + +```python +resp = requests.get( + f"https://api.massive.com/v2/aggs/grouped/locale/us/market/stocks/2026-07-30", + params={"adjusted": "true", "apiKey": API_KEY}, + timeout=10, +) +``` + +### 5.4 End-of-day (EOD) โ€” single ticker, previous close + +For a quick "yesterday's close" for one symbol: + +``` +GET /v2/aggs/ticker/{ticker}/prev?adjusted=true&apiKey=YOUR_API_KEY +``` + +Response fields: `c` (close), `h` (high), `l` (low), `o` (open), `v` (volume), `vw` (VWAP), +`t` (timestamp). + +### 5.5 Historical bars (for future charting needs beyond SSE-accumulated sparklines) + +``` +GET /v2/aggs/ticker/{ticker}/range/{multiplier}/{timespan}/{from}/{to}?adjusted=true&sort=asc&limit=50000 +``` + +```python +aggs = [] +for a in client.list_aggs(ticker="AAPL", multiplier=1, timespan="day", + from_="2026-01-01", to="2026-07-30", limit=50000): + aggs.append(a) +``` + +FinAlly doesn't need this initially โ€” the frontend builds sparklines from the SSE stream it has +already seen since page load (per `PLAN.md` ยง10) โ€” but it's here in case a "load more history" +feature is added later. + +## 6. Why FinAlly Polls REST Instead of Using the WebSocket + +Massive offers a WebSocket product (`wss://socket.massive.com/stocks`, channels like `T.AAPL` for +trades, `Q.AAPL` for quotes, `AM.AAPL` for minute aggregates) for true tick-by-tick streaming. +FinAlly does **not** use it, per `PLAN.md` ยง6: + +- The free tier's WebSocket access is far more restricted than even the 5 req/min REST limit +- A persistent outbound WebSocket connection from the backend adds reconnection/backoff + complexity that a simple polling loop avoids +- FinAlly's own client-facing stream is already SSE (`/api/stream/prices`), which is one-way and + polling-friendly โ€” the backend's *internal* refresh cadence (2โ€“15s against Massive) is decoupled + from the *external* cadence it pushes to the browser (~500ms, reusing the last known cache value + between polls), so students don't see "choppy" updates even though the upstream data itself only + changes every several seconds on the free tier + +## 7. Error Handling Notes + +- Per-ticker errors come back *inside* a 200 OK batch response (`"error": "NOT_FOUND"` in the + ticker's own object) rather than failing the whole request โ€” the client must check each entry. +- A `429 Too Many Requests` means the poll interval is too aggressive for the current plan tier; + back off and keep serving the last cached prices rather than raising to the frontend. +- Network/timeout errors should also fall back to last-known-cache values so a transient Massive + outage doesn't blank out the watchlist โ€” the SSE stream should never emit "no data." + +## 8. Sources + +- [Polygon.io is Now Massive](https://massive.com/blog/polygon-is-now-massive) +- [Stocks REST API Overview](https://massive.com/docs/rest/stocks/overview) +- [Full Market Snapshot](https://massive.com/docs/rest/stocks/snapshots/full-market-snapshot) +- [Single Ticker Snapshot](https://massive.com/docs/rest/stocks/snapshots/single-ticker-snapshot) +- [Unified Snapshot](https://massive.com/docs/rest/stocks/snapshots/unified-snapshot) +- [Previous Day Bar (OHLC)](https://massive.com/docs/rest/stocks/aggregates/previous-day-bar) +- [Daily Market Summary / Grouped Daily](https://massive.com/docs/rest/stocks/aggregates/daily-market-summary) +- [Custom Bars (OHLC)](https://massive.com/docs/rest/stocks/aggregates/custom-bars) +- [Massive + Python blog post](https://massive.com/blog/polygon-io-with-python-for-stock-market-data) +- [massive-com/client-python (GitHub)](https://github.com/massive-com/client-python) +- [What is the request limit for Massive's RESTful APIs?](https://massive.com/knowledge-base/article/what-is-the-request-limit-for-massives-restful-apis) diff --git a/planning/review.md b/planning/review.md new file mode 100644 index 000000000..e71b472c2 --- /dev/null +++ b/planning/review.md @@ -0,0 +1,243 @@ +# Review: Changes Since Last Commit + +Base commit: `14550e1 Ready for Teams`. Reviewed via `git status`/`git diff` for tracked files, plus +direct reading of all untracked files/directories, cross-checked against `planning/PLAN.md`, +`CLAUDE.md`, `backend/CLAUDE.md`, and the actual implementation in `backend/app/market/` +(`cache.py`, `models.py`, `massive_client.py`, `simulator.py`). All findings below were +independently verified against source, not just asserted. + +Files in scope: +- Modified: `.claude/settings.json`, `README.md` +- Untracked: `.claude/agents/` (`change-reviewer.md`, `codex-reviewer.md`, `reviewer.md`), + `.claude/commands/doc-review.md`, `planning/MARKET_INTERFACE.md`, `planning/MARKET_SIMULATOR.md`, + `planning/MASSIVE_API.md`, `planning/review.md` (this file) + +--- + +## 1. planning/MARKET_INTERFACE.md, MARKET_SIMULATOR.md, MASSIVE_API.md โ€” CRITICAL + +These three new top-level `planning/` files share filenames with, but are near-total rewrites of, +files that already exist in `planning/archive/`. Diffing new vs. archived confirms they are not +duplicates โ€” they describe a materially different design, and the **archived** versions are what +the shipped, tested code in `backend/app/market/` actually implements. `CLAUDE.md` states the +market data component is complete and points readers to `planning/MARKET_DATA_SUMMARY.md` and +`planning/archive/`; it gives no indication a second, competing, top-level copy of these design +docs should exist or that the design is being revisited. Concrete, verified mismatches: + +### 1a. PriceCache API mismatch +- `planning/MARKET_INTERFACE.md` (line ~100-102) specifies `self._lock = asyncio.Lock()` and + `async def set(self, update: PriceUpdate) -> None`, i.e. callers construct `PriceUpdate` objects + themselves and await an async setter. +- The real `backend/app/market/cache.py` uses `from threading import Lock` (a synchronous lock) + and a synchronous `def update(self, ticker: str, price: float, timestamp: float | None = None) + -> PriceUpdate`. The cache itself constructs the `PriceUpdate` and computes `previous_price`. + There is no `set()` method and nothing here is `async`. +- Code written against the new doc (`await cache.set(PriceUpdate(...))`) would not run against the + real class at all โ€” wrong method name, wrong signature, wrong sync/async model. + +### 1b. PriceUpdate model mismatch +- New doc (`MARKET_INTERFACE.md` lines ~21-31): `timestamp: datetime`, a `Direction(str, Enum)` + with `UP`/`DOWN`/`FLAT` stored as a field, `@dataclass(frozen=True)` (no `slots`). +- Real `backend/app/market/models.py`: `timestamp: float` (Unix seconds), `direction` is a + computed `@property` returning a plain `str` ("up"/"down"/"flat"), the dataclass is + `@dataclass(frozen=True, slots=True)`, and there's a `to_dict()` serialization helper used for + SSE transmission that the new doc never mentions. + +### 1c. MassiveDataSource / Massive API call shape mismatch +- `MARKET_INTERFACE.md` and `MASSIVE_API.md` (line ~140) both sketch + `client.get_snapshot_all("stocks", tickers=[...])` โ€” market type passed as a raw string โ€” and + describe parsing `entry.day.close` / `entry.prev_day.close` as the seed for the first poll. +- The real `backend/app/market/massive_client.py` imports `SnapshotMarketType` from + `massive.rest.models` and calls `get_snapshot_all(market_type=SnapshotMarketType.STOCKS, + tickers=...)` โ€” an enum, not a string โ€” and only ever reads `snap.last_trade.price` / + `snap.last_trade.timestamp`; it never touches `day` or `prev_day`. Because the real code funnels + every poll through the same `cache.update()` used by the simulator, a ticker's first price + update always has `previous_price == price` ("flat"), not seeded from `prevDay.close` as the new + doc describes. +- A future agent following the new doc's sketch to "align" or refactor `massive_client.py` would + break it against the pinned `massive` client โ€” the real code's use of the enum (not a string) + was presumably arrived at by reading the actual library, and the new doc regresses that. + +### 1d. GBMSimulator mismatch โ€” seeding and sector grouping +- `MARKET_SIMULATOR.md` (lines ~13, 112-114) specifies `GBMSimulator(tickers, seed: int | None = + None)` built on `np.random.default_rng(seed)`, explicitly naming "deterministic enough to test + (seedable RNG)" as a design goal. +- The real `backend/app/market/simulator.py` has no `seed` parameter โ€” `GBMSimulator.__init__` + takes only `tickers`, `dt`, `event_probability`, and draws from the **global** + `np.random.standard_normal` / `random.random()` / `random.uniform()` state rather than an + injectable `Generator`. This is a genuine behavioral gap versus the doc's own stated goal, not a + naming difference โ€” tests against the real simulator cannot be seeded the way the doc implies. +- The real `simulator.py` imports `TSLA_CORR` alongside `INTRA_TECH_CORR` / `INTRA_FINANCE_CORR` / + `CROSS_GROUP_CORR` from `seed_prices.py`, i.e. TSLA is deliberately special-cased with its own + correlation constant rather than folded into a generic sector map โ€” a distinct design choice the + new doc's simpler `SECTOR`/`SAME_SECTOR_CORR` scheme doesn't capture. + +### Why this matters +Two documents with identical filenames now live in two different `planning/` locations with +contradictory content, and nothing in the repo states which is authoritative. The archived copies +match the real code; the new top-level copies do not. If these are meant to describe a **proposed +future refactor**, that intent needs to be explicit (e.g., "proposed redesign โ€” not yet +implemented, see open questions") so nobody mistakes them for current documentation. If they were +added by accident (e.g., regenerated from a stale prompt without reading the existing +implementation), they should be deleted โ€” stale, wrong specs sitting next to a completed, tested +subsystem are actively harmful, since the natural first move for an agent picking up chat/portfolio +work is to read `planning/*.md`, and these three files would hand it an incorrect contract for +`app/market/`. + +**Recommendation:** Either (a) delete these three new files and rely on +`planning/archive/` + `planning/MARKET_DATA_SUMMARY.md`, or (b) if a redesign is genuinely +intended, clearly label them as a proposal, state *why* the change is warranted (none of the +observed differences are motivated in the text as written), and reconcile them with the currently +passing test suite (73 tests per `README.md`/`MARKET_DATA_SUMMARY.md`) before anyone implements +against them. + +--- + +## 2. .claude/settings.json โ€” Medium/High + +Adds a `Stop` hook: +``` +"command": "if [ -z \"$FINALLY_STOP_HOOK_ACTIVE\" ]; then FINALLY_STOP_HOOK_ACTIVE=1 claude -p 'Use the change-reviewer agent to review all changes since the last commit and write the result to planning/review.md'; fi" +``` + +- **Re-entrancy guard is correct.** Prefixing `FINALLY_STOP_HOOK_ACTIVE=1` onto the `claude -p` + invocation scopes the env var to that child process (and any hooks it triggers), so the spawned + review session's own Stop event sees the guard set and skips re-triggering. No hole found here. +- **High: the hook's prompt and the agent it invokes disagree.** The hook dispatches to "the + change-reviewer agent" with the instruction "review all changes since the last commit," but + `.claude/agents/change-reviewer.md`'s body says only: "You review the file planning/Plan.md and + write your feedback to planning/review.md." This is a real, load-bearing ambiguity โ€” this very + review run had to decide whether to scope narrowly to `PLAN.md` or broadly to all changes. + Until `change-reviewer.md`'s body is fixed to match both its own `description` and what the hook + actually asks it to do, automatic Stop-triggered reviews risk silently narrowing scope back to + just `planning/PLAN.md`, defeating the hook's purpose (which is specifically to catch drift + across all changes โ€” including the Sec 1 issue above, which a `PLAN.md`-only review would never + surface, since `PLAN.md` itself is unchanged in this diff). +- **Medium: runs on every Stop, synchronously, with a full nested `claude -p` invocation, and + always clobbers `planning/review.md`.** This fires on every turn boundary the harness treats as + "Stop," not just meaningful checkpoints. Since the target file is always overwritten with no + append/versioning, a substantive review can be silently replaced by a near-empty/no-op review the + next time the hook fires with nothing new to review. Consider gating on `git diff --quiet && + git status --porcelain` (skip if nothing changed since the last commit) to avoid needless LLM + calls and noisy overwrites. +- No other keys changed; the `enabledPlugins` block is untouched. JSON is well-formed. + +--- + +## 3. .claude/agents/ (new: change-reviewer.md, codex-reviewer.md, reviewer.md) + +### 3a. change-reviewer.md โ€” description/body mismatch (confirmed, causes real ambiguity) +- Frontmatter `description`: "carry out a comprehensive review of all changes since the last + commit." +- Body: "You review the file planning/Plan.md and write your feedback to planning/review.md." +- As noted in Sec 2, this directly caused scope ambiguity for both the Stop hook and this review + invocation. Fix by rewriting the body to match the description โ€” review the full diff and + untracked files, not just `PLAN.md` โ€” since that's clearly the intended behavior given how it's + invoked from `settings.json`. + +### 3b. reviewer.md โ€” near-duplicate with the same mismatch +- `description`: generic "reviews code and provides feedback on improvements, best practices, and + potential issues." +- `body`: byte-for-byte identical to `change-reviewer.md`'s โ€” "You review the file + planning/Plan.md and write your feedback to planning/review.md." +- `reviewer.md` and `change-reviewer.md` are currently functionally identical despite different + names/descriptions, and neither's body matches its own description. Consolidate to one agent, or + differentiate them explicitly (e.g., `Reviewer` for ad-hoc review of a named doc, `change-reviewer` + specifically for the "since last commit" workflow the hook uses). +- Naming convention is inconsistent: `change-reviewer` / `codex-reviewer` are lowercase-hyphenated; + `Reviewer` is TitleCase. Pick one convention. + +### 3c. codex-reviewer.md โ€” case-sensitivity bug + undocumented external dependency +- Shells out to: `codex exec "please review the file planning/plan.md and write your feedback to + planning/review.md"`. +- The actual file is `planning/PLAN.md`, not `planning/plan.md`. This works by accident on macOS's + default case-insensitive filesystem but will fail to find the file on any case-sensitive + filesystem โ€” Linux, most CI runners, and the project's own Docker image (Python 3.12 slim on + Linux, per `PLAN.md` ยง11). Fix the casing. +- This agent depends on an external `codex` CLI not mentioned anywhere in `CLAUDE.md` or + `planning/PLAN.md` as part of the toolchain (`uv`, Node/npm, and Docker are the documented + stack). If `codex` isn't installed/authenticated, this agent fails with a raw shell error rather + than a clear message. At minimum document the dependency; ideally add a pre-check with a clear + failure message. + +### 3d. Overlapping write targets, no coordination +- `change-reviewer.md`, `reviewer.md`, and `codex-reviewer.md` (via `codex`) all unconditionally + overwrite `planning/review.md` โ€” no append, no timestamp, no namespacing. The previous run's + feedback is silently lost every time any of the three fires. With the new Stop hook auto-firing + one of them on every Stop event, this is no longer theoretical. +- `.claude/commands/doc-review.md` uses a *different* convention entirely โ€” it appends findings + into a new section **within the reviewed doc itself**, not a separate file. Two incompatible + "where does feedback go" conventions now exist side by side in the same change set; worth + standardizing on one. + +### 3e. No tool-access restriction on any of the three new agents +- None declare a `Tools:` frontmatter field, so all default to full tool access (confirmed via the + environment's agent listing, which shows "Tools: All tools" for all three). Given their stated + job โ€” read a doc (or diff), write feedback to another doc โ€” scoping to read-only plus a narrow + write allowance (no arbitrary Bash/network) would reduce blast radius, particularly for + `codex-reviewer.md`, which already shells out to an external binary. + +--- + +## 4. .claude/commands/doc-review.md + +- Typo: "add questions, clarifications, or feedback **toa** new section at the end" โ€” missing + space, should read "to a new section." +- See Sec 3d above: this command's "append into the same file" convention conflicts with the + agents' "overwrite a separate `planning/review.md`" convention. Both patterns exist in this + change set for functionally the same task ("review a planning doc"). + +--- + +## 5. README.md โ€” no issues found + +The diff accurately reflects reality: it demotes "Quick Start" (the Docker one-liner) to a +"planned" state, adds an honest "Status" section calling out that only the market data backend +(`backend/app/market/`) is built, links to `planning/MARKET_DATA_SUMMARY.md`, and replaces the +Docker quick-start with the actual runnable command (`cd backend && uv sync && uv run +market_data_demo.py`), which matches `backend/CLAUDE.md`'s documented demo command and the +`backend/market_data_demo.py` file present in the repo. The trimmed directory-structure block +(dropping `frontend/`, `test/`, `db/`, `scripts/`, none of which exist yet) is also accurate. Low +risk, well-scoped, nothing to flag. + +Minor nit (not worth a severity tier): the "Not yet started" bullet list mentions "Database schema, +portfolio/trade endpoints, watchlist endpoints" but doesn't explicitly call out the `chat_messages` +table / chat persistence from `PLAN.md` ยง7 โ€” a very small gap, purely cosmetic. + +--- + +## Summary by Severity + +**Critical** +- Sec 1: `planning/MARKET_INTERFACE.md`, `MARKET_SIMULATOR.md`, `MASSIVE_API.md` describe a design + that diverges from the actual, tested `backend/app/market/` implementation in at least four + concrete, verified ways (async vs. sync `PriceCache` API, `PriceUpdate` shape/types, Massive + client call signature and fields read, `GBMSimulator` seeding and TSLA sector handling) โ€” and + they duplicate/shadow filenames that already exist, correctly, in `planning/archive/`, with no + authoritativeness marker anywhere in the repo. This is the highest-impact issue: left as-is, it + will actively mislead whoever next extends the market data layer or builds downstream code + against it (portfolio valuation, SSE consumers, chat trade execution). + +**High** +- Sec 2 / 3a: `change-reviewer.md`'s body ("review planning/Plan.md") doesn't match either its own + `description` or the prompt the new Stop hook actually sends it ("review all changes since the + last commit"). This is a live bug โ€” this very review run had to resolve that exact ambiguity โ€” + and if resolved the wrong way by a future automated run, it would have missed the Sec 1 issue + entirely, since `PLAN.md` itself is unchanged in this diff. + +**Medium** +- Sec 2: Stop hook fires unconditionally on every Stop event and always clobbers + `planning/review.md` with no versioning; consider a `git diff --quiet` guard to skip no-op runs. +- Sec 3b-3e: `reviewer.md` duplicates `change-reviewer.md` with the same description/body mismatch; + `codex-reviewer.md` has a case-sensitivity bug (`plan.md` vs. `PLAN.md`) that will break on + Linux/CI and an undocumented `codex` CLI dependency; three agents plus one slash command + implement two incompatible "where does feedback go" conventions; none of the new agents restrict + tool access. + +**Low** +- Sec 4: Typo ("toa" โ†’ "to a") in `.claude/commands/doc-review.md`. +- Sec 5: `README.md`'s "Not yet started" list omits explicit mention of chat message persistence + (very minor, cosmetic). + +**No issues** +- Sec 5: `README.md` changes are accurate, honest about project status, and well-scoped. From 834461b50508e11d812d7a4f97d078d76542530a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 31 Jul 2026 12:35:38 +0000 Subject: [PATCH 2/2] Add detailed market data backend design document Documents the unified MarketDataSource interface, PriceCache, GBM simulator, and Massive REST client with code snippets that reflect the actual shipped implementation in backend/app/market/ (verified against source and the passing test suite), rather than the divergent drafts flagged in planning/review.md. --- planning/MARKET_DATA_DESIGN.md | 1163 ++++++++++++++++++++++++++++++++ 1 file changed, 1163 insertions(+) create mode 100644 planning/MARKET_DATA_DESIGN.md diff --git a/planning/MARKET_DATA_DESIGN.md b/planning/MARKET_DATA_DESIGN.md new file mode 100644 index 000000000..0c933bf9e --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,1163 @@ +# Market Data Backend โ€” Design + +Authoritative, implementation-accurate design for the FinAlly market data subsystem: the unified +interface, the in-memory price cache, the GBM simulator, the Massive (Polygon.io) REST client, the +SSE streaming endpoint, and FastAPI lifecycle wiring. + +**Status:** this describes the code as it actually ships in `backend/app/market/` (8 modules, 73 +passing tests, 84% coverage). It is generated from a direct reading of the source, not derived from +an earlier proposal โ€” see `planning/MARKET_DATA_SUMMARY.md` for the short version and +`planning/archive/` for prior-iteration drafts (some details there, e.g. simulator seeding and the +Massive call shape, were superseded during code review and are corrected here). + +## Table of Contents + +1. [Architecture at a Glance](#1-architecture-at-a-glance) +2. [Data Model โ€” `models.py`](#2-data-model--modelspy) +3. [Unified Interface โ€” `interface.py`](#3-unified-interface--interfacepy) +4. [Price Cache โ€” `cache.py`](#4-price-cache--cachepy) +5. [Factory โ€” `factory.py`](#5-factory--factorypy) +6. [Simulator โ€” `seed_prices.py` + `simulator.py`](#6-simulator--seed_pricespy--simulatorpy) +7. [Massive API Client โ€” `massive_client.py`](#7-massive-api-client--massive_clientpy) +8. [SSE Streaming Endpoint โ€” `stream.py`](#8-sse-streaming-endpoint--streampy) +9. [FastAPI Lifecycle Integration](#9-fastapi-lifecycle-integration) +10. [Watchlist Coordination](#10-watchlist-coordination) +11. [Testing Strategy](#11-testing-strategy) +12. [Error Handling & Edge Cases](#12-error-handling--edge-cases) +13. [Configuration Reference](#13-configuration-reference) + +--- + +## 1. Architecture at a Glance + +``` + create_market_data_source(cache) + โ”‚ + MASSIVE_API_KEY set & non-empty? + โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ดโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” + yes no + โ”‚ โ”‚ + โ–ผ โ–ผ + MassiveDataSource SimulatorDataSource + (REST poller, ~15s) (GBM, ~500ms) + โ”‚ โ”‚ + โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ + โ–ผ + PriceCache + (thread-safe, in-memory, + version counter) + โ”‚ + โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ผโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” + โ–ผ โ–ผ โ–ผ + GET /api/stream/prices Portfolio Trade + (SSE, ~500ms) valuation execution +``` + +`backend/app/market/` is a self-contained subsystem. Everything outside it โ€” REST routes, trade +execution, the frontend โ€” depends only on `PriceCache` and `PriceUpdate`. Neither the simulator nor +the Massive client is ever referenced by name outside this package; they're selected once, at +startup, by `create_market_data_source()`. + +``` +backend/app/market/ +โ”œโ”€โ”€ __init__.py # re-exports PriceUpdate, PriceCache, MarketDataSource, +โ”‚ # create_market_data_source, create_stream_router +โ”œโ”€โ”€ models.py # PriceUpdate +โ”œโ”€โ”€ interface.py # MarketDataSource ABC +โ”œโ”€โ”€ cache.py # PriceCache +โ”œโ”€โ”€ factory.py # create_market_data_source() +โ”œโ”€โ”€ seed_prices.py # SEED_PRICES, TICKER_PARAMS, DEFAULT_PARAMS, correlation constants +โ”œโ”€โ”€ simulator.py # GBMSimulator + SimulatorDataSource +โ”œโ”€โ”€ massive_client.py # MassiveDataSource +โ””โ”€โ”€ stream.py # create_stream_router() โ€” SSE endpoint factory +``` + +--- + +## 2. Data Model โ€” `models.py` + +`PriceUpdate` is the only object that crosses the boundary out of `app/market/`. Both data sources +produce it (indirectly, via `PriceCache.update()` โ€” see ยง4); every consumer (SSE, portfolio +valuation, trade execution) consumes only this type. + +```python +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass(frozen=True, slots=True) +class PriceUpdate: + """Immutable snapshot of a single ticker's price at a point in time.""" + + ticker: str + price: float + previous_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: + 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: + """'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: + """Single serialization point, used by the SSE endpoint (and any REST + response that echoes a price).""" + return { + "ticker": self.ticker, + "price": self.price, + "previous_price": self.previous_price, + "timestamp": self.timestamp, + "change": self.change, + "change_percent": self.change_percent, + "direction": self.direction, + } +``` + +Design notes: + +- **`frozen=True, slots=True`** โ€” value objects created many times a second across two hot loops + (simulator tick, Massive poll); frozen makes them safe to hand to concurrent readers without + copying, slots trims per-instance memory. +- **`timestamp` is `float` (Unix seconds), not `datetime`** โ€” cheaper to construct, trivially + JSON-serializable, and matches what both producers naturally have on hand (`time.time()` for the + simulator, a converted Massive epoch-millis field for the live feed). +- **`direction`/`change`/`change_percent` are computed properties**, not stored fields โ€” they can + never drift out of sync with `price`/`previous_price`. + +--- + +## 3. Unified Interface โ€” `interface.py` + +```python +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class MarketDataSource(ABC): + """Contract for market data providers. + + Implementations push price updates into a shared PriceCache on their own + schedule. Downstream code never calls the data source directly for prices โ€” + it reads from the cache. + + Lifecycle: + source = create_market_data_source(cache) + await source.start(["AAPL", "GOOGL", ...]) + # ... app runs ... + await source.add_ticker("TSLA") + await source.remove_ticker("GOOGL") + # ... app shutting down ... + await source.stop() + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing price updates for the given tickers. + + Starts a background task that periodically writes to the PriceCache. + Must be called exactly once. Calling start() twice is undefined behavior. + """ + + @abstractmethod + async def stop(self) -> None: + """Stop the background task and release resources. Safe to call + multiple times.""" + + @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.""" +``` + +This is the entire surface area either implementation exposes. Neither `SimulatorDataSource` nor +`MassiveDataSource` exposes its internals (no HTTP client, no simulation state) publicly, so callers +can't accidentally couple to one concrete implementation. + +**Why the source pushes into a cache instead of returning prices from a method call:** it decouples +timing. The simulator ticks every ~500ms; Massive polls every ~15s on the free tier. The SSE layer +reads the cache at its own fixed ~500ms cadence regardless of which producer is active or how often +it actually refreshes data โ€” see ยง8. + +--- + +## 4. Price Cache โ€” `cache.py` + +The single point of truth between producer and every consumer. One data source writes; the SSE +endpoint, portfolio valuation, and trade execution all read. + +```python +from __future__ import annotations + +import time +from threading import Lock + +from .models import PriceUpdate + + +class PriceCache: + """Thread-safe in-memory cache of the latest price for each ticker. + + Writers: SimulatorDataSource or MassiveDataSource (one at a time). + Readers: SSE streaming endpoint, portfolio valuation, trade execution. + """ + + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._lock = Lock() + self._version: int = 0 # Monotonically increasing; bumped on every update + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + """Record a new price for a ticker. Returns the created PriceUpdate. + + Automatically computes direction and change from the previous price. + If this is the first update for the ticker, previous_price == price + (direction == 'flat'). + """ + with self._lock: + ts = timestamp or time.time() + prev = self._prices.get(ticker) + previous_price = prev.price if prev else price + + update = PriceUpdate( + ticker=ticker, + price=round(price, 2), + previous_price=round(previous_price, 2), + timestamp=ts, + ) + self._prices[ticker] = update + self._version += 1 + return update + + def get(self, ticker: str) -> PriceUpdate | None: + with self._lock: + return self._prices.get(ticker) + + def get_all(self) -> dict[str, PriceUpdate]: + """Snapshot of all current prices. Returns a shallow copy so callers + can iterate without holding the lock.""" + with self._lock: + return dict(self._prices) + + def get_price(self, ticker: str) -> float | None: + update = self.get(ticker) + return update.price if update else None + + def remove(self, ticker: str) -> None: + with self._lock: + self._prices.pop(ticker, None) + + @property + def version(self) -> int: + """Bumped on every update() call. Lets the SSE loop cheaply detect + 'has anything changed since I last looked' without diffing the dict.""" + 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 +``` + +**Design points:** + +- **`PriceCache` owns `PriceUpdate` construction** โ€” callers pass a raw `(ticker, price, + timestamp?)`, and the cache itself looks up the previous value and computes `previous_price`. + Neither data source constructs a `PriceUpdate` directly; both just call `cache.update(...)`. This + keeps "what counts as the previous price" defined in exactly one place. +- **`threading.Lock`, not `asyncio.Lock`.** The Massive client's synchronous `RESTClient` call runs + inside `asyncio.to_thread(...)` โ€” a real OS thread, which an `asyncio.Lock` would not protect + against. A plain mutex works correctly from both the event loop and any thread-pool worker. +- **The cache never loses a value except via explicit `remove()`.** Nothing times it out. This is + what makes Massive rate limits or a slow poll invisible to the frontend โ€” stale-but-present data + beats a gap. + +--- + +## 5. Factory โ€” `factory.py` + +The only place in the codebase that reads `MASSIVE_API_KEY`. + +```python +from __future__ import annotations + +import logging +import os + +from .cache import PriceCache +from .interface import MarketDataSource +from .massive_client import MassiveDataSource +from .simulator import SimulatorDataSource + +logger = logging.getLogger(__name__) + + +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """Create the appropriate market data source based on environment variables. + + - MASSIVE_API_KEY set and non-empty โ†’ MassiveDataSource (real market data) + - Otherwise โ†’ SimulatorDataSource (GBM simulation) + + Returns an unstarted source. Caller must await source.start(tickers). + """ + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + + if api_key: + logger.info("Market data source: Massive API (real data)") + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + else: + logger.info("Market data source: GBM Simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +`massive` is a core dependency (`pyproject.toml`), imported at module top level in both +`massive_client.py` and here โ€” no lazy import. Whether or not a student ever sets +`MASSIVE_API_KEY`, `uv sync` installs `massive` and the import always succeeds; only the *choice* of +which class to instantiate is conditional. + +Usage at app startup: + +```python +price_cache = PriceCache() +source = create_market_data_source(price_cache) +await source.start(initial_tickers) # e.g. ["AAPL", "GOOGL", ...] from the watchlist table +``` + +--- + +## 6. Simulator โ€” `seed_prices.py` + `simulator.py` + +The default data source โ€” no external dependencies beyond `numpy`, no API key, runs fully offline. + +### 6.1 Seed prices and per-ticker parameters โ€” `seed_prices.py` + +Constants only. Shared by `simulator.py` 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 +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, e.g. via watchlist/chat) +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"}, +} + +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, even though it's in the "tech" set +``` + +Ticker not in `SEED_PRICES`/`TICKER_PARAMS` (added at runtime via the watchlist or the AI chat) get +a random seed price in `[50, 300]` and `DEFAULT_PARAMS` โ€” see `_add_ticker_internal` below. + +### 6.2 `GBMSimulator` โ€” the math engine + +Discrete-time geometric Brownian motion: + +``` +S(t+dt) = S(t) * exp((mu - sigmaยฒ/2) * dt + sigma * sqrt(dt) * Z) +``` + +`Z` is a *correlated* standard normal draw, not independent per ticker โ€” see the Cholesky step +below โ€” so sector groups move together the way real markets do, rather than each ticker jittering +in isolation. + +```python +from __future__ import annotations + +import asyncio +import logging +import math +import random + +import numpy as np + +from .cache import PriceCache +from .interface import MarketDataSource +from .seed_prices import ( + CORRELATION_GROUPS, + CROSS_GROUP_CORR, + DEFAULT_PARAMS, + INTRA_FINANCE_CORR, + INTRA_TECH_CORR, + SEED_PRICES, + TICKER_PARAMS, + TSLA_CORR, +) + +logger = logging.getLogger(__name__) + + +class GBMSimulator: + """Geometric Brownian Motion simulator for correlated stock prices. + + Math: + S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) + + The tiny dt (~8.5e-8 for 500ms ticks over 252 trading days * 6.5h/day) + produces sub-cent moves per tick that accumulate naturally over time. + """ + + # 252 trading days * 6.5 hours/day * 3600 seconds/hour = 5,896,800 "trading seconds" per year + TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 + DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8, for a 500ms tick + + def __init__( + self, + tickers: list[str], + dt: float = DEFAULT_DT, + event_probability: float = 0.001, + ) -> None: + self._dt = dt + self._event_prob = event_probability + self._tickers: list[str] = [] + self._prices: dict[str, float] = {} + self._params: dict[str, dict[str, float]] = {} + self._cholesky: np.ndarray | None = None + + for ticker in tickers: + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + def step(self) -> dict[str, float]: + """Advance all tickers by one time step. Returns {ticker: new_price}. + 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) + + # ~0.1% chance per tick per ticker of a 2-5% shock (visual drama) + if random.random() < self._event_prob: + shock_magnitude = random.uniform(0.02, 0.05) + shock_sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock_magnitude * shock_sign + logger.debug( + "Random event on %s: %.1f%% %s", + ticker, shock_magnitude * 100, "up" if shock_sign > 0 else "down", + ) + + result[ticker] = round(self._prices[ticker], 2) + + return result + + def add_ticker(self, ticker: str) -> None: + """Add a ticker to the simulation. Rebuilds the correlation matrix.""" + if ticker in self._prices: + return + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + def remove_ticker(self, ticker: str) -> None: + """Remove a ticker from the simulation. Rebuilds the correlation matrix.""" + if ticker not in self._prices: + return + self._tickers.remove(ticker) + del self._prices[ticker] + del self._params[ticker] + self._rebuild_cholesky() + + def get_price(self, ticker: str) -> float | None: + return self._prices.get(ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + # --- internals --- + + def _add_ticker_internal(self, ticker: str) -> None: + if ticker in self._prices: + return + self._tickers.append(ticker) + self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) + self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) + + def _rebuild_cholesky(self) -> None: + """O(n^2), called on every add/remove โ€” fine at this scale (n < 50).""" + n = len(self._tickers) + if n <= 1: + self._cholesky = None + return + + corr = np.eye(n) + for i in range(n): + for j in range(i + 1, n): + rho = self._pairwise_correlation(self._tickers[i], self._tickers[j]) + corr[i, j] = corr[j, i] = rho + + self._cholesky = np.linalg.cholesky(corr) + + @staticmethod + def _pairwise_correlation(t1: str, t2: str) -> float: + """Same tech sector: 0.6. Same finance sector: 0.5. TSLA with anything: + 0.3 (it does its own thing). Everything else: 0.3.""" + tech = CORRELATION_GROUPS["tech"] + finance = CORRELATION_GROUPS["finance"] + + if t1 == "TSLA" or t2 == "TSLA": + return TSLA_CORR + if t1 in tech and t2 in tech: + return INTRA_TECH_CORR + if t1 in finance and t2 in finance: + return INTRA_FINANCE_CORR + return CROSS_GROUP_CORR +``` + +**Why Cholesky:** drawing `n` independent standard normals and left-multiplying by the Cholesky +factor `L` of a target correlation matrix `ฮฃ` (`L @ L.T == ฮฃ`) produces a vector of normals with +exactly that correlation structure. It's the standard trick for correlated Monte Carlo draws and is +cheap enough to redo on every ticker add/remove at watchlist scale (โ‰ค a few dozen names). + +**Why GBM specifically:** returns are log-normal, prices can never go negative (it's an +exponential), and `(mu, sigma)` map directly onto "steady mover" vs. "choppy" stock intuitions โ€” +easy to hand-tune per ticker and get results that feel roughly right without being predictive. + +### 6.3 `SimulatorDataSource` โ€” async wrapper implementing `MarketDataSource` + +```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 immediately so SSE has data on its very 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") + logger.info("Simulator started with %d tickers", len(tickers)) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("Simulator stopped") + + async def add_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.add_ticker(ticker) + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) # visible immediately + 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: + 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") # one bad tick doesn't kill the feed + await asyncio.sleep(self._interval) +``` + +Key behaviors: immediate cache seeding on `start()`/`add_ticker()` (no blank watchlist row while +waiting for the first tick), clean cancellation on `stop()`, and per-tick exception isolation in +`_run_loop` so a transient failure doesn't take down the whole background task. + +--- + +## 7. Massive API Client โ€” `massive_client.py` + +The optional, real-data source, active only when `MASSIVE_API_KEY` is set. Polls the batched +multi-ticker snapshot endpoint โ€” one HTTP call regardless of watchlist size โ€” on an interval sized +for the free tier's 5 requests/minute budget. + +```python +from __future__ import annotations + +import asyncio +import logging + +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +from .cache import PriceCache +from .interface import MarketDataSource + +logger = logging.getLogger(__name__) + + +class MassiveDataSource(MarketDataSource): + """MarketDataSource backed by the Massive (Polygon.io) REST API. + + Polls GET /v2/snapshot/locale/us/markets/stocks/tickers for all watched + tickers in a single API call, then writes results to the PriceCache. + + Rate limits: + - Free tier: 5 req/min โ†’ poll every 15s (default) + - Paid tiers: higher limits โ†’ poll every 2-5s + """ + + def __init__( + self, + api_key: str, + price_cache: PriceCache, + poll_interval: float = 15.0, + ) -> None: + self._api_key = api_key + self._cache = price_cache + self._interval = poll_interval + self._tickers: list[str] = [] + self._task: asyncio.Task | None = None + self._client: RESTClient | None = None + + async def start(self, tickers: list[str]) -> None: + self._client = RESTClient(api_key=self._api_key) + self._tickers = list(tickers) + await self._poll_once() # immediate first poll so the cache has data right away + self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") + logger.info( + "Massive poller started: %d tickers, %.1fs interval", len(tickers), self._interval + ) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + self._client = None + logger.info("Massive poller stopped") + + async def add_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + if ticker not in self._tickers: + self._tickers.append(ticker) + logger.info("Massive: added ticker %s (will appear on next poll)", ticker) + + async def remove_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + self._tickers = [t for t in self._tickers if t != ticker] + self._cache.remove(ticker) + logger.info("Massive: removed ticker %s", ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + # --- internal --- + + async def _poll_loop(self) -> None: + """First poll already happened in start().""" + while True: + await asyncio.sleep(self._interval) + await self._poll_once() + + async def _poll_once(self) -> None: + if not self._tickers or not self._client: + return + + try: + # RESTClient is synchronous โ€” offload to 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 + timestamp = snap.last_trade.timestamp / 1000.0 # ms -> seconds + self._cache.update(ticker=snap.ticker, price=price, timestamp=timestamp) + processed += 1 + except (AttributeError, TypeError) as e: + # Per-ticker bad entry (e.g. NOT_FOUND) โ€” skip it, keep the rest of the batch + 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) + # Never re-raise: the whole cache is left untouched (last-known prices + # keep serving), and the loop retries on the next scheduled interval. + # Common causes: 401 (bad key), 429 (rate limit), network/timeout errors. + + def _fetch_snapshots(self) -> list: + """Synchronous call to the Massive REST API. Runs inside asyncio.to_thread.""" + return self._client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=self._tickers, + ) +``` + +### Why the multi-ticker snapshot endpoint + +The **Full Market Snapshot** endpoint (`GET /v2/snapshot/locale/us/markets/stocks/tickers`) accepts +a comma-separated ticker list โ€” up to 250 symbols โ€” and returns the latest trade/quote/day bar for +each in one response. That's what makes a single `RESTClient.get_snapshot_all(...)` call cover the +entire watchlist regardless of size, which matters directly for the free tier's 5 req/min ceiling: +one poll of 10 tickers costs exactly the same as one poll of 1. + +### Fields actually consumed + +Only `snap.last_trade.price` and `snap.last_trade.timestamp` are read from each snapshot entry โ€” +the client doesn't touch `day`/`prevDay` bars. `previous_price` for direction/change is always +computed by `PriceCache.update()` from whatever was cached from the *previous poll* (see ยง4), so the +very first update for a ticker is always `direction == "flat"` (`previous_price == price`), the same +convention the simulator uses on its first tick for a new ticker. + +### Error handling philosophy + +| Situation | Behavior | +|---|---| +| Per-ticker bad entry (`NOT_FOUND`, missing fields) | That entry is skipped (`AttributeError`/`TypeError` caught); the rest of the batch is still processed. | +| `401 Unauthorized` (bad key) | Whole poll fails, logged as error; cache untouched; retried on the next interval. | +| `429 Too Many Requests` | Same โ€” logged, cache untouched, retried next interval. Sign the poll interval is too aggressive for the plan tier; raise `poll_interval` via configuration if it recurs. | +| Network timeout | Same โ€” logged, cache untouched, retried next interval. | + +The invariant across all of these: **a failed poll never clears or blanks the cache.** The SSE +stream keeps serving the last known prices, so a transient Massive outage or a rate-limit hiccup is +invisible to the frontend โ€” it just sees prices stop moving for a few cycles rather than +disappearing. + +### Why REST polling, not the Massive WebSocket + +Massive offers a WebSocket product for true tick-by-tick pushes, but FinAlly doesn't use it: the +free tier's WebSocket access is more restricted than even the 5 req/min REST limit, and a persistent +outbound WebSocket from the backend adds reconnect/backoff complexity a simple polling loop avoids. +FinAlly's own client-facing feed is already SSE โ€” one-way, polling-friendly โ€” so the backend's +internal refresh cadence against Massive (15s free tier) is fully decoupled from the external +cadence it pushes to the browser (~500ms, replaying the last known value between real upstream +updates โ€” see ยง8). + +--- + +## 8. SSE Streaming Endpoint โ€” `stream.py` + +```python +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import AsyncGenerator + +from fastapi import APIRouter, Request +from fastapi.responses import StreamingResponse + +from .cache import PriceCache + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/stream", tags=["streaming"]) + + +def create_stream_router(price_cache: PriceCache) -> APIRouter: + """Factory pattern: injects the PriceCache without module-level 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 shaped like: + + data: {"AAPL": {"ticker": "AAPL", "price": 190.50, ...}, ...} + """ + return StreamingResponse( + _generate_events(price_cache, request), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", # disable nginx buffering if proxied + }, + ) + + return router + + +async def _generate_events( + price_cache: PriceCache, + request: Request, + interval: float = 0.5, +) -> AsyncGenerator[str, None]: + yield "retry: 1000\n\n" # tell the browser to auto-reconnect after 1s on drop + + last_version = -1 + client_ip = request.client.host if request.client else "unknown" + logger.info("SSE client connected: %s", client_ip) + + try: + while True: + if await request.is_disconnected(): + logger.info("SSE client disconnected: %s", client_ip) + break + + current_version = price_cache.version + if current_version != last_version: + last_version = current_version + prices = price_cache.get_all() + if prices: + data = {ticker: update.to_dict() for ticker, update in prices.items()} + yield f"data: {json.dumps(data)}\n\n" + + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("SSE stream cancelled for: %s", client_ip) +``` + +### Wire format + +``` +data: {"AAPL":{"ticker":"AAPL","price":190.50,"previous_price":190.42,"timestamp":1707580800.5,"change":0.08,"change_percent":0.042,"direction":"up"},"GOOGL":{...}} + +``` + +Frontend consumption: + +```javascript +const eventSource = new EventSource('/api/stream/prices'); +eventSource.onmessage = (event) => { + const prices = JSON.parse(event.data); // { "AAPL": { ticker, price, ... }, ... } + // update watchlist rows, trigger flash animation on changed tickers, append to sparkline buffers +}; +``` + +### Why version-based change detection + +`price_cache.version` is bumped on every `PriceCache.update()` call (ยง4). The SSE loop polls it +every 500ms; if it hasn't changed since the last iteration, nothing is sent โ€” a single integer +comparison replaces diffing the whole price dict on every tick, and (in the Massive case) it means +the SSE loop naturally skips 29 out of 30 500ms ticks between real upstream polls, without any +Massive-specific logic in `stream.py` at all. + +### Why poll-and-push instead of event-driven (pub/sub) + +The endpoint polls the cache on a fixed interval rather than being notified by the producer. This +keeps updates evenly spaced regardless of upstream jitter, which matters because the frontend +accumulates SSE payloads into sparkline series โ€” even spacing makes for a clean chart. It also means +the SSE layer needs zero knowledge of, or coupling to, whichever `MarketDataSource` happens to be +running. + +--- + +## 9. FastAPI Lifecycle Integration + +The market data system starts and stops with the app via FastAPI's `lifespan` context manager. + +```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): + # --- startup --- + price_cache = PriceCache() + app.state.price_cache = price_cache + + source = create_market_data_source(price_cache) + app.state.market_source = source + + initial_tickers = await load_watchlist_tickers() # from SQLite; seeded with the default 10 + await source.start(initial_tickers) + + app.include_router(create_stream_router(price_cache)) + + yield # app is running + + # --- shutdown --- + await source.stop() + + +app = FastAPI(title="FinAlly", lifespan=lifespan) + + +def get_price_cache() -> PriceCache: + return app.state.price_cache + + +def get_market_source() -> MarketDataSource: + return app.state.market_source +``` + +### Consuming the cache from other routes + +```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 the watchlist table ... + await source.add_ticker(payload.ticker) + # ... + + +@router.delete("/watchlist/{ticker}") +async def remove_from_watchlist(ticker: str, source: MarketDataSource = Depends(get_market_source)): + # ... delete from the watchlist table ... + await source.remove_ticker(ticker) + # ... +``` + +Both routes are written against `MarketDataSource`, never against `SimulatorDataSource` or +`MassiveDataSource` โ€” swapping the active source (i.e. setting/unsetting `MASSIVE_API_KEY` and +restarting) requires no change here. + +--- + +## 10. Watchlist Coordination + +### Adding a ticker + +``` +User (or LLM) โ†’ POST /api/watchlist {ticker: "PYPL"} + โ†’ insert into watchlist table (SQLite) + โ†’ await source.add_ticker("PYPL") + Simulator: seeds a price, rebuilds the Cholesky matrix, writes into the cache immediately + Massive: appended to the polled ticker list, appears on the next scheduled poll + โ†’ respond with the ticker (+ price if already cached) +``` + +### Removing a ticker + +``` +User (or LLM) โ†’ DELETE /api/watchlist/PYPL + โ†’ delete from watchlist table (SQLite) + โ†’ await source.remove_ticker("PYPL") + Simulator: dropped from GBMSimulator, Cholesky rebuilt, removed from cache + Massive: dropped from the polled ticker list, removed from cache + โ†’ respond with success +``` + +### Edge case: ticker removed from the watchlist but still held + +If the user drops a ticker from the watchlist while still holding a position, the data source +must keep tracking it so portfolio valuation stays accurate โ€” the watchlist route, not the market +data layer, is responsible for this check: + +```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 nothing is held + + return {"status": "ok"} +``` + +--- + +## 11. Testing Strategy + +The real suite lives in `backend/tests/market/` (6 modules, 73 tests, 84% coverage โ€” run with +`uv run --extra dev pytest -v` from `backend/`). Shape of the coverage: + +| Module | Focus | +|---|---| +| `test_models.py` | `PriceUpdate` properties (`change`, `change_percent`, `direction`, `to_dict()`) across up/down/flat/zero-previous-price cases. | +| `test_cache.py` | `update`/`get`/`get_all`/`get_price`/`remove`, first-update-is-flat, direction/change on subsequent updates, `version` increments exactly once per `update()`. | +| `test_simulator.py` | `GBMSimulator` unit tests: prices stay positive over many steps, initial price matches the seed, `add_ticker`/`remove_ticker` (including duplicate/nonexistent no-ops), unknown tickers get a random seed in range, empty-ticker-list `step()` returns `{}`, Cholesky is `None` for a single ticker and non-`None` once a second is added. | +| `test_simulator_source.py` | `SimulatorDataSource` integration: `start()` populates the cache before the first tick, prices actually move over several ticks, `stop()` is idempotent, `add_ticker`/`remove_ticker` propagate to both the simulator and the cache. | +| `test_factory.py` | `create_market_data_source` returns `MassiveDataSource` when `MASSIVE_API_KEY` is set (`monkeypatch.setenv`) and `SimulatorDataSource` otherwise (`monkeypatch.delenv`), including the empty-string/whitespace case. | +| `test_massive.py` | `MassiveDataSource` with `_fetch_snapshots` mocked: successful batch parsing updates the cache; a malformed snapshot (missing `last_trade`) is skipped without affecting other tickers in the same batch; an exception raised during polling leaves the cache untouched rather than clearing it. | + +Representative examples: + +```python +# test_cache.py +def test_direction_up(): + cache = PriceCache() + cache.update("AAPL", 190.00) + update = cache.update("AAPL", 191.00) + assert update.direction == "up" + assert update.change == 1.00 + +def test_version_increments(): + cache = PriceCache() + v0 = cache.version + cache.update("AAPL", 190.00) + assert cache.version == v0 + 1 +``` + +```python +# test_simulator.py +def test_prices_are_positive(): + """GBM prices can never go negative (exp() is always positive).""" + sim = GBMSimulator(tickers=["AAPL"]) + for _ in range(10_000): + prices = sim.step() + assert prices["AAPL"] > 0 + +def test_cholesky_rebuilds_on_add(): + sim = GBMSimulator(tickers=["AAPL"]) + sim.add_ticker("GOOGL") + assert sim.get_tickers() == ["AAPL", "GOOGL"] +``` + +```python +# test_massive.py +async def test_api_error_does_not_crash(): + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + + with patch.object(source, "_fetch_snapshots", side_effect=Exception("network error")): + await source._poll_once() # must not raise + + assert cache.get_price("AAPL") is None # no update happened; cache stays untouched +``` + +**Why `SimulatorDataSource` also doubles as a fake for other subsystems' tests:** because both +implementations satisfy the same `MarketDataSource` ABC, anything downstream (portfolio valuation, +trade execution, chat trade auto-execution) that needs *a* data source in its own tests can just use +`SimulatorDataSource` with a short `update_interval` rather than building a bespoke fake. + +--- + +## 12. Error Handling & Edge Cases + +- **Empty watchlist at startup.** `start([])` is valid for both sources โ€” the simulator's `step()` + returns `{}`, the Massive poller's `_poll_once()` short-circuits (`if not self._tickers: return`). + The SSE endpoint simply sends nothing until a ticker is added. +- **Trading a ticker with no cached price yet** (just added, Massive hasn't polled): the trade route + should treat `price_cache.get_price(ticker) is None` as a 400, not attempt the trade at a + fabricated price. In practice this only affects the Massive path โ€” the simulator seeds the cache + synchronously inside `add_ticker()`. +- **Invalid Massive API key.** The first poll (inside `start()`) fails with `401`; it's caught, + logged, and the poller keeps retrying every `poll_interval` seconds. Nothing crashes; the SSE + connection reports "connected" but streams no data for that ticker set until the key is fixed + and the process restarted (env vars are read once, at `create_market_data_source()` call time). +- **Thread safety under load.** `PriceCache`'s `threading.Lock` guards a tiny critical section (dict + read + write); at watchlist scale (โ‰ค dozens of tickers, one writer, N SSE readers) contention is + negligible. This isn't a bottleneck worth engineering around for this project. +- **Floating-point/precision.** Prices are `round()`-ed to 2 decimal places at the point of writing + into `GBMSimulator._prices` and again in `PriceCache.update()`; the GBM exponential formulation is + numerically stable and always positive, so there's no risk of drifting or negative prices even + over a long-running demo session. + +--- + +## 13. Configuration Reference + +| Parameter | Where | Default | Notes | +|---|---|---|---| +| `MASSIVE_API_KEY` | Environment variable | unset | Non-empty โ†’ `MassiveDataSource`; unset/empty โ†’ `SimulatorDataSource`. Read once, in `create_market_data_source()`. | +| `update_interval` | `SimulatorDataSource.__init__` | `0.5` s | Simulator tick cadence. | +| `event_probability` | `GBMSimulator.__init__` (via `SimulatorDataSource`) | `0.001` | Per-ticker, per-tick chance of a 2โ€“5% shock move. | +| `dt` | `GBMSimulator.__init__` | `~8.48e-8` (`0.5 / TRADING_SECONDS_PER_YEAR`) | GBM time step, derived from the 500ms tick cadence over a 252-day ร— 6.5h trading year. | +| `poll_interval` | `MassiveDataSource.__init__` | `15.0` s | Sized for the free tier's 5 req/min; lower for paid tiers. | +| SSE push interval | `_generate_events(interval=...)` | `0.5` s | Cache poll cadence on the SSE side; independent of the upstream producer's own cadence. | +| SSE retry directive | `_generate_events` (`retry: 1000`) | `1000` ms | Browser `EventSource` auto-reconnect delay after a dropped connection. | + +Everything in this table is a constructor default โ€” all are overridable per-instance without +touching call sites elsewhere, since every consumer depends only on the `MarketDataSource` / +`PriceCache` interfaces.