From b91fc3934910fa06e89f31a342a68982839b3359 Mon Sep 17 00:00:00 2001 From: heseraj Date: Mon, 24 Aug 2026 22:05:24 -0700 Subject: [PATCH 1/2] added github cli to be able to check work on web --- .claude/agents/codewhale-reviwer.md | 13 + .claude/agents/reviewer.md | 6 + .claude/commands/doc-review.md | 1 + .codewhale/state/subagents.v1.lock | 0 README.md | 71 ++-- backend/app/market/massive_client.py | 17 +- backend/tests/market/test_massive.py | 16 +- .../.claude-plugins/plugin.json | 5 + independent-reviewer/hooks/hooks.json | 14 + planning/MARKET_DATA_SUMMARY.md | 104 ------ planning/MARKET_INTERFACE.md | 345 ++++++++++++++++++ planning/MARKET_SIMULATOR.md | 255 +++++++++++++ planning/MASSIVE_API.md | 233 ++++++++++++ planning/PLAN.md | 145 +++++--- planning/REVIEW.md | 25 ++ planning/review-claude.md | 71 ++++ planning/review-deepseek.md | 72 ++++ 17 files changed, 1184 insertions(+), 209 deletions(-) create mode 100644 .claude/agents/codewhale-reviwer.md create mode 100644 .claude/agents/reviewer.md create mode 100644 .claude/commands/doc-review.md create mode 100644 .codewhale/state/subagents.v1.lock create mode 100644 independent-reviewer/.claude-plugins/plugin.json create mode 100644 independent-reviewer/hooks/hooks.json delete mode 100644 planning/MARKET_DATA_SUMMARY.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 create mode 100644 planning/review-claude.md create mode 100644 planning/review-deepseek.md diff --git a/.claude/agents/codewhale-reviwer.md b/.claude/agents/codewhale-reviwer.md new file mode 100644 index 000000000..86224cfda --- /dev/null +++ b/.claude/agents/codewhale-reviwer.md @@ -0,0 +1,13 @@ +--- +name: codewhale-reviwer +description: This agent reviews code comprehensively and provides feedback on potential improvements, best practices, and optimizations using deepseek and codewhale. +--- + +You are using a different AI agent to carry out a review of the file planning/PLAN.md. You will write your feedback in planning/review-deepseek.md. You will use the deepseek API to analyze the code and provide feedback on potential improvements, best practices, and optimizations. +You must execute the following shell command to carry out the review: + +`codewhale exec --auto "Please review all the changes since last commit and write your feedback in planning/review-deepseek.md"` + +write you feedback in planning/review-deepseek.md +This will run the review and save the results. +DO NOT RUN THE REVIEW YOURSELF. diff --git a/.claude/agents/reviewer.md b/.claude/agents/reviewer.md new file mode 100644 index 000000000..0d3733fc3 --- /dev/null +++ b/.claude/agents/reviewer.md @@ -0,0 +1,6 @@ +--- +name: reviewer +description: this agent reviews code comprehensively and provides feedback on potential improvements, best practices, and optimizations. +--- + +You review the file planning/PLAN.md and write your feedback in planning/review-claude.md diff --git a/.claude/commands/doc-review.md b/.claude/commands/doc-review.md new file mode 100644 index 000000000..de9bc3ba9 --- /dev/null +++ b/.claude/commands/doc-review.md @@ -0,0 +1 @@ +Review the documentation in the planning folder called $ARGUEENTS and add questions, clarifications or feedback to a new section at the end along with any opportunities to simplify diff --git a/.codewhale/state/subagents.v1.lock b/.codewhale/state/subagents.v1.lock new file mode 100644 index 000000000..e69de29bb diff --git a/README.md b/README.md index 3f2582ae2..0eb5b9a1b 100644 --- a/README.md +++ b/README.md @@ -1,62 +1,61 @@ # FinAlly — AI Trading Workstation -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. +FinAlly (Finance Ally) is a visually stunning, AI-powered trading workstation: it streams live market data, lets you trade a simulated portfolio, and includes an LLM chat assistant that can analyze your positions and execute trades on your behalf. Think Bloomberg terminal with an AI copilot. -Built entirely by coding agents as a capstone project for an agentic AI coding course. +This is the capstone project for an agentic AI coding course — built entirely by coding agents to demonstrate how orchestrated AI agents can produce a production-quality full-stack application. -## Features +## Status -- **Live price streaming** via SSE with green/red flash animations -- **Simulated portfolio** — $10k virtual cash, market orders, instant fills -- **Portfolio visualizations** — heatmap (treemap), P&L chart, positions table -- **AI chat assistant** — analyzes holdings, suggests and auto-executes trades -- **Watchlist management** — track tickers manually or via AI -- **Dark terminal aesthetic** — Bloomberg-inspired, data-dense layout +The project is under active development. The **market data subsystem** (live-price simulator, Massive/Polygon.io client, price cache, SSE streaming) is complete and lives in `backend/app/market/`. The rest of the platform — portfolio/trading, watchlist, LLM chat, frontend, and Docker packaging — is still to be built. + +See [`planning/PLAN.md`](planning/PLAN.md) for the full project specification, and [`planning/MARKET_DATA_SUMMARY.md`](planning/MARKET_DATA_SUMMARY.md) for details on the completed market data component. ## Architecture -Single Docker container serving everything on port 8000: +A single Docker container, single port (8000): + +- **Frontend**: Next.js + TypeScript, built as a static export, served by FastAPI +- **Backend**: FastAPI (Python), managed with `uv` +- **Database**: SQLite, volume-mounted for persistence +- **Real-time data**: Server-Sent Events (SSE) +- **AI**: LiteLLM → OpenRouter (Cerebras inference), structured outputs for trade execution +- **Market data**: simulator by default, real data via the Massive API if a key is provided -- **Frontend**: Next.js (static export) with TypeScript and Tailwind CSS -- **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) +Full rationale and details are in `planning/PLAN.md`. -## Quick Start +## Getting Started (backend, current state) ```bash -# Clone and configure -cp .env.example .env -# Add your OPENROUTER_API_KEY to .env +cd backend +uv sync --extra dev # install dependencies +uv run --extra dev pytest # run tests +uv run market_data_demo.py # live terminal dashboard with simulated prices +``` -# Run with Docker -docker build -t finally . -docker run -v finally-data:/app/db -p 8000:8000 --env-file .env finally +See [`backend/CLAUDE.md`](backend/CLAUDE.md) and [`backend/README.md`](backend/README.md) for backend developer docs. -# Open http://localhost:8000 -``` +A one-command Docker launch (`scripts/start_mac.sh` / `scripts/start_windows.ps1`) is planned once the frontend and portfolio/chat backend are in place. ## Environment Variables -| Variable | Required | Description | -|---|---|---| -| `OPENROUTER_API_KEY` | Yes | OpenRouter API key for AI chat | -| `MASSIVE_API_KEY` | No | Massive (Polygon.io) key for real market data; omit to use simulator | -| `LLM_MOCK` | No | Set `true` for deterministic mock LLM responses (testing) | +```bash +OPENROUTER_API_KEY= # required for LLM chat +MASSIVE_API_KEY= # optional; enables real market data instead of the simulator +LLM_MOCK=false # set true for deterministic mock LLM responses (testing) +``` ## Project Structure ``` finally/ -├── frontend/ # Next.js static export -├── backend/ # FastAPI uv project -├── planning/ # Project documentation and agent contracts -├── test/ # Playwright E2E tests -├── db/ # SQLite volume mount (runtime) -└── scripts/ # Start/stop helpers +├── frontend/ # Next.js TypeScript app (static export) — planned +├── backend/ # FastAPI uv project — market data implemented +├── planning/ # Project specification and design docs for agents +├── scripts/ # Docker start/stop scripts — planned +├── test/ # Playwright E2E tests — planned +└── db/ # SQLite volume mount point ``` ## License -See [LICENSE](LICENSE). +See [`LICENSE`](LICENSE). diff --git a/backend/app/market/massive_client.py b/backend/app/market/massive_client.py index 00bc7b2aa..b1d229b0b 100644 --- a/backend/app/market/massive_client.py +++ b/backend/app/market/massive_client.py @@ -6,7 +6,6 @@ import logging from massive import RESTClient -from massive.rest.models import SnapshotMarketType from .cache import PriceCache from .interface import MarketDataSource @@ -21,8 +20,11 @@ class MassiveDataSource(MarketDataSource): 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 + - Free (Basic) tier: does not include this endpoint at all (403) — a + Starter plan or above is required for get_snapshot_all. + - Starter and above: effectively unlimited (soft guidance: stay under + 100 req/s), so the poll interval is a design choice rather than a + rate-limit constraint; default here is a conservative 15s. """ def __init__( @@ -99,8 +101,8 @@ async def _poll_once(self) -> None: for snap in snapshots: try: price = snap.last_trade.price - # Massive timestamps are Unix milliseconds → convert to seconds - timestamp = snap.last_trade.timestamp / 1000.0 + # LastTrade timestamps are Unix nanoseconds → convert to seconds + timestamp = snap.last_trade.sip_timestamp / 1_000_000_000 self._cache.update( ticker=snap.ticker, price=price, @@ -122,7 +124,4 @@ async def _poll_once(self) -> None: def _fetch_snapshots(self) -> list: """Synchronous call to the Massive REST API. Runs in a thread.""" - return self._client.get_snapshot_all( - market_type=SnapshotMarketType.STOCKS, - tickers=self._tickers, - ) + return self._client.get_snapshot_all("stocks", self._tickers) diff --git a/backend/tests/market/test_massive.py b/backend/tests/market/test_massive.py index cdd7dbd24..cd44f872d 100644 --- a/backend/tests/market/test_massive.py +++ b/backend/tests/market/test_massive.py @@ -8,13 +8,13 @@ from app.market.massive_client import MassiveDataSource -def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock: +def _make_snapshot(ticker: str, price: float, timestamp_ns: int) -> MagicMock: """Create a mock Massive snapshot object.""" snap = MagicMock() snap.ticker = ticker snap.last_trade = MagicMock() snap.last_trade.price = price - snap.last_trade.timestamp = timestamp_ms + snap.last_trade.sip_timestamp = timestamp_ns return snap @@ -34,8 +34,8 @@ async def test_poll_updates_cache(self): source._client = MagicMock() # Satisfy the _poll_once guard mock_snapshots = [ - _make_snapshot("AAPL", 190.50, 1707580800000), - _make_snapshot("GOOGL", 175.25, 1707580800000), + _make_snapshot("AAPL", 190.50, 1707580800000000000), + _make_snapshot("GOOGL", 175.25, 1707580800000000000), ] with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): @@ -55,7 +55,7 @@ async def test_malformed_snapshot_skipped(self): source._tickers = ["AAPL", "BAD"] source._client = MagicMock() # Satisfy the _poll_once guard - good_snap = _make_snapshot("AAPL", 190.50, 1707580800000) + good_snap = _make_snapshot("AAPL", 190.50, 1707580800000000000) bad_snap = MagicMock() bad_snap.ticker = "BAD" bad_snap.last_trade = None # Will cause AttributeError @@ -84,7 +84,7 @@ async def test_api_error_does_not_crash(self): assert cache.get_price("AAPL") is None # No update happened async def test_timestamp_conversion(self): - """Test that timestamps are converted from milliseconds to seconds.""" + """Test that timestamps are converted from nanoseconds to seconds.""" cache = PriceCache() source = MassiveDataSource( api_key="test-key", @@ -94,7 +94,7 @@ async def test_timestamp_conversion(self): source._tickers = ["AAPL"] source._client = MagicMock() # Satisfy the _poll_once guard - mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)] + mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000000000)] with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): await source._poll_once() @@ -189,7 +189,7 @@ async def test_start_immediate_poll(self): cache = PriceCache() source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) - mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)] + mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000000000)] with patch("app.market.massive_client.RESTClient"): with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): diff --git a/independent-reviewer/.claude-plugins/plugin.json b/independent-reviewer/.claude-plugins/plugin.json new file mode 100644 index 000000000..3acd01ede --- /dev/null +++ b/independent-reviewer/.claude-plugins/plugin.json @@ -0,0 +1,5 @@ +{ + "name": "independent-reviewer", + "description": "carries an indpendent reviews of all changes since last commit", + "version": "0.1.0" +} diff --git a/independent-reviewer/hooks/hooks.json b/independent-reviewer/hooks/hooks.json new file mode 100644 index 000000000..85328d737 --- /dev/null +++ b/independent-reviewer/hooks/hooks.json @@ -0,0 +1,14 @@ +{ + "hooks": { + "stops": [ + { + "hooks": [ + { + "type": "command", + "command": "codewhale exec auto \"review changes since last commit and suggest improvements and resulst in a doccuments planning/codewhale-reveiw.md\"" + } + ] + } + ] + } +} diff --git a/planning/MARKET_DATA_SUMMARY.md b/planning/MARKET_DATA_SUMMARY.md deleted file mode 100644 index ae518283a..000000000 --- a/planning/MARKET_DATA_SUMMARY.md +++ /dev/null @@ -1,104 +0,0 @@ -# Market Data Backend — Summary - -**Status:** Complete, tested, reviewed, all issues resolved. - -## What Was Built - -A complete market data subsystem in `backend/app/market/` (8 modules, ~500 lines) providing live price simulation and real market data via a unified interface. - -### Architecture - -``` -MarketDataSource (ABC) -├── SimulatorDataSource → GBM simulator (default, no API key needed) -└── MassiveDataSource → Polygon.io REST poller (when MASSIVE_API_KEY set) - │ - ▼ - PriceCache (thread-safe, in-memory) - │ - ├──→ SSE stream endpoint (/api/stream/prices) - ├──→ Portfolio valuation - └──→ Trade execution -``` - -### Modules - -| File | Purpose | -|------|---------| -| `models.py` | `PriceUpdate` — immutable frozen dataclass (ticker, price, previous_price, timestamp, change, direction) | -| `interface.py` | `MarketDataSource` — abstract base class defining `start/stop/add_ticker/remove_ticker/get_tickers` | -| `cache.py` | `PriceCache` — thread-safe price store with version counter for SSE change detection | -| `seed_prices.py` | Realistic seed prices, per-ticker GBM params (drift/volatility), correlation groups | -| `simulator.py` | `GBMSimulator` (Geometric Brownian Motion with Cholesky-correlated moves) + `SimulatorDataSource` | -| `massive_client.py` | `MassiveDataSource` — REST polling client for Polygon.io via the `massive` package | -| `factory.py` | `create_market_data_source()` — selects simulator or Massive based on `MASSIVE_API_KEY` env var | -| `stream.py` | `create_stream_router()` — FastAPI SSE endpoint factory using version-based change detection | - -### Key Design Decisions - -- **Strategy pattern** — both data sources implement the same ABC; downstream code is source-agnostic -- **PriceCache as single point of truth** — producers write, consumers read; no direct coupling -- **GBM with correlated moves** — Cholesky decomposition of sector-based correlation matrix; tech stocks correlate at 0.6, finance at 0.5, cross-sector at 0.3 -- **Random shock events** — ~0.1% chance per tick per ticker of a 2-5% move for visual drama -- **SSE over WebSockets** — simpler, one-way push, universal browser support - -## Test Suite - -**73 tests, all passing.** 6 test modules in `backend/tests/market/`. - -| Module | Tests | Coverage | -|--------|-------|----------| -| test_models.py | 11 | models.py: 100% | -| test_cache.py | 13 | cache.py: 100% | -| test_simulator.py | 17 | simulator.py: 98% | -| test_simulator_source.py | 10 | (integration tests) | -| test_factory.py | 7 | factory.py: 100% | -| test_massive.py | 13 | massive_client.py: 56% (expected — API methods mocked) | - -Overall coverage: 84%. - -## Code Review & Fixes Applied - -A comprehensive code review identified 7 issues. All were resolved: - -1. **pyproject.toml build config** — added `[tool.hatch.build.targets.wheel] packages = ["app"]` -2. **Lazy imports removed** — `massive` is a core dependency; imports moved to top level -3. **SSE return type fixed** — `_generate_events` annotated as `AsyncGenerator[str, None]` -4. **Public `get_tickers()`** — added to `GBMSimulator` to avoid private attribute access -5. **Correlation constants cleaned up** — removed unused `DEFAULT_CORR`, consolidated into `CROSS_GROUP_CORR` -6. **Unused test imports removed** — `pytest`, `math`, `asyncio` cleaned from 4 test files -7. **Massive test mocks fixed** — `source._client` set in tests, patches target correct names - -## Demo - -A Rich terminal demo is available at `backend/market_data_demo.py`: - -```bash -cd backend -uv run market_data_demo.py -``` - -Displays a live-updating dashboard with all 10 tickers, sparklines, color-coded direction arrows, and an event log for notable price moves. Runs 60 seconds or until Ctrl+C. - -## Usage for Downstream Code - -```python -from app.market import PriceCache, create_market_data_source - -# Startup -cache = PriceCache() -source = create_market_data_source(cache) # Reads MASSIVE_API_KEY -await source.start(["AAPL", "GOOGL", "MSFT", ...]) - -# Read prices -update = cache.get("AAPL") # PriceUpdate or None -price = cache.get_price("AAPL") # float or None -all_prices = cache.get_all() # dict[str, PriceUpdate] - -# Dynamic watchlist -await source.add_ticker("TSLA") -await source.remove_ticker("GOOGL") - -# Shutdown -await source.stop() -``` diff --git a/planning/MARKET_INTERFACE.md b/planning/MARKET_INTERFACE.md new file mode 100644 index 000000000..95d11236b --- /dev/null +++ b/planning/MARKET_INTERFACE.md @@ -0,0 +1,345 @@ +# Market Data Interface Design + +Unified Python interface for market data in FinAlly. Two implementations — `SimulatorDataSource` and `MassiveDataSource` — behind one abstract `MarketDataSource` interface, selected at startup by `create_market_data_source()`. All downstream code (SSE streaming, portfolio valuation, trade execution) is source-agnostic: it only ever talks to the shared `PriceCache`, never to a data source directly. + +This document describes the interface as implemented today in `backend/app/market/` (`interface.py`, `cache.py`, `models.py`, `factory.py`, `simulator.py`, `massive_client.py`, `stream.py`), and calls out one place where the implementation is behind the project spec (`planning/PLAN.md` §6). + +## Core Data Model (`models.py`) + +```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: + if self.price > self.previous_price: + return "up" + elif self.price < self.previous_price: + return "down" + return "flat" + + def to_dict(self) -> dict: + 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, + } +``` + +`change`, `change_percent`, and `direction` are computed properties rather than stored fields — the class stays a minimal, immutable fact (`ticker`, `price`, `previous_price`, `timestamp`) and derives the display fields on read. This is the only object type that leaves the market data layer; SSE payloads and history responses are both built from `to_dict()`. + +## Abstract 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. + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing price updates for the given tickers. Call exactly once.""" + + @abstractmethod + async def stop(self) -> None: + """Stop the background task and release resources. Safe to call multiple times.""" + + @abstractmethod + async def add_ticker(self, ticker: str) -> None: + """Add a ticker to the active set. No-op if already present.""" + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Remove a ticker from the active set. Also removes it from the PriceCache.""" + + @abstractmethod + def get_tickers(self) -> list[str]: + """Return the current list of actively tracked tickers.""" +``` + +Both implementations own a background `asyncio.Task` (an internal poll/step loop) started in `start()` and cancelled in `stop()`. The interface intentionally has no `get_price()` method — prices are pulled from `PriceCache`, not pushed synchronously through this interface, so callers never block on a network round trip or a simulation step. + +## Price Cache (`cache.py`) + +Thread-safe in-memory store both data sources write to and the SSE streamer / portfolio code reads from. + +```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.""" + + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._lock = Lock() + self._version: int = 0 # bumped on every update; lets SSE detect "anything changed" + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + 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]: + 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: + return self._version +``` + +A `threading.Lock` (not an `asyncio.Lock`) is deliberate: `MassiveDataSource` calls the synchronous Massive SDK inside `asyncio.to_thread`, so cache writes can happen from a worker thread as well as the event loop. The `version` counter lets the SSE generator (`stream.py`) cheaply skip re-serializing and re-sending the full price map when nothing has changed between its poll ticks, instead of diffing the dict. + +### ⚠️ Gap vs. `planning/PLAN.md` §6: no rolling history buffer yet + +The project spec says: + +> A single background task ... writes to an in-memory `PriceCache`, which holds, per ticker: the latest price, previous price, timestamp, **and a bounded rolling history buffer (most recent 500 points)** ... SSE streams and the history endpoint (`GET /api/market/history/{ticker}`) both read from this same cache. + +**As implemented today, `PriceCache` only stores the single latest `PriceUpdate` per ticker** — no history buffer, and `GET /api/market/history/{ticker}` does not exist yet anywhere in `backend/app/`. This means sparkline pre-population and the main chart's "seed from history, then extend from SSE" behavior (PLAN.md §10) currently have no server-side data source to call on load. + +This is the one place this document intentionally diverges from "describe what's implemented" — the recommended shape, consistent with the rest of the design, is a small addition to `PriceCache` rather than a new component: + +```python +from collections import deque + +class PriceCache: + def __init__(self, history_size: int = 500) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._history: dict[str, deque[PriceUpdate]] = {} + self._history_size = history_size + self._lock = Lock() + self._version: int = 0 + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + with self._lock: + ... + self._prices[ticker] = update + self._history.setdefault(ticker, deque(maxlen=self._history_size)).append(update) + self._version += 1 + return update + + def get_history(self, ticker: str) -> list[PriceUpdate]: + with self._lock: + return list(self._history.get(ticker, ())) +``` + +`deque(maxlen=...)` gives O(1) bounded appends for free instead of manual list slicing. `remove()` should also drop `self._history.pop(ticker, None)` to match. The SSE update cadence is ~500ms (`SimulatorDataSource`'s default `update_interval`), so a 500-point buffer covers roughly 4 minutes of simulator history per ticker — for Massive's 15s poll interval it covers roughly 2 hours, which is plenty for a sparkline. `GET /api/market/history/{ticker}` (PLAN.md §8) is then a thin FastAPI route: `return [u.to_dict() for u in price_cache.get_history(ticker)]`, 404 if the ticker isn't tracked. + +## Factory (`factory.py`) + +Selects the data source at startup based on `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: + """MASSIVE_API_KEY set and non-empty -> MassiveDataSource, else -> SimulatorDataSource. + + Returns an unstarted source. Caller must await source.start(tickers). + """ + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + + if api_key: + logger.info("Market data source: Massive API (real data)") + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + else: + logger.info("Market data source: GBM Simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +`.strip()` matters: an `.env` file with `MASSIVE_API_KEY=` (present but empty) must fall through to the simulator, not attempt a Massive connection with an empty key. + +## `MassiveDataSource` (`massive_client.py`) + +Polls the Massive snapshot endpoint for the tracked ticker set — see `MASSIVE_API.md` for the wire-level details and confirmed issues with the current field/method usage (`get_snapshot_all` call shape, `last_trade` timestamp field name, free-tier plan restrictions). Structurally: + +```python +class MassiveDataSource(MarketDataSource): + 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 isn't empty on connect + self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") + + async def _poll_once(self) -> None: + if not self._tickers or not self._client: + return + try: + snapshots = await asyncio.to_thread(self._fetch_snapshots) + for snap in snapshots: + try: + self._cache.update(ticker=snap.ticker, price=snap.last_trade.price, ...) + except (AttributeError, TypeError): + ... # skip malformed snapshot, keep going + except Exception: + ... # log and let the next poll cycle retry — never crash the loop +``` + +Two resilience choices worth keeping in any future rewrite: an **immediate poll inside `start()`** (so the SSE stream has data on the very first client connection, not just after the first `poll_interval` elapses), and a **per-snapshot try/except inside the batch loop** (one malformed ticker in a 20-ticker response doesn't drop the other 19). + +## `SimulatorDataSource` (`simulator.py`) + +Wraps a `GBMSimulator` (full math and structure in `MARKET_SIMULATOR.md`) in the same async-loop shape: + +```python +class SimulatorDataSource(MarketDataSource): + 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) + for ticker in tickers: + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + + async def _run_loop(self) -> None: + while True: + try: + if self._sim: + for ticker, price in self._sim.step().items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +Both implementations seed the cache synchronously inside `start()` before the background task exists, for the same reason: a client that connects to the SSE stream immediately after startup should see prices on its first tick, not an empty payload. + +## Integration with SSE (`stream.py`) + +`create_stream_router(price_cache)` returns a FastAPI `APIRouter` (a factory, not a module-level router, so the cache is injected rather than reached via a global). The generator polls `price_cache.version` every 500ms and only serializes/sends when it has changed: + +```python +async def _generate_events(price_cache: PriceCache, request: Request, interval: float = 0.5): + yield "retry: 1000\n\n" + last_version = -1 + while True: + if await request.is_disconnected(): + break + current_version = price_cache.version + if current_version != last_version: + last_version = current_version + prices = price_cache.get_all() + if prices: + yield f"data: {json.dumps({t: u.to_dict() for t, u in prices.items()})}\n\n" + await asyncio.sleep(interval) +``` + +## Tracked Ticker Set + +Per PLAN.md §6, the set passed to `start()` / `add_ticker()` / `remove_ticker()` must always be **watchlist ∪ positions.ticker** for the user, not just the watchlist — this keeps a live price available for the positions table and portfolio valuation even if a held ticker is removed from the watchlist. Neither `MarketDataSource` implementation enforces this itself; it's the responsibility of the calling code (wherever watchlist/position mutations happen) to recompute the union and call `add_ticker`/`remove_ticker` accordingly, only removing a ticker once it's in neither set. + +## File Structure + +``` +backend/ + app/ + market/ + __init__.py # Public API re-exports + models.py # PriceUpdate dataclass + interface.py # MarketDataSource ABC + cache.py # PriceCache (thread-safe, latest price + version counter) + factory.py # create_market_data_source() + massive_client.py # MassiveDataSource + simulator.py # GBMSimulator + SimulatorDataSource + seed_prices.py # SEED_PRICES, TICKER_PARAMS, correlation constants + stream.py # create_stream_router() — SSE endpoint factory +``` + +## Lifecycle + +1. **App startup**: create `PriceCache`, call `create_market_data_source(price_cache)`, then `await source.start(initial_tickers)` where `initial_tickers` is the watchlist ∪ positions union at boot +2. **Watchlist/position changes**: recompute the tracked set, call `source.add_ticker()` / `source.remove_ticker()` as needed +3. **SSE streaming**: `GET /api/stream/prices` reads `PriceCache.get_all()` on a 500ms poll, gated by `version` +4. **History pre-population** *(spec'd, not yet implemented — see gap above)*: `GET /api/market/history/{ticker}` reads `PriceCache.get_history(ticker)` +5. **Trade execution**: reads current price via `PriceCache.get(ticker)` / `get_price(ticker)` +6. **App shutdown**: `await source.stop()` diff --git a/planning/MARKET_SIMULATOR.md b/planning/MARKET_SIMULATOR.md new file mode 100644 index 000000000..5bcc3ea81 --- /dev/null +++ b/planning/MARKET_SIMULATOR.md @@ -0,0 +1,255 @@ +# Market Simulator Design + +Approach and code structure for simulating realistic stock prices when `MASSIVE_API_KEY` is unset — FinAlly's default mode. Implemented in `backend/app/market/simulator.py` (the `GBMSimulator` engine and the `SimulatorDataSource` adapter) and `backend/app/market/seed_prices.py` (constants). + +## Overview + +The simulator uses **Geometric Brownian Motion (GBM)** — the standard model underlying Black-Scholes option pricing. Prices evolve continuously with random noise, can never go negative (the update is multiplicative via `exp()`), and produce the lognormal return distribution seen in real markets, rather than a plain random walk that could drift to zero or negative. + +`SimulatorDataSource` steps the simulation every `update_interval` seconds (default **0.5s**, matching the SSE cadence in `MARKET_INTERFACE.md`), producing a continuous, correlated stream of price changes. + +## GBM Math + +At each time step, a price evolves as: + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +- `S(t)` — current price +- `mu` — annualized drift (expected return), e.g. `0.05` for 5%/year +- `sigma` — annualized volatility, e.g. `0.20` for 20%/year +- `dt` — this time step as a fraction of a trading year +- `Z` — a standard normal random draw (correlated across tickers — see below) + +For 500ms ticks over a ~252-trading-day, 6.5-hour-session year: + +```python +TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 +DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ≈ 8.48e-8 +``` + +This tiny `dt` produces sub-cent moves per tick that accumulate into realistic-looking intraday ranges over minutes of wall-clock time, rather than jumping wildly on every 500ms update. + +## Correlated Moves + +Real stocks don't move independently — tech names tend to move together, etc. The simulator builds a **Cholesky decomposition** of a correlation matrix and uses it to turn independent normal draws into correlated ones: given correlation matrix `C`, `L = cholesky(C)`, then `Z_correlated = L @ Z_independent`. This is standard for simulating correlated GBM paths and is exact as long as `C` is a valid (positive semi-definite) correlation matrix — which a matrix built from pairwise correlations in `[0, 1)` with a unit diagonal always is. + +Correlation structure, from `seed_prices.py`: + +```python +CORRELATION_GROUPS: dict[str, set[str]] = { + "tech": {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"}, + "finance": {"JPM", "V"}, +} + +INTRA_TECH_CORR = 0.6 # tech stocks move together +INTRA_FINANCE_CORR = 0.5 # finance stocks move together +CROSS_GROUP_CORR = 0.3 # between sectors, or either ticker unknown +TSLA_CORR = 0.3 # TSLA is in the tech set but is deliberately excluded from that 0.6 — it does its own thing +``` + +Pairwise lookup (`GBMSimulator._pairwise_correlation`) checks TSLA first — even though it's a member of the `tech` group set, it's special-cased to `0.3` with everything, never the `0.6` intra-tech rate. Unknown tickers (dynamically added, not in either group) fall through to `CROSS_GROUP_CORR = 0.3` with everyone. + +## Random Events + +Every step, each ticker independently has a small probability of a sudden shock — a 2-5% jump, up or down — layered on top of the normal GBM step. This adds drama and keeps a long-running dashboard visually interesting instead of drifting flat. + +```python +event_probability = 0.001 # ~0.1% chance per ticker per tick + +if random.random() < event_probability: + shock_magnitude = random.uniform(0.02, 0.05) + shock_sign = random.choice([-1, 1]) + price *= 1 + shock_magnitude * shock_sign +``` + +At 2 ticks/sec, `0.001` per ticker per tick means an event on a *given* ticker roughly every 500 seconds (~8 minutes); with the 10-ticker default watchlist, expect *some* ticker to have an event roughly every 50 seconds. + +## Seed Prices & Per-Ticker Parameters (`seed_prices.py`) + +```python +SEED_PRICES: dict[str, float] = { + "AAPL": 190.00, "GOOGL": 175.00, "MSFT": 420.00, "AMZN": 185.00, "TSLA": 250.00, + "NVDA": 800.00, "META": 500.00, "JPM": 195.00, "V": 280.00, "NFLX": 600.00, +} + +TICKER_PARAMS: dict[str, dict[str, float]] = { + "AAPL": {"sigma": 0.22, "mu": 0.05}, + "GOOGL": {"sigma": 0.25, "mu": 0.05}, + "MSFT": {"sigma": 0.20, "mu": 0.05}, + "AMZN": {"sigma": 0.28, "mu": 0.05}, + "TSLA": {"sigma": 0.50, "mu": 0.03}, # high volatility + "NVDA": {"sigma": 0.40, "mu": 0.08}, # high volatility, strong drift + "META": {"sigma": 0.30, "mu": 0.05}, + "JPM": {"sigma": 0.18, "mu": 0.04}, # low volatility (bank) + "V": {"sigma": 0.17, "mu": 0.04}, # low volatility (payments) + "NFLX": {"sigma": 0.35, "mu": 0.05}, +} + +DEFAULT_PARAMS: dict[str, float] = {"sigma": 0.25, "mu": 0.05} +``` + +Tickers added dynamically that aren't in `SEED_PRICES` start at a random price uniform in `[$50, $300]`, and use `DEFAULT_PARAMS` (`sigma=0.25`, `mu=0.05`) since there's no real-world volatility to look up for an arbitrary user-entered symbol. + +## Implementation (`GBMSimulator`) + +```python +import math +import random +import numpy as np + + +class GBMSimulator: + """GBM simulator for correlated stock prices.""" + + TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 + DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR + + 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 every tracked ticker by one time step. Hot path — called every 500ms.""" + n = len(self._tickers) + if n == 0: + return {} + + z_independent = np.random.standard_normal(n) + z = self._cholesky @ z_independent if self._cholesky is not None else z_independent + + result: dict[str, float] = {} + for i, ticker in enumerate(self._tickers): + mu, sigma = self._params[ticker]["mu"], self._params[ticker]["sigma"] + + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z[i] + self._prices[ticker] *= math.exp(drift + diffusion) + + if random.random() < self._event_prob: + shock = random.uniform(0.02, 0.05) * random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock + + result[ticker] = round(self._prices[ticker], 2) + + return result + + def add_ticker(self, ticker: str) -> None: + if ticker in self._prices: + return + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + def remove_ticker(self, ticker: str) -> None: + if ticker not in self._prices: + return + self._tickers.remove(ticker) + del self._prices[ticker] + del self._params[ticker] + self._rebuild_cholesky() + + def get_price(self, ticker: str) -> float | None: + return self._prices.get(ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + def _add_ticker_internal(self, ticker: str) -> None: + if ticker in self._prices: + return + self._tickers.append(ticker) + self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) + self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) + + def _rebuild_cholesky(self) -> None: + """O(n^2) rebuild whenever tickers are added/removed. n stays small (<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: + if t1 == "TSLA" or t2 == "TSLA": + return TSLA_CORR + tech, finance = CORRELATION_GROUPS["tech"], CORRELATION_GROUPS["finance"] + 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 +``` + +`step()` is the hot path — called on every simulator tick for every tracked ticker — so it stays allocation-light: one vectorized `numpy` draw per step for all tickers at once, rather than per-ticker `random.gauss()` calls, and the Cholesky matrix is cached and only rebuilt on `add_ticker`/`remove_ticker`, not on every `step()`. + +## `SimulatorDataSource` — Wiring into the Async Loop + +`GBMSimulator` itself is synchronous and has no knowledge of asyncio, the cache, or timing — `SimulatorDataSource` (in the same `simulator.py` module) is the thin `MarketDataSource` adapter that owns the background task and writes results into the shared `PriceCache` (full contract in `MARKET_INTERFACE.md`): + +```python +class SimulatorDataSource(MarketDataSource): + 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) + for ticker in tickers: + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + + async def _run_loop(self) -> None: + while True: + try: + if self._sim: + for ticker, price in self._sim.step().items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +The `try/except Exception` around the step (rather than letting an exception propagate and kill the task) matters: a single bad draw shouldn't silently stop all price updates for the rest of the process — the loop logs and keeps going on the next tick. + +## File Structure + +``` +backend/ + app/ + market/ + simulator.py # GBMSimulator (pure math engine) + SimulatorDataSource (async adapter) + seed_prices.py # SEED_PRICES, TICKER_PARAMS, DEFAULT_PARAMS, correlation constants +``` + +`seed_prices.py` holds only constant dictionaries — no logic — so tuning starting prices, per-ticker volatility, or sector correlations never requires touching `simulator.py`. + +## Behavior Notes + +- Prices never go negative — GBM's `exp()` update is always positive, unlike an additive random walk +- The tiny `dt` produces sub-cent moves per tick that accumulate naturally into realistic multi-minute price action +- With `sigma=0.50` (TSLA) and 2 ticks/sec, a simulated trading session produces roughly the right intraday range for a genuinely volatile stock +- The correlation matrix is guaranteed positive semi-definite by construction (pairwise correlations in `[0.3, 0.6]`, unit diagonal), so `np.linalg.cholesky` never raises — no need to defensively catch a non-PSD failure +- Cholesky rebuild is `O(n^2)` but `n` stays under ~50 (the watchlist cap of 20 plus any held positions), so it's cheap even on every watchlist edit +- `round(self._prices[ticker], 2)` in `step()`'s return value only rounds what's *emitted* to the cache each tick — the simulator's internal `self._prices[ticker]` stays unrounded so tiny per-tick drifts aren't lost to repeated rounding error over a long-running session diff --git a/planning/MASSIVE_API.md b/planning/MASSIVE_API.md new file mode 100644 index 000000000..48f489897 --- /dev/null +++ b/planning/MASSIVE_API.md @@ -0,0 +1,233 @@ +# Massive API Reference (formerly Polygon.io) + +Reference documentation for the Massive REST API as used in FinAlly's market data layer (`backend/app/market/massive_client.py`). Polygon.io rebranded as Massive on **October 30, 2025**; existing API keys, accounts, and the legacy `api.polygon.io` host continue to work. This document reflects the live docs at `massive.com/docs` and the `massive` PyPI package (currently pinned `>=1.0.0` in `backend/pyproject.toml`, resolving to `2.2.0` in `backend/uv.lock`) as of 2026-08-24. + +## Overview + +- **Base URL**: `https://api.massive.com` (legacy `https://api.polygon.io` still supported) +- **Python package**: `massive` (`pip install -U massive` / `uv add massive`), min Python 3.9 +- **Auth**: API key, either passed to `RESTClient(api_key=...)` or read automatically from the `MASSIVE_API_KEY` environment variable when `RESTClient()` is constructed with no arguments — this is exactly the env var FinAlly already uses to select the data source, so no separate credential wiring is needed +- **Response objects**: the client deserializes JSON into typed model classes (e.g. `TickerSnapshot`, `Agg`, `LastTrade`) rather than returning raw dicts + +## Pricing Tiers & Rate Limits + +| Tier | Price | Rate limit | Data recency | History | +|------|-------|-----------|---------------|---------| +| Basic (free) | $0/mo | 5 requests/min | End-of-day only | 2 years | +| Starter | $29/mo | Unlimited (soft guidance: stay under 100 req/s) | 15-minute delayed | 5 years | +| Developer | $79/mo | Unlimited | 15-minute delayed, + trades data | 10 years | +| Advanced | $199/mo | Unlimited | Real-time | 20+ years, + quotes/financials | + +**⚠️ This matters for FinAlly specifically**: the multi-ticker snapshot endpoint (§1 below), which is the one FinAlly polls, is documented as **not included in the Basic (free) plan** — it requires Starter or above. A free-tier `MASSIVE_API_KEY` will get a `403` from `get_snapshot_all`, not a slow-but-working response. The `previous close` aggregate endpoint (§3) *is* available on Basic. See "Discrepancies & Gotchas" at the end of this document — this affects whether FinAlly's "free tier: poll every 15s" assumption is viable at all, since the endpoint it needs isn't free. + +## Client Initialization + +```python +from massive import RESTClient + +# Reads MASSIVE_API_KEY from the environment automatically +client = RESTClient() + +# Or pass explicitly +client = RESTClient(api_key="your_key_here") +``` + +## Endpoints Used in FinAlly + +### 1. Snapshot — All Tickers (Primary Endpoint) + +Gets current prices for multiple tickers in a **single API call** — the endpoint FinAlly's poller uses. + +**REST**: `GET /v2/snapshot/locale/us/markets/stocks/tickers?tickers=AAPL,GOOGL,MSFT` + +Query params: `tickers` (comma-separated, case-sensitive), `include_otc` (bool, default `false`). + +**Python client** — verified against the official `stocks-snapshots_all.py` example in `massive-com/client-python`: + +```python +from massive import RESTClient +from massive.rest.models import TickerSnapshot, Agg + +client = RESTClient() # MASSIVE_API_KEY read from environment + +tickers = ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"] + +# market_type is a plain string, and it — plus tickers — are POSITIONAL args +snapshots = client.get_snapshot_all("stocks", tickers) + +for snap in snapshots: + if not isinstance(snap, TickerSnapshot): + continue + print(f"{snap.ticker}: ${snap.last_trade.price}") + print(f" Today's change: {snap.todays_change_percent:.2f}%") + if isinstance(snap.day, Agg): + print(f" Day OHLC: O={snap.day.open} H={snap.day.high} L={snap.day.low} C={snap.day.close}") + print(f" Volume: {snap.day.volume}") +``` + +**`TickerSnapshot` fields** (verified from the model source, `massive/rest/models/snapshot.py`): + +| Field | Type | Notes | +|---|---|---| +| `ticker` | `str` | | +| `day` | `Agg` | today's running OHLCV bar | +| `prev_day` | `Agg` | **previous session's** OHLCV bar — there is no `day.previous_close` | +| `last_trade` | `LastTrade` | | +| `last_quote` | `LastQuote` | bid/ask | +| `min` | `MinuteSnapshot` | latest minute bar | +| `todays_change` | `float` | absolute change vs. previous close | +| `todays_change_percent` | `float` | percentage change vs. previous close — use this directly instead of computing it from `day`/`prev_day` | +| `updated` | `int` | Unix nanoseconds | +| `fair_market_value` | `float` | Business plan only | + +**`LastTrade` fields** (verified from `massive/rest/models/trades.py`) — **there is no plain `.timestamp` field**: + +| Field | Type | Notes | +|---|---|---| +| `price` | `float` | | +| `size` | `float` | | +| `exchange` | `int` | numeric exchange code, not a string like `"XNYS"` | +| `sip_timestamp` | `int` | Unix **nanoseconds**, SIP-received time — the one to use for "when was this priced" | +| `participant_timestamp` | `int` | Unix nanoseconds, exchange-reported time | +| `trf_timestamp` | `int` | Unix nanoseconds, TRF-reported time (off-exchange trades only) | +| `conditions` | `list[int]` | trade condition codes | +| `correction`, `id`, `trf_id`, `tape`, `fractional_size` | — | rarely needed for a snapshot poller | + +`Agg` fields (day/prev_day): `open`, `high`, `low`, `close`, `volume`, `vwap`, `timestamp` (Unix **milliseconds** — note this differs from `LastTrade`'s nanosecond timestamps), `transactions`, `otc`. + +### 2. Single Ticker Snapshot + +For a detail view when the user clicks one ticker. + +**REST**: `GET /v3/snapshot?ticker.any_of=AAPL` (the newer "Unified Snapshot" endpoint, also covers options/forex/crypto in one call, `ticker.any_of` accepts up to 250 comma-separated tickers) — or the older per-ticker path `GET /v2/snapshot/locale/us/markets/stocks/tickers/{ticker}`. + +**Python client**: + +```python +snapshot = client.get_snapshot_ticker("stocks", ticker="AAPL") + +print(f"Price: ${snapshot.last_trade.price}") +print(f"Bid/Ask: ${snapshot.last_quote.bid} / ${snapshot.last_quote.ask}") +print(f"Day range: ${snapshot.day.low} - ${snapshot.day.high}") +``` + +### 3. Previous Close + +Previous day's OHLC for a ticker — available even on the free Basic tier, useful for seed prices or as a Basic-tier fallback when the snapshot endpoint isn't accessible. + +**REST**: `GET /v2/aggs/ticker/{ticker}/prev` + +**Python client** — verified against the official `stocks-previous_close.py` example: + +```python +from massive import RESTClient + +client = RESTClient() +aggs = client.get_previous_close_agg("AAPL") + +for agg in aggs: + print(f"Previous close: ${agg.close}") + print(f"OHLC: O={agg.open} H={agg.high} L={agg.low} C={agg.close}") + print(f"Volume: {agg.volume}") +``` + +**Response** (raw REST): +```json +{ + "adjusted": true, + "status": "OK", + "ticker": "AAPL", + "resultsCount": 1, + "results": [ + { "T": "AAPL", "o": 115.55, "h": 117.59, "l": 114.13, "c": 115.97, "v": 131704427, "vw": 116.3058, "t": 1605042000000 } + ] +} +``` + +### 4. Aggregates (Bars) + +Historical OHLCV bars over a date range — not needed for live polling, but the natural source if FinAlly ever adds a longer-range historical chart (beyond the in-memory rolling buffer described in `MARKET_INTERFACE.md`). + +**REST**: `GET /v2/aggs/ticker/{ticker}/range/{multiplier}/{timespan}/{from}/{to}` + +**Python client**: +```python +aggs = list(client.list_aggs( + ticker="AAPL", + multiplier=1, + timespan="day", + from_="2026-07-01", + to="2026-08-01", + limit=50000, +)) + +for a in aggs: + print(f"t={a.timestamp} O={a.open} H={a.high} L={a.low} C={a.close} V={a.volume}") +``` + +### 5. Last Trade / Last Quote + +Individual endpoints for the most recent trade or NBBO quote on one ticker — rarely needed once the snapshot endpoint is in use, since a snapshot already embeds both. + +```python +trade = client.get_last_trade(ticker="AAPL") +print(f"Last trade: ${trade.price} x {trade.size}") + +quote = client.get_last_quote(ticker="AAPL") +print(f"Bid: ${quote.bid} x {quote.bid_size}") +print(f"Ask: ${quote.ask} x {quote.ask_size}") +``` + +## How FinAlly Uses the API + +The Massive poller runs as a background asyncio task, one batched call per interval regardless of ticker count: + +1. Collects the tracked ticker set (watchlist ∪ positions — see `MARKET_INTERFACE.md`) +2. Calls `get_snapshot_all("stocks", tickers)` in a thread (the client is synchronous) — one API call +3. Extracts `last_trade.price` and `last_trade.sip_timestamp` from each snapshot +4. Writes to the shared in-memory `PriceCache` +5. Sleeps for the poll interval, then repeats + +```python +import asyncio +from massive import RESTClient + +async def poll_massive(api_key: str, get_tickers, price_cache, interval: float = 15.0) -> None: + client = RESTClient(api_key=api_key) + + while True: + tickers = get_tickers() + if tickers: + snapshots = await asyncio.to_thread(client.get_snapshot_all, "stocks", tickers) + for snap in snapshots: + if snap.last_trade is None: + continue + price_cache.update( + ticker=snap.ticker, + price=snap.last_trade.price, + timestamp=snap.last_trade.sip_timestamp / 1_000_000_000, # ns -> seconds + ) + + await asyncio.sleep(interval) +``` + +## Error Handling + +The client raises exceptions for HTTP errors: +- **401**: invalid API key +- **403**: plan doesn't include the endpoint (this is the practical failure mode on the Basic tier for the snapshot endpoint — see below) +- **429**: rate limit exceeded (Basic tier: 5 req/min) +- **5xx**: server errors (client has built-in retry, 3 attempts by default) + +FinAlly's poller (`massive_client.py::_poll_once`) already wraps each poll cycle in a broad `try/except` and logs rather than crashing, so a `403`/`429`/network blip degrades to "stale prices until the next successful poll" rather than taking down the app. + +## Discrepancies & Gotchas Found vs. FinAlly's Current Code and Prior Docs + +Verified against the live `massive-com/client-python` GitHub repo (README, official examples, and the `TickerSnapshot`/`LastTrade` model source) as of this research pass: + +1. **Free tier likely can't use the snapshot endpoint at all.** Massive's docs state the "Full Market Snapshot" / all-tickers snapshot family is "Included in Stocks Starter, Developer, Advanced, and Business plans — Not included in Stocks Basic." If a user runs FinAlly with a free Massive key, `get_snapshot_all` will plausibly 403 rather than just being slow at 5 req/min. `get_previous_close_agg` *is* available on Basic. Worth deciding: either document that Massive real-data mode requires a paid (Starter+) key, or fall back to polling `get_previous_close_agg` per ticker on Basic keys (loses intraday movement). +2. **`get_snapshot_all` signature differs from what's in `backend/app/market/massive_client.py`.** The current code calls it as `client.get_snapshot_all(market_type=SnapshotMarketType.STOCKS, tickers=self._tickers)`, importing `SnapshotMarketType` from `massive.rest.models`. The verified official example calls it positionally as `client.get_snapshot_all("stocks", tickers)` — a plain string, not an enum — and no `SnapshotMarketType` symbol was found anywhere in the package's public model exports during this research. This import may simply not exist in `massive==2.2.0` and would raise `ImportError` at startup. **Recommend verifying this against the installed package directly** (`python -c "from massive.rest.models import SnapshotMarketType"`) since it wasn't installable in this research environment (no package-index network access) — flagging as unverified-but-suspicious rather than confirmed-broken. +3. **`LastTrade` has no `.timestamp` attribute.** `massive_client.py` reads `snap.last_trade.timestamp`. The verified model only exposes `sip_timestamp`, `participant_timestamp`, and `trf_timestamp` — all in **nanoseconds**, not milliseconds. The current code's ms→s conversion (`/ 1000.0`) would also be off by 1,000,000x if the field did exist at millisecond resolution; nanosecond timestamps need `/ 1_000_000_000`. This is a likely real bug, not just a docs staleness issue — worth a follow-up ticket against `massive_client.py` and `test_massive.py` (whose mocks currently model `last_trade.timestamp` in milliseconds). +4. **`day.previous_close` doesn't exist.** The prior archived `MASSIVE_API.md` and the interface sketch both reference `snap.day.previous_close` / `day.change_percent`. The real model has a separate `prev_day: Agg` field (previous session's full OHLCV) and top-level `todays_change_percent` — there's no `previous_close` key nested under `day`. +5. **Rate limit table simplification confirmed accurate in spirit**: 5 req/min free, unlimited (with a 100 req/s courtesy guideline) on all paid tiers — this part of the prior docs holds up. +6. **The `/v3/snapshot` "Unified Snapshot" endpoint** (`ticker.any_of=...`, up to 250 tickers, spans stocks/options/forex/crypto in one call) is newer and not what FinAlly currently uses or needs — the stocks-only `/v2/snapshot/locale/us/markets/stocks/tickers` endpoint remains supported and is simpler for this project's stocks-only scope. diff --git a/planning/PLAN.md b/planning/PLAN.md index bc1811b33..a56dd649c 100644 --- a/planning/PLAN.md +++ b/planning/PLAN.md @@ -22,13 +22,13 @@ The user runs a single Docker command (or a provided start script). A browser op ### What the User Can Do - **Watch prices stream** — prices flash green (uptick) or red (downtick) with subtle CSS animations that fade -- **View sparkline mini-charts** — price action beside each ticker in the watchlist, accumulated on the frontend from the SSE stream since page load (sparklines fill in progressively) +- **View sparkline mini-charts** — price action beside each ticker in the watchlist, pre-populated on load from server-side price history and then extended live from the SSE stream - **Click a ticker** to see a larger detailed chart in the main chart area -- **Buy and sell shares** — market orders only, instant fill at current price, no fees, no confirmation dialog +- **Buy and sell shares** — market orders only, fractional quantities supported, instant fill at current price, no fees, no confirmation dialog - **Monitor their portfolio** — a heatmap (treemap) showing positions sized by weight and colored by P&L, plus a P&L chart tracking total portfolio value over time - **View a positions table** — ticker, quantity, average cost, current price, unrealized P&L, % change - **Chat with the AI assistant** — ask about their portfolio, get analysis, and have the AI execute trades and manage the watchlist through natural language -- **Manage the watchlist** — add/remove tickers manually or via the AI chat +- **Manage the watchlist** — add/remove tickers manually or via the AI chat, up to 20 tickers ### Visual Design @@ -39,6 +39,7 @@ The user runs a single Docker command (or a provided start script). A browser op - **Responsive but desktop-first**: optimized for wide screens, functional on tablet ### Color Scheme + - Accent Yellow: `#ecad0a` - Blue Primary: `#209dd7` - Purple Secondary: `#753991` (submit buttons) @@ -71,14 +72,14 @@ The user runs a single Docker command (or a provided start script). A browser op ### Why These Choices -| Decision | Rationale | -|---|---| -| SSE over WebSockets | One-way push is all we need; simpler, no bidirectional complexity, universal browser support | -| Static Next.js export | Single origin, no CORS issues, one port, one container, simple deployment | -| SQLite over Postgres | No auth = no multi-user = no need for a database server; self-contained, zero config | -| Single Docker container | Students run one command; no docker-compose for production, no service orchestration | -| uv for Python | Fast, modern Python project management; reproducible lockfile; what students should learn | -| Market orders only | Eliminates order book, limit order logic, partial fills — dramatically simpler portfolio math | +| Decision | Rationale | +| ----------------------- | --------------------------------------------------------------------------------------------- | +| SSE over WebSockets | One-way push is all we need; simpler, no bidirectional complexity, universal browser support | +| Static Next.js export | Single origin, no CORS issues, one port, one container, simple deployment | +| SQLite over Postgres | No auth = no multi-user = no need for a database server; self-contained, zero config | +| Single Docker container | Students run one command; no docker-compose for production, no service orchestration | +| uv for Python | Fast, modern Python project management; reproducible lockfile; what students should learn | +| Market orders only | Eliminates order book, limit order logic, partial fills — dramatically simpler portfolio math | --- @@ -88,7 +89,7 @@ The user runs a single Docker command (or a provided start script). A browser op finally/ ├── frontend/ # Next.js TypeScript project (static export) ├── backend/ # FastAPI uv project (Python) -│ └── db/ # Schema definitions, seed data, migration logic +│ └── database/ # Schema definitions, seed data, migration logic ├── planning/ # Project-wide documentation for agents │ ├── PLAN.md # This document │ └── ... # Additional agent reference docs @@ -110,7 +111,7 @@ finally/ - **`frontend/`** is a self-contained Next.js project. It knows nothing about Python. It talks to the backend via `/api/*` endpoints and `/api/stream/*` SSE endpoints. Internal structure is up to the Frontend Engineer agent. - **`backend/`** is a self-contained uv project with its own `pyproject.toml`. It owns all server logic including database initialization, schema, seed data, API routes, SSE streaming, market data, and LLM integration. Internal structure is up to the Backend/Market Data agents. -- **`backend/db/`** contains schema SQL definitions and seed logic. The backend lazily initializes the database on first request — creating tables and seeding default data if the SQLite file doesn't exist or is empty. +- **`backend/database/`** contains schema SQL definitions and seed logic. Named distinctly from the top-level `db/` to avoid confusion between "the schema code" and "the runtime data file." The backend lazily initializes the database on first request — creating tables and seeding default data if the SQLite file doesn't exist or is empty. - **`db/`** at the top level is the runtime volume mount point. The SQLite file (`db/finally.db`) is created here by the backend and persists across container restarts via Docker volume. - **`planning/`** contains project-wide documentation, including this plan. All agents reference files here as the shared contract. - **`test/`** contains Playwright E2E tests and supporting infrastructure (e.g., `docker-compose.test.yml`). Unit tests live within `frontend/` and `backend/` respectively, following each framework's conventions. @@ -159,25 +160,30 @@ Both the simulator and the Massive client implement the same abstract interface. ### Massive API (Optional) - REST API polling (not WebSocket) — simpler, works on all tiers -- Polls for the union of all watched tickers on a configurable interval +- Polls the tracked ticker set (see "Tracked Tickers" below) in a **single batched call per interval** (`get_snapshot_all(tickers=[...])`) — call count depends only on poll frequency, not on how many tickers are tracked, so the watchlist cap below is not needed to stay under the rate limit - Free tier (5 calls/min): poll every 15 seconds - Paid tiers: poll every 2-15 seconds depending on tier - Parses REST response into the same format as the simulator -### Shared Price Cache +### Tracked Tickers + +The set of tickers passed to the market data source (`start()` / `add_ticker()` / `remove_ticker()`) is always **`watchlist ∪ positions.ticker`** for the current user, not just the watchlist. This guarantees a live price is always available for the positions table, portfolio heatmap, and total-value calculation even if a held ticker is removed from the watchlist. A ticker is only untracked once it is in neither the watchlist nor any open position. + +To keep this set (and the SSE payload / GBM correlation matrix) bounded, the **watchlist is capped at 20 tickers**; `POST /api/watchlist` returns `400` if the cap would be exceeded. Positions are not capped — a position always keeps its ticker tracked regardless of the watchlist cap. + +### Shared Price Cache & History -- A single background task (simulator or Massive poller) writes to an in-memory price cache -- The cache holds the latest price, previous price, and timestamp for each ticker -- SSE streams read from this cache and push updates to connected clients -- This architecture supports future multi-user scenarios without changes to the data layer +- A single background task (simulator or Massive poller) writes to an in-memory `PriceCache`, which holds, per ticker: the latest price, previous price, timestamp, and a bounded rolling history buffer (most recent 500 points) +- SSE streams and the history endpoint (`GET /api/market/history/{ticker}`, see §8) both read from this same cache — producers and consumers stay decoupled, and this supports future multi-user scenarios without changes to the data layer ### SSE Streaming - Endpoint: `GET /api/stream/prices` - Long-lived SSE connection; client uses native `EventSource` API -- Server pushes price updates for all tickers known to the system at a regular cadence (~500ms) — in the single-user model this is equivalent to the user's watchlist +- Server pushes price updates for the tracked ticker set (see "Tracked Tickers" above) at a regular cadence (~500ms) - Each SSE event contains ticker, price, previous price, timestamp, and change direction - Client handles reconnection automatically (EventSource has built-in retry) +- On connect, the frontend first calls `GET /api/market/history/{ticker}` for each visible ticker to pre-populate sparklines/charts, then appends subsequent points from the SSE stream --- @@ -196,11 +202,13 @@ The backend checks for the SQLite database on startup (or first request). If the All tables include a `user_id` column defaulting to `"default"`. This is hardcoded for now (single-user) but enables future multi-user support without schema migration. **users_profile** — User state (cash balance) + - `id` TEXT PRIMARY KEY (default: `"default"`) - `cash_balance` REAL (default: `10000.0`) - `created_at` TEXT (ISO timestamp) **watchlist** — Tickers the user is watching + - `id` TEXT PRIMARY KEY (UUID) - `user_id` TEXT (default: `"default"`) - `ticker` TEXT @@ -208,6 +216,7 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - UNIQUE constraint on `(user_id, ticker)` **positions** — Current holdings (one row per ticker per user) + - `id` TEXT PRIMARY KEY (UUID) - `user_id` TEXT (default: `"default"`) - `ticker` TEXT @@ -217,6 +226,7 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - UNIQUE constraint on `(user_id, ticker)` **trades** — Trade history (append-only log) + - `id` TEXT PRIMARY KEY (UUID) - `user_id` TEXT (default: `"default"`) - `ticker` TEXT @@ -226,17 +236,19 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - `executed_at` TEXT (ISO timestamp) **portfolio_snapshots** — Portfolio value over time (for P&L chart). Recorded every 30 seconds by a background task, and immediately after each trade execution. + - `id` TEXT PRIMARY KEY (UUID) - `user_id` TEXT (default: `"default"`) - `total_value` REAL - `recorded_at` TEXT (ISO timestamp) **chat_messages** — Conversation history with LLM + - `id` TEXT PRIMARY KEY (UUID) - `user_id` TEXT (default: `"default"`) - `role` TEXT (`"user"` or `"assistant"`) - `content` TEXT -- `actions` TEXT (JSON — trades executed, watchlist changes made; null for user messages) +- `actions` TEXT (JSON — trades executed, watchlist changes made; null for user messages). This is a denormalized snapshot for rendering the chat transcript only — `trades` and `watchlist` remain the source of truth for portfolio/watchlist state. - `created_at` TEXT (ISO timestamp) ### Default Seed Data @@ -249,33 +261,39 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod ## 8. API Endpoints ### Market Data -| Method | Path | Description | -|--------|------|-------------| -| GET | `/api/stream/prices` | SSE stream of live price updates | + +| Method | Path | Description | +| ------ | ----------------------------- | ------------------------------------------------------------------------ | +| GET | `/api/stream/prices` | SSE stream of live price updates | +| GET | `/api/market/history/{ticker}` | Recent price history for a ticker (from the in-memory rolling buffer), used to pre-populate charts/sparklines on load | ### Portfolio -| Method | Path | Description | -|--------|------|-------------| -| GET | `/api/portfolio` | Current positions, cash balance, total value, unrealized P&L | -| POST | `/api/portfolio/trade` | Execute a trade: `{ticker, quantity, side}` | -| GET | `/api/portfolio/history` | Portfolio value snapshots over time (for P&L chart) | + +| Method | Path | Description | +| ------ | ------------------------ | ------------------------------------------------------------ | +| GET | `/api/portfolio` | Current positions, cash balance, total value, unrealized P&L | +| POST | `/api/portfolio/trade` | Execute a trade: `{ticker, quantity, side}` | +| GET | `/api/portfolio/history` | Portfolio value snapshots over time (for P&L chart) | ### Watchlist -| Method | Path | Description | -|--------|------|-------------| -| GET | `/api/watchlist` | Current watchlist tickers with latest prices | -| POST | `/api/watchlist` | Add a ticker: `{ticker}` | -| DELETE | `/api/watchlist/{ticker}` | Remove a ticker | + +| Method | Path | Description | +| ------ | ------------------------- | -------------------------------------------- | +| GET | `/api/watchlist` | Current watchlist tickers with latest prices | +| POST | `/api/watchlist` | Add a ticker: `{ticker}`. `400` if the watchlist is already at the 20-ticker cap | +| DELETE | `/api/watchlist/{ticker}` | Remove a ticker | ### Chat -| Method | Path | Description | -|--------|------|-------------| -| POST | `/api/chat` | Send a message, receive complete JSON response (message + executed actions) | + +| Method | Path | Description | +| ------ | ----------- | --------------------------------------------------------------------------- | +| POST | `/api/chat` | Send a message, receive complete JSON response (message + executed actions) | ### System -| Method | Path | Description | -|--------|------|-------------| -| GET | `/api/health` | Health check (for Docker/deployment) | + +| Method | Path | Description | +| ------ | ------------- | ------------------------------------ | +| GET | `/api/health` | Health check (for Docker/deployment) | --- @@ -290,7 +308,7 @@ There is an OPENROUTER_API_KEY in the .env file in the project root. When the user sends a chat message, the backend: 1. Loads the user's current portfolio context (cash, positions with P&L, watchlist with live prices, total portfolio value) -2. Loads recent conversation history from the `chat_messages` table +2. Loads recent conversation history from the `chat_messages` table (last 20 messages, i.e. ~10 exchanges, to bound prompt size/latency/cost) 3. Constructs a prompt with a system message, portfolio context, conversation history, and the user's new message 4. Calls the LLM via LiteLLM → OpenRouter, requesting structured output, using the cerebras-inference skill 5. Parses the complete structured JSON response @@ -305,22 +323,19 @@ The LLM is instructed to respond with JSON matching this schema: ```json { "message": "Your conversational response to the user", - "trades": [ - {"ticker": "AAPL", "side": "buy", "quantity": 10} - ], - "watchlist_changes": [ - {"ticker": "PYPL", "action": "add"} - ] + "trades": [{ "ticker": "AAPL", "side": "buy", "quantity": 10 }], + "watchlist_changes": [{ "ticker": "PYPL", "action": "add" }] } ``` - `message` (required): The conversational text shown to the user -- `trades` (optional): Array of trades to auto-execute. Each trade goes through the same validation as manual trades (sufficient cash for buys, sufficient shares for sells) -- `watchlist_changes` (optional): Array of watchlist modifications +- `trades` (optional): Array of trades to auto-execute. `quantity` may be fractional (see §10). Each trade goes through the same validation as manual trades (sufficient cash for buys, sufficient shares for sells) +- `watchlist_changes` (optional): Array of watchlist modifications. `action` is `"add"` or `"remove"` — an `"add"` that would exceed the 20-ticker cap fails validation the same way an invalid trade does, and the error is surfaced to the LLM in the response ### Auto-Execution Trades specified by the LLM execute automatically — no confirmation dialog. This is a deliberate design choice: + - It's a simulated environment with fake money, so the stakes are zero - It creates an impressive, fluid demo experience - It demonstrates agentic AI capabilities — the core theme of the course @@ -330,6 +345,7 @@ If a trade fails validation (e.g., insufficient cash), the error is included in ### System Prompt Guidance The LLM should be prompted as "FinAlly, an AI trading assistant" with instructions to: + - Analyze portfolio composition, risk concentration, and P&L - Suggest trades with reasoning - Execute trades when the user asks or agrees @@ -340,6 +356,7 @@ The LLM should be prompted as "FinAlly, an AI trading assistant" with instructio ### LLM Mock Mode When `LLM_MOCK=true`, the backend returns deterministic mock responses instead of calling OpenRouter. This enables: + - Fast, free, reproducible E2E tests - Development without an API key - CI/CD pipelines @@ -352,12 +369,12 @@ When `LLM_MOCK=true`, the backend returns deterministic mock responses instead o The frontend is a single-page application with a dense, terminal-inspired layout. The specific component architecture and layout system is up to the Frontend Engineer, but the UI should include these elements: -- **Watchlist panel** — grid/table of watched tickers with: ticker symbol, current price (flashing green/red on change), daily change %, and a sparkline mini-chart (accumulated from SSE since page load) -- **Main chart area** — larger chart for the currently selected ticker, with at minimum price over time. Clicking a ticker in the watchlist selects it here. +- **Watchlist panel** — grid/table of watched tickers with: ticker symbol, current price (flashing green/red on change), daily change %, and a sparkline mini-chart (pre-populated from `GET /api/market/history/{ticker}` on load, then extended live from SSE) +- **Main chart area** — larger chart for the currently selected ticker, with at minimum price over time, seeded the same way (history endpoint on load + SSE thereafter). Clicking a ticker in the watchlist selects it here. - **Portfolio heatmap** — treemap visualization where each rectangle is a position, sized by portfolio weight, colored by P&L (green = profit, red = loss) - **P&L chart** — line chart showing total portfolio value over time, using data from `portfolio_snapshots` - **Positions table** — tabular view of all positions: ticker, quantity, avg cost, current price, unrealized P&L, % change -- **Trade bar** — simple input area: ticker field, quantity field, buy button, sell button. Market orders, instant fill. +- **Trade bar** — simple input area: ticker field, quantity field (accepts decimal input, rounded to 4 decimal places), buy button, sell button. Market orders, instant fill. - **AI chat panel** — docked/collapsible sidebar. Message input, scrolling conversation history, loading indicator while waiting for LLM response. Trade executions and watchlist changes shown inline as confirmations. - **Header** — portfolio total value (updating live), connection status indicator, cash balance @@ -404,12 +421,14 @@ The `db/` directory in the project root maps to `/app/db` in the container. The ### Start/Stop Scripts **`scripts/start_mac.sh`** (macOS/Linux): + - Builds the Docker image if not already built (or if `--build` flag passed) - Runs the container with the volume mount, port mapping, and `.env` file - Prints the URL to access the app - Optionally opens the browser **`scripts/stop_mac.sh`** (macOS/Linux): + - Stops and removes the running container - Does NOT remove the volume (data persists) @@ -428,12 +447,14 @@ The container is designed to deploy to AWS App Runner, Render, or any container ### Unit Tests (within `frontend/` and `backend/`) **Backend (pytest)**: + - Market data: simulator generates valid prices, GBM math is correct, Massive API response parsing works, both implementations conform to the abstract interface - Portfolio: trade execution logic, P&L calculations, edge cases (selling more than owned, buying with insufficient cash, selling at a loss) - LLM: structured output parsing handles all valid schemas, graceful handling of malformed responses, trade validation within chat flow - API routes: correct status codes, response shapes, error handling **Frontend (React Testing Library or similar)**: + - Component rendering with mock data - Price flash animation triggers correctly on price changes - Watchlist CRUD operations @@ -447,6 +468,7 @@ The container is designed to deploy to AWS App Runner, Render, or any container **Environment**: Tests run with `LLM_MOCK=true` by default for speed and determinism. **Key Scenarios**: + - Fresh start: default watchlist appears, $10k balance shown, prices are streaming - Add and remove a ticker from the watchlist - Buy shares: cash decreases, position appears, portfolio updates @@ -454,3 +476,22 @@ The container is designed to deploy to AWS App Runner, Render, or any container - Portfolio visualization: heatmap renders with correct colors, P&L chart has data points - AI chat (mocked): send a message, receive a response, trade execution appears inline - SSE resilience: disconnect and verify reconnection + +--- + +## 13. Doc Review Changelog (2026-08-24) + +A doc review raised 7 questions plus feedback on naming/duplication/repetition (full original notes in git history). All were resolved and folded into the sections above: + +| # | Resolution | Where | +|---|---|---| +| Tracked tickers vs. watchlist | Market data always tracks `watchlist ∪ positions.ticker`, not just the watchlist | §6 "Tracked Tickers" | +| No historical price data | Added an in-memory rolling history buffer (500 points/ticker) in `PriceCache` + `GET /api/market/history/{ticker}` to seed charts on load | §6, §8, §10 | +| Massive rate limit vs. ticker count | Confirmed via `massive_client.py`: polling is one batched call per interval regardless of ticker count, so it doesn't constrain the watchlist cap | §6 "Massive API" | +| Watchlist size limit | Capped at 20 tickers; `POST /api/watchlist` returns `400` past the cap. Positions are never capped. | §2, §6, §8 | +| Unbounded chat history sent to LLM | Bounded to the last 20 messages (~10 exchanges) | §9 | +| `watchlist_changes.action` values | Documented explicitly as `"add"` \| `"remove"` | §9 | +| Fractional quantity input | Trade bar accepts decimal quantities, rounded to 4 decimal places | §10 | +| `backend/db/` vs. top-level `db/` naming collision | Renamed the schema-code directory to `backend/database/`; top-level `db/` stays the runtime volume mount | §4 | +| `chat_messages.actions` duplicating `trades`/`watchlist` | Documented as a denormalized display snapshot only — `trades`/`watchlist` remain the source of truth | §7 | +| Repetition between "Shared Price Cache" and "SSE Streaming" | Merged into "Shared Price Cache & History" + a tightened "SSE Streaming" | §6 | diff --git a/planning/REVIEW.md b/planning/REVIEW.md new file mode 100644 index 000000000..9c013042f --- /dev/null +++ b/planning/REVIEW.md @@ -0,0 +1,25 @@ +# PLAN.md Review + +## Review — 2026-08-24 + +The plan has a clear product boundary and resolves the earlier watchlist, price-history, and chat-context gaps. The remaining items are chiefly contracts that need to be explicit before frontend and backend work proceeds independently. + +### Questions and clarifications + +1. **Define complete request and response schemas.** The endpoint table names the routes but does not specify response payloads, required-field validation, success status codes, or an error envelope. In particular, document the shapes for portfolio positions/P&L, watchlist prices and daily change, history points, trade results, chat actions and per-action failures, and SSE events. This is the most important shared contract for parallel implementation. +2. **Specify trade atomicity and price semantics.** State that a trade validates and updates cash, position, trade history, and immediate portfolio snapshot in one SQLite transaction. Also define the authoritative fill price: the latest `PriceCache` price at the transaction's start, including the behavior when a ticker has no current price or is being added to tracking. +3. **Define numeric and ticker validation centrally.** Document canonical ticker normalization (for example, uppercase and trimming), supported symbols, maximum ticker length, finite positive quantity requirements, four-decimal rounding rule, currency rounding, and the treatment of a position whose remaining quantity is near zero. These rules must be shared by manual trades, chat actions, and watchlist routes. +4. **Resolve market-data startup and failure behavior.** Specify how much history the simulator pre-populates before the first client request; whether the API returns an empty/partial history while a newly tracked ticker is warming up; and how Massive failures, invalid symbols, stale prices, and rate-limit responses appear in the API/UI. A stale-price timestamp/status is needed so the user is not shown a live-looking price after a failed poll. +5. **Define SSE wire and connection behavior.** State the event name and JSON schema, heartbeat interval, retry interval, ordering/deduplication expectations, and whether each event is a single ticker update or a batch. The connection dot also needs exact transitions for initial connection, a healthy but stale stream, errors, and reconnection. +6. **Clarify daily-change source.** The UI requires daily change %, but the cache only defines latest and previous tick prices. Decide whether this is tick-over-tick change, a session-open/previous-close change, or unavailable for the simulator. Labeling tick movement as daily change would be misleading. +7. **Resolve the database-volume wording.** Section 11 calls `finally-data` a named Docker volume, while the following paragraph says the project-root `db/` directory maps to `/app/db`; the shown `docker run` command uses the named volume. Choose one default and update the Dockerfile, Compose file, and scripts consistently. If persistence across application updates is required, specify the documented reset procedure. +8. **Define initialization and snapshot retention.** Clarify whether startup initialization happens only in the FastAPI lifespan handler or may happen on requests, how concurrent requests avoid duplicate seeding, whether an initial portfolio snapshot is inserted, and how/when portfolio snapshots are pruned. The 30-second recorder should also define its behavior when price data is unavailable. +9. **Make mock-chat behavior contractual.** List deterministic mock inputs and resulting actions, including at least one buy, sell, invalid action, and no-action response. E2E scenarios can then assert stable outcomes rather than depend on implementation-specific keyword matching. +10. **Document static-export routing.** Define how FastAPI serves static assets and falls back to the exported SPA entry point without intercepting `/api/*`. This avoids deep-link 404s and route-order ambiguity in the single-container deployment. + +### Opportunities to simplify + +- Replace repeated prose references to tracked tickers with a single invariant: `tracked_tickers = watchlist ∪ open_position_tickers`; link to it from the market-data, trade, and watchlist sections. +- Consolidate the endpoint table and frontend data requirements into one compact API-contract section with schemas and error rules. It removes duplication while making the interfaces implementable. +- Treat the future multi-user fields as schema preparation only. Because every request currently uses `"default"`, avoid claiming that the in-memory `PriceCache` already supports multi-user behavior; it is globally keyed and market data is shared. A short future-auth note is clearer. +- Choose one charting library rather than leaving two alternatives. This reduces dependency, rendering, and test uncertainty for the frontend implementation. diff --git a/planning/review-claude.md b/planning/review-claude.md new file mode 100644 index 000000000..501d5917e --- /dev/null +++ b/planning/review-claude.md @@ -0,0 +1,71 @@ +# PLAN.md Diff Review — Claude + +**Scope**: `git diff planning/PLAN.md` (93 insertions / 52 deletions), reviewed against the rest of the document, `planning/MARKET_DATA_SUMMARY.md`, the already-built `backend/app/market/*` code, and the two sibling untracked reviews (`planning/REVIEW.md`, `planning/DSeekREVIEW.md`), which both turn out to already be reviews of this same post-diff text. + +## Verdict + +The diff is a coherent, well-targeted response to a prior review round — it resolves seven real ambiguities (tracked-tickers vs. watchlist, sparkline seeding, fractional quantities, watchlist cap, chat-history bound, `watchlist_changes.action` enum, the `backend/db`/`db` naming collision) and the §13 changelog's description of *what text changed* is accurate. The main problem is not the prose itself but a **documentation/implementation mismatch that the diff creates**: it writes several of these resolutions in the past tense ("Added...", "resolved") inside a project where a sibling doc and `CLAUDE.md` already declare the market-data component "complete" — but the corresponding code does not contain the feature. A reader (especially a downstream frontend agent) can reasonably conclude these are shipped, and they are not. + +--- + +## Findings (priority order) + +### 1. HIGH — §13 documents the history-buffer feature as resolved; it does not exist in the already-"completed" market-data code + +Diff text (new): +> A single background task (simulator or Massive poller) writes to an in-memory `PriceCache`, which holds, per ticker: the latest price, previous price, timestamp, and **a bounded rolling history buffer (most recent 500 points)** +> ... +> | No historical price data | Added an in-memory rolling history buffer (500 points/ticker) in `PriceCache` + `GET /api/market/history/{ticker}` to seed charts on load | §6, §8, §10 | + +I checked `backend/app/market/cache.py` directly: `PriceCache` stores only `self._prices: dict[str, PriceUpdate]` — the latest update per ticker, nothing else. There is no history array, no 500-point buffer, no pruning logic, and no `history` route anywhere under `backend/app/market/` (the only stream route is `/api/stream/prices`). `planning/MARKET_DATA_SUMMARY.md` states the market-data subsystem is "Complete, tested, reviewed, all issues resolved," and root `CLAUDE.md` says the market-data component "has been completed." Put together, a downstream agent reading §13 plus those two documents has no signal that `GET /api/market/history/{ticker}` is still unbuilt — the changelog phrasing ("Added...", "All were resolved") reads as a statement about the codebase, not just the spec text. + +**Fix**: Reword the §13 entry to make clear this is a *specification* resolution, not a shipped one — e.g. "Resolved at the plan level; **not yet implemented** — `PriceCache` currently holds only the latest price per ticker, so `/api/market/history/{ticker}` and the 500-point buffer are outstanding backend work." Also worth a one-line addendum to `MARKET_DATA_SUMMARY.md` noting the buffer is a planned extension to the "complete" subsystem, so the two docs don't contradict each other. (This mirrors `planning/DSeekREVIEW.md` finding A1, which verified the same gap independently against the code.) + +### 2. MEDIUM — New "4 decimal places" rounding rule is stated only for the frontend trade bar, not as a server-side validation rule for API/LLM-issued trades + +Diff text (new, §2 / §9 / §10): +> **Buy and sell shares** — market orders only, **fractional quantities supported**... +> `trades` (optional): Array of trades to auto-execute. **`quantity` may be fractional (see §10)**. Each trade goes through the same validation as manual trades... +> **Trade bar** — ... quantity field (**accepts decimal input, rounded to 4 decimal places**), buy button, sell button. + +Fractional quantities are a brand-new concept introduced by this diff (the schema's `quantity REAL` was already fractional-capable, but the plan previously implied whole-share trading only). The 4-decimal-place rounding rule is stated exclusively in §10, which is UI-only ("Trade bar"). §9 explicitly points an LLM-issued trade's fractional-quantity rule back to §10 ("see §10"), but the LLM never goes through the trade-bar UI — it calls the trade path directly. §8's `POST /api/portfolio/trade` description (`{ticker, quantity, side}`) still has no validation rule at all. This leaves it undefined whether 4-decimal rounding is: +- a UI input mask only (so the API accepts arbitrary precision), or +- a server-side invariant enforced on every trade regardless of source (manual API call, chat/LLM trade). + +Given the LLM can supply quantities like `3.33333`, and the API can be called directly by a test or future client, this is a real fork point for the backend implementer. + +**Fix**: State the 4-decimal rounding (and "must be finite, positive" — already flagged in REVIEW.md #3) as a server-side validation rule in §7 or §8, applying uniformly to manual trades and LLM-issued trades, and have §10's trade-bar bullet reference it rather than being the sole source of the rule. + +### 3. MEDIUM — Diff touches the Massive/SSE sections but leaves the adjacent per-event SSE description effectively unspecified/incorrect + +The diff rewrites "Shared Price Cache" → "Shared Price Cache & History" and rewords the SSE bullet to route through the new "Tracked Tickers" section, but it leaves this sentence essentially untouched in spirit: +> Each SSE event contains ticker, price, previous price, timestamp, and change direction + +This reads as one event per ticker. The already-built `backend/app/market/stream.py` instead sends one **batched** JSON object per send (`{"AAPL": {...}, "GOOGL": {...}}`), gated on a version counter, with no `event:` name and no heartbeat. Since the diff was already touching this exact section (adding "Tracked Tickers" and rewiring the history-preload flow through it), it was a natural place to also correct the per-event framing to match the implemented wire format — leaving it as-is means the frontend contract is still wrong in a section this diff specifically edited. (Independently confirmed against code in `planning/DSeekREVIEW.md` A3.) + +**Fix**: While touching §6 again for finding #1, also update the SSE bullet to describe the batched-payload shape, version-gated sends, and absence of heartbeat — or explicitly flag it as a known frontend-contract gap if you don't want to change behavior yet. + +### 4. LOW — §13's framing invites over-reading "resolved" as "implemented" more broadly + +Beyond the history-buffer row specifically (#1), the section's closing sentence — "All were resolved and folded into the sections above" — is accurate for a documentation review (the prior round's *questions* were resolved by editing text) but is easy to misread project-wide, especially next to `MARKET_DATA_SUMMARY.md`'s "all issues resolved" for the code. Two rows in the table (Tracked Tickers, Massive rate-limit) *are* already true of the shipped code — verified: `massive_client.py` does batch all tracked tickers into one `_fetch_snapshots` call — but the history-buffer row is not, and a reader has no way to tell the difference without cross-checking source. + +**Fix**: Either split the table into "spec-level" vs. "implemented" resolutions, or add a one-line caveat above the table: "These are plan-text resolutions; check `backend/app/market/` for current implementation status of any given item." + +### 5. LOW — Minor, adjacent inconsistency not fixed while the section was open + +The diff added a new Massive bullet about batched polling but left the pre-existing, unrelated bullet in the same section reading "Paid tiers: poll every 2-15 seconds depending on tier," while `massive_client.py`'s own docstring says paid tiers poll "every 2-5s." Not introduced by this diff, but since the diff was already editing this exact bullet block, it was a low-cost opportunity to reconcile. (Also flagged in `planning/DSeekREVIEW.md` B8.) + +### 6. Informational — no consistency problems found in the core additions + +Checked and consistent: +- The 20-ticker watchlist cap is stated uniformly across §2, §6, and §8, and doesn't conflict with the 10-ticker default seed list in §7. +- "Tracked tickers = watchlist ∪ positions.ticker" is introduced once in §6 and referenced (not restated) from §8's SSE description — no duplication problem. +- The `chat_messages.actions` denormalization note (§7) doesn't contradict the trades/watchlist tables elsewhere; it explicitly subordinates itself to them. +- Directory rename `backend/db/` → `backend/database/` (§4): checked the actual repo — no `backend/db` or `backend/database` directory has been created yet (market data lives entirely in `backend/app/market/`), so this rename isn't undoing already-completed work; it's a clean, uncontested fix for the naming collision the changelog claims to resolve. +- §13's "7 questions plus feedback on naming/duplication/repetition" count checks out against the 10 table rows (7 substantive questions + 3 naming/duplication/repetition items), so the summary sentence isn't miscounted. + +--- + +## Note on `planning/REVIEW.md` and `planning/DSeekREVIEW.md` + +Both are reviews of the **current (post-diff) PLAN.md**, not of a prior version — `DSeekREVIEW.md` says so explicitly ("includes the §13 doc-review changelog"), and `REVIEW.md`'s framing ("resolves the earlier watchlist, price-history, and chat-context gaps... remaining items are chiefly contracts") is consistent with that too. So they aren't independent context to compare against the diff — they're downstream evaluations of it, and this review's finding #1 corroborates (and was cross-checked against source for) `DSeekREVIEW.md`'s A1/A3, the two highest-severity items there. Their broader findings (full request/response schemas, trade atomicity, numeric validation, static-export routing fallback, etc.) are real but are pre-existing gaps in PLAN.md, not things this specific diff introduced or worsened — they're appropriately out of scope for a diff-focused review and worth folding into a future revision. diff --git a/planning/review-deepseek.md b/planning/review-deepseek.md new file mode 100644 index 000000000..00a347160 --- /dev/null +++ b/planning/review-deepseek.md @@ -0,0 +1,72 @@ +# PLAN.md Diff Review — DeepSeek (Codewhale) + +**Scope**: all changes since last commit (`git diff HEAD`), i.e. `planning/PLAN.md` (modified) plus the untracked files `planning/REVIEW.md`, `planning/review-claude.md`, `.claude/agents/*`, `.claude/commands/*`, `.codewhale/*`. Findings were cross-checked against the already-built `backend/app/market/*` code. + +## Verdict + +The `PLAN.md` diff is a well-targeted follow-up to a prior doc-review round. It resolves real ambiguities — tracked tickers vs. watchlist, sparkline seeding, the watchlist cap, chat-history bound, the `watchlist_changes.action` enum, the `backend/db` vs `db` naming collision — and the §13 changelog accurately describes *what text changed*. The substantive problem is a **spec-vs-implementation mismatch**: the diff writes several resolutions in past tense ("Added…", "All were resolved") inside a plan whose sibling docs already declare the market-data component "complete", but the corresponding code does not contain the feature. A downstream agent can reasonably read these as shipped, and they are not. + +## Findings (priority order) + +### 1. HIGH — History buffer + `/api/market/history/{ticker}` documented as resolved, but not implemented + +New text (§6, §8, §13): +> `PriceCache`, which holds, per ticker: … and **a bounded rolling history buffer (most recent 500 points)** +> | No historical price data | Added an in-memory rolling history buffer (500 points/ticker) in `PriceCache` + `GET /api/market/history/{ticker}` … | + +Verified against `backend/app/market/cache.py`: `PriceCache` stores only `self._prices: dict[str, PriceUpdate]` — the latest update per ticker, with no history array, no 500-point bound, no pruning. A grep across `backend/app/market/` finds **no `/history` route**; the only route is `@router.get("/prices")` under `/api/stream`. `planning/MARKET_DATA_SUMMARY.md` calls the market-data subsystem "Complete, tested, reviewed, all issues resolved," so nothing signals to a frontend agent that `GET /api/market/history/{ticker}` and the buffer are still outstanding backend work. + +**Fix**: reword §13 so it is unambiguous that this is a *plan-level* resolution, not a shipped one — e.g. "Resolved in the spec; **not yet implemented** — `PriceCache` currently holds only the latest price per ticker, so the 500-point buffer and `/api/market/history/{ticker}` are outstanding backend work." Add the same one-line caveat to `MARKET_DATA_SUMMARY.md`. + +### 2. MEDIUM — "4 decimal places" fractional-quantity rule is stated only for the UI, not as a server invariant + +Fractional quantities are introduced by this diff. The 4-dp rule appears solely in §10 (the "Trade bar" bullet, UI-only), and §9 explicitly points an LLM-issued trade's rule back to §10 ("see §10"). But the LLM and any direct API caller never pass through the trade-bar UI, and `POST /api/portfolio/trade` (§8) still specifies no validation rule. It is therefore undefined whether 4-dp rounding is (a) a UI input mask only, or (b) a server-side invariant on every trade regardless of source (manual, chat/LLM, test). This is a real fork for the backend implementer. + +**Fix**: state 4-dp rounding (plus "finite, positive quantity") as a server-side validation rule in §7/§8, applied uniformly to manual and LLM-issued trades, and have §10 reference it rather than be the sole source. + +### 3. MEDIUM — SSE wire description remains wrong in a section the diff edited + +The diff rewrote the surrounding bullets ("Shared Price Cache & History", "Tracked Tickers") but left this sentence effectively intact: +> Each SSE event contains ticker, price, previous price, timestamp, and change direction + +This reads as one event per ticker. The implemented `backend/app/market/stream.py` instead yields one **batched** JSON object per send (`data: {"AAPL": {...}, "GOOGL": {...}}`), gated on a version counter, with no `event:` name and no heartbeat (only a `retry: 1000` directive). Since the diff was already editing this exact section, correcting the per-event framing to match the shipped wire format was in-scope; leaving it means the frontend contract is still wrong in a section this diff specifically touched. + +**Fix**: update the SSE bullet to describe the batched-payload shape, version-gated sends, and absence of heartbeat — or flag it explicitly as a known frontend-contract gap. + +### 4. MEDIUM — "daily change %" is required by the UI but not producible from the data model + +§10 requires "daily change %" in the watchlist panel and positions table, but `PriceUpdate`/`PriceCache` only stores `price` and `previous_price` (tick-over-tick). `PriceUpdate.change_percent` computes change from the *previous tick*, not from a session-open or previous-close baseline. Labeling tick movement "daily change" would be misleading, and the plan never resolves the source. This predates the diff but the diff touched the price-cache and UI bullets without closing it. + +**Fix**: define daily-change semantics explicitly (previous close / session open / tick-over-tick), or drop the "daily" qualifier and relabel as "change %". + +### 5. LOW — §13 "resolved" framing invites over-reading more broadly + +The closing line "All were resolved and folded into the sections above" is accurate for a *doc* review but easy to misread project-wide. Two rows (Tracked Tickers, Massive rate-limit) are already true of the code — verified: `massive_client.py` batches all tracked tickers into one `get_snapshot_all(tickers=self._tickers)` call — but the history-buffer row is not, and a reader cannot tell the difference without cross-checking source. + +**Fix**: split the table into "spec-level" vs "implemented" resolutions, or add a one-line caveat above it ("plan-text resolutions; check `backend/app/market/` for implementation status"). + +### 6. LOW — "full original notes in git history" is inaccurate + +§13 says the review's "full original notes" are "in git history", but the review artifacts (`planning/REVIEW.md`, `planning/review-claude.md`) are untracked working-tree files, not committed history. Minor, but it overstates the audit trail a reader can actually `git log` for. + +### 7. LOW — Adjacent Massive polling interval left inconsistent while the section was open + +The diff added a new Massive bullet about batched polling but left the neighboring bullet "Paid tiers: poll every 2-15 seconds depending on tier", whereas `massive_client.py`'s own docstring says paid tiers poll "every 2-5s". Not introduced by this diff, but a low-cost reconciliation since the bullet block was already being edited. + +### 8. LOW — Untracked review/agent scaffolding has cosmetic defects + +- `.claude/agents/codewhale-reviwer.md`: filename/`name` typo ("reviwer"), "write you feedback", and a self-contradictory instruction set — it tells the agent to run `codewhale execute "…"` and then says "DO NOT RUN THE REVIEW YOURSELF." +- `.claude/commands/doc-review.md`: `$ARGUEENTS` (typo; shell variables are case-sensitive, so this would expand empty). +- `planning/review-claude.md` references `planning/DSeekREVIEW.md`, which does not exist in the working tree (only `REVIEW.md` and `review-claude.md` are present) — a dangling cross-reference. + +## Consistency notes (verified, no problem found) + +- The 20-ticker watchlist cap is stated consistently across §2, §6, §8, and does not conflict with the 10-ticker default seed list in §7. +- "Tracked tickers = watchlist ∪ positions.ticker" is introduced once in §6 and referenced (not restated) from §8 — no duplication problem. +- The `chat_messages.actions` denormalization note (§7) explicitly subordinates itself to `trades`/`watchlist`; no contradiction. +- `backend/db/` → `backend/database/` rename (§4): the repo currently has neither directory (`backend/app/market/` holds market data), so this is a clean, uncontested fix for the naming collision, not a rollback of completed work. +- §13's "7 questions plus feedback on naming/duplication/repetition" count matches the 10 table rows (7 substantive + 3 naming/duplication/repetition). + +## Note on sibling reviews + +`planning/REVIEW.md` and `planning/review-claude.md` are both reviews of the *current post-diff* `PLAN.md`, not of a prior version. Their highest-severity items (spec-vs-code mismatch for the history buffer, SSE wire-format gap) were re-verified here against source and corroborate findings 1 and 3. Their remaining items (full request/response schemas, trade atomicity, numeric/ticker validation, static-export routing, mock-chat contract, snapshot retention) are real but pre-existing gaps in `PLAN.md`, not introduced or worsened by this diff — appropriately out of scope for a diff-focused review and worth folding into a future revision. From 3ba6ea79e508cfc3502a3d42d35ab5741efa2c2e Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 05:42:30 +0000 Subject: [PATCH 2/2] Add detailed market data backend design doc Consolidates the unified MarketDataSource interface, GBM simulator, and Massive API client into one implementation-ready design, verified against the current backend/app/market/ code. Also designs the history-buffer/GET /api/market/history/{ticker} feature that PLAN.md specifies but the codebase doesn't implement yet, clearly separating what's shipped from what's proposed. --- planning/MARKET_DATA_DESIGN.md | 1134 ++++++++++++++++++++++++++++++++ 1 file changed, 1134 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..2023458fc --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,1134 @@ +# Market Data Backend — Detailed Design + +Implementation-ready design for the FinAlly market data subsystem: the unified `MarketDataSource` interface, the in-memory `PriceCache`, the GBM simulator, the Massive API client, SSE streaming, and the still-outstanding history buffer / `GET /api/market/history/{ticker}` endpoint. + +**Sources consulted**: `planning/PLAN.md` §6–§8, the current per-component docs (`planning/MARKET_INTERFACE.md`, `planning/MARKET_SIMULATOR.md`, `planning/MASSIVE_API.md`), the doc-review passes (`planning/review-claude.md`, `planning/review-deepseek.md`, `planning/REVIEW.md`), the archived first-pass design and code review (`planning/archive/MARKET_DATA_DESIGN.md`, `planning/archive/MARKET_DATA_REVIEW.md`), and the actual code under `backend/app/market/` and `backend/tests/market/` as of 2026-08-25. + +## Status legend + +Every section below is tagged so this document doesn't repeat the spec/implementation mismatch the review round flagged in the archived design: + +- ✅ **Implemented** — matches `backend/app/market/*` today; snippets are verbatim from source. +- 🚧 **Proposed** — designed here, not yet in the codebase. Needed to close a gap between `planning/PLAN.md` and the shipped code. + +--- + +## Table of Contents + +1. [Architecture Overview](#1-architecture-overview) +2. [File Structure](#2-file-structure) +3. [Data Model — `models.py`](#3-data-model--modelspy) ✅ +4. [Price Cache — `cache.py`](#4-price-cache--cachepy) ✅ + 🚧 history buffer +5. [Abstract Interface — `interface.py`](#5-abstract-interface--interfacepy) ✅ +6. [Seed Prices & Ticker Parameters — `seed_prices.py`](#6-seed-prices--ticker-parameters) ✅ +7. [GBM Simulator — `simulator.py`](#7-gbm-simulator--simulatorpy) ✅ +8. [Massive API Client — `massive_client.py`](#8-massive-api-client--massive_clientpy) ✅ +9. [Factory — `factory.py`](#9-factory--factorypy) ✅ +10. [SSE Streaming Endpoint — `stream.py`](#10-sse-streaming-endpoint--streampy) ✅ +11. [History Endpoint — `GET /api/market/history/{ticker}`](#11-history-endpoint) 🚧 +12. [FastAPI Lifecycle Integration](#12-fastapi-lifecycle-integration) 🚧 +13. [Watchlist Coordination](#13-watchlist-coordination) 🚧 +14. [Daily-Change-% Semantics](#14-daily-change--semantics) +15. [Testing Strategy](#15-testing-strategy) +16. [Error Handling & Edge Cases](#16-error-handling--edge-cases) +17. [Configuration Summary](#17-configuration-summary) + +--- + +## 1. Architecture Overview + +``` + ┌──────────────────────────┐ +watchlist ∪ │ MarketDataSource (ABC) │ +positions.ticker │ ┌──────────────────────┐ │ + ──────────────┼─▶│ SimulatorDataSource │ │ (GBM math engine) + │ └──────────────────────┘ │ + │ ┌──────────────────────┐ │ + │ │ MassiveDataSource │ │ (REST poller) + │ └──────────────────────┘ │ + └────────────┬─────────────┘ + │ writes + ▼ + ┌──────────────────┐ + │ PriceCache │ (thread-safe, shared) + │ latest + history │ + └────────┬──────────┘ + reads │ reads + ┌───────────────┴───────────────┐ + ▼ ▼ + GET /api/stream/prices (SSE) GET /api/market/history/{ticker} + │ │ + ▼ ▼ + Frontend: live ticks Frontend: chart/sparkline seed +``` + +One interface, two implementations, selected once at startup by `create_market_data_source()` based on `MASSIVE_API_KEY`. Everything downstream — SSE, the history endpoint, portfolio valuation, trade execution — talks only to `PriceCache`. Neither data source is ever called directly outside `backend/app/market/`. + +--- + +## 2. File Structure + +``` +backend/ + app/ + market/ + __init__.py # Public API re-exports ✅ + models.py # PriceUpdate dataclass ✅ + cache.py # PriceCache (latest price + history buffer) ✅ (history 🚧) + interface.py # MarketDataSource ABC ✅ + seed_prices.py # SEED_PRICES, TICKER_PARAMS, correlations ✅ + simulator.py # GBMSimulator + SimulatorDataSource ✅ + massive_client.py # MassiveDataSource ✅ + factory.py # create_market_data_source() ✅ + stream.py # SSE endpoint factory ✅ + history.py # GET /api/market/history router 🚧 new file + main.py # FastAPI app, lifespan wiring 🚧 doesn't exist yet + tests/ + market/ + test_models.py test_cache.py test_factory.py ✅ + test_simulator.py test_simulator_source.py test_massive.py ✅ + test_history.py 🚧 new file +``` + +`backend/app/main.py` does not exist yet — nothing under `backend/app/market/` is wired into a running FastAPI app today. §12 below is the missing glue. + +--- + +## 3. Data Model — `models.py` + +✅ Implemented exactly as follows. `PriceUpdate` is the only type that leaves the market data layer — SSE payloads, history responses, and (later) portfolio valuation all consume it via `to_dict()`. + +```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: + if self.price > self.previous_price: + return "up" + elif self.price < self.previous_price: + return "down" + return "flat" + + def to_dict(self) -> dict: + 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, + } +``` + +- `frozen=True, slots=True`: value objects created many times per second, safe to hand across async tasks / threads without copying, and cheap to allocate. +- `change` / `change_percent` / `direction` are **tick-over-tick** (this update vs. the immediately prior one), not session-relative — see §14 for why that matters and what to do about it. + +--- + +## 4. Price Cache — `cache.py` + +### 4.1 Current implementation ✅ + +```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.""" + + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._lock = Lock() + self._version: int = 0 # bumped on every update + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + 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]: + 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: + return self._version + + def __len__(self) -> int: + with self._lock: + return len(self._prices) + + def __contains__(self, ticker: str) -> bool: + with self._lock: + return ticker in self._prices +``` + +**Why `threading.Lock`, not `asyncio.Lock`**: `MassiveDataSource` runs the synchronous Massive SDK inside `asyncio.to_thread`, so writes can originate from a real OS thread, not just the event loop. `asyncio.Lock` only synchronizes coroutines on one loop and would not protect against that. `threading.Lock` protects both. + +**Why a version counter**: the SSE loop polls the cache every ~500ms regardless of the underlying source's cadence (2 ticks/sec for the simulator, one poll per 15s for Massive). Without `version`, it would re-serialize and re-send the full price map on every tick even when Massive hasn't produced anything new. `version` lets it skip the no-op ticks cheaply — an `int` compare instead of a dict diff. + +### 4.2 Proposed extension: rolling history buffer 🚧 + +`planning/PLAN.md` §6 specifies that `PriceCache` also holds "a bounded rolling history buffer (most recent 500 points)" per ticker, feeding both the SSE stream and `GET /api/market/history/{ticker}`. **This does not exist in the code today** — `PriceCache` stores only the single latest `PriceUpdate` per ticker. This is the one concrete gap blocking sparkline/chart pre-population (`planning/PLAN.md` §2, §10). + +The fix is additive, not a rewrite — a bounded deque alongside the existing single-value store: + +```python +from __future__ import annotations + +import time +from collections import deque +from threading import Lock + +from .models import PriceUpdate + + +class PriceCache: + """Thread-safe in-memory cache of the latest price *and* a bounded + rolling history per ticker. + """ + + def __init__(self, history_size: int = 500) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._history: dict[str, deque[PriceUpdate]] = {} + self._history_size = history_size + self._lock = Lock() + self._version: int = 0 + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + 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._history.setdefault(ticker, deque(maxlen=self._history_size)).append(update) + self._version += 1 + return update + + def get_history(self, ticker: str) -> list[PriceUpdate]: + """Oldest-first list of up to `history_size` recent updates for a ticker.""" + with self._lock: + return list(self._history.get(ticker, ())) + + def remove(self, ticker: str) -> None: + with self._lock: + self._prices.pop(ticker, None) + self._history.pop(ticker, None) + + # get, get_all, get_price, version, __len__, __contains__ unchanged from §4.1 +``` + +Design notes: + +- **`deque(maxlen=...)`** gives O(1) amortized bounded appends for free — no manual `list[-500:]` slicing on every tick, and memory per ticker is capped regardless of how long the process runs. +- **Coverage math**: at the simulator's 500ms cadence, 500 points ≈ 4 minutes of history — plenty for a sparkline, and the main chart is expected to extend live from SSE after this seed (`planning/PLAN.md` §6, §10), not to hold a full trading day. At Massive's 15s poll interval, 500 points ≈ 2 hours. +- **`remove()`** must drop both `_prices[ticker]` and `_history[ticker]` — a ticker fully un-tracked (removed from both watchlist and positions, per §13) should not leak its buffer. +- **`history_size` is a constructor parameter, not a hardcoded `500`**, so tests can use a small buffer (e.g. `PriceCache(history_size=3)`) to assert eviction behavior without pushing 500 updates. +- No new locking primitive — reuses the existing `threading.Lock`, so the thread-safety story from §4.1 is unchanged. + +--- + +## 5. Abstract Interface — `interface.py` + +✅ Implemented exactly as follows. No `get_price()` method by design — this is a **push** interface (source → cache on its own schedule); callers read prices from `PriceCache`, never block on a network round trip or a simulation step through this interface. + +```python +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class MarketDataSource(ABC): + """Contract for market data providers. + + Implementations push price updates into a shared PriceCache on their own + schedule. Downstream code never calls the data source directly for prices — + it reads from the cache. + + Lifecycle: + source = create_market_data_source(cache) + await source.start(["AAPL", "GOOGL", ...]) + # ... app runs ... + await source.add_ticker("TSLA") + await source.remove_ticker("GOOGL") + # ... app shutting down ... + await source.stop() + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing price updates for the given tickers. + + Starts a background task that periodically writes to the PriceCache. + Must be called exactly once. Calling start() twice is undefined behavior. + """ + + @abstractmethod + async def stop(self) -> None: + """Stop the background task and release resources. + + Safe to call multiple times. After stop(), the source will not write + to the cache again. + """ + + @abstractmethod + async def add_ticker(self, ticker: str) -> None: + """Add a ticker to the active set. No-op if already present. + + The next update cycle will include this ticker. + """ + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Remove a ticker from the active set. No-op if not present. + + Also removes the ticker from the PriceCache. + """ + + @abstractmethod + def get_tickers(self) -> list[str]: + """Return the current list of actively tracked tickers.""" +``` + +Both implementations own an internal `asyncio.Task` started in `start()` and cancelled in `stop()`, so this ABC needs no `run()`/`loop()` method of its own — the task's existence is an implementation detail of each subclass. + +--- + +## 6. Seed Prices & Ticker Parameters + +**File: `seed_prices.py`.** ✅ Constants only, no logic, no imports beyond the type hints — tuning a starting price or a sector's volatility never touches `simulator.py`. + +```python +"""Seed prices and per-ticker parameters for the market simulator.""" + +SEED_PRICES: dict[str, float] = { + "AAPL": 190.00, "GOOGL": 175.00, "MSFT": 420.00, "AMZN": 185.00, "TSLA": 250.00, + "NVDA": 800.00, "META": 500.00, "JPM": 195.00, "V": 280.00, "NFLX": 600.00, +} + +TICKER_PARAMS: dict[str, dict[str, float]] = { + "AAPL": {"sigma": 0.22, "mu": 0.05}, + "GOOGL": {"sigma": 0.25, "mu": 0.05}, + "MSFT": {"sigma": 0.20, "mu": 0.05}, + "AMZN": {"sigma": 0.28, "mu": 0.05}, + "TSLA": {"sigma": 0.50, "mu": 0.03}, # High volatility + "NVDA": {"sigma": 0.40, "mu": 0.08}, # High volatility, strong drift + "META": {"sigma": 0.30, "mu": 0.05}, + "JPM": {"sigma": 0.18, "mu": 0.04}, # Low volatility (bank) + "V": {"sigma": 0.17, "mu": 0.04}, # Low volatility (payments) + "NFLX": {"sigma": 0.35, "mu": 0.05}, +} + +DEFAULT_PARAMS: dict[str, float] = {"sigma": 0.25, "mu": 0.05} + +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 +``` + +Tickers not in `SEED_PRICES` (dynamically added via chat or the watchlist route) start at a uniform random price in `[$50, $300]` and use `DEFAULT_PARAMS` — there's no real-world volatility figure to look up for an arbitrary user-entered symbol. Note there is deliberately **no separate `DEFAULT_CORR` constant** — the fallback for any pair not covered by the tech/finance/TSLA special cases is `CROSS_GROUP_CORR`, and the two names are not duplicated (an earlier draft had both `DEFAULT_CORR` and `CROSS_GROUP_CORR` at the same value, which a code review flagged as confusing; the current code has one name for one concept). + +--- + +## 7. GBM Simulator — `simulator.py` + +✅ Implemented. Two classes in one module: `GBMSimulator` (pure math, synchronous, no asyncio/cache knowledge) and `SimulatorDataSource` (the async `MarketDataSource` adapter). + +### 7.1 Math + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +- `S(t)` current price, `mu` annualized drift, `sigma` annualized volatility, `dt` this tick as a fraction of a trading year, `Z` a correlated standard-normal draw. +- Multiplicative (`exp()`-based) update ⇒ prices can never go negative, unlike an additive random walk. +- `dt = 0.5 / (252 * 6.5 * 3600) ≈ 8.48e-8` for 500ms ticks over a 252-day, 6.5h/day trading year — tiny enough that each tick moves sub-cent amounts which accumulate into realistic intraday ranges over minutes of wall-clock time. + +### 7.2 Correlated moves + +Real stocks don't move independently. The simulator builds a Cholesky decomposition `L` of a correlation matrix `C` (`L = cholesky(C)`) and applies it to independent normal draws: `Z_correlated = L @ Z_independent`. Exact as long as `C` is positive semi-definite, which any matrix built from pairwise correlations in `[0, 1)` with a unit diagonal always is — so `np.linalg.cholesky` never needs a defensive `try/except` for a non-PSD failure here. + +Pairwise correlation lookup checks TSLA first (`0.3` with everything, even though it's nominally in the `tech` set), then tech-tech (`0.6`), then finance-finance (`0.5`), else `0.3`. + +### 7.3 Random events + +Each tick, each ticker independently has a small chance of a sudden 2–5% shock: + +```python +if random.random() < event_probability: # default 0.001 + shock = random.uniform(0.02, 0.05) * random.choice([-1, 1]) + price *= 1 + shock +``` + +At 2 ticks/sec, `0.001` per ticker per tick ⇒ an event on a *given* ticker roughly every 500s (~8 min); across a 10-ticker default watchlist, expect *some* ticker to have an event roughly every 50s — enough drama for a live dashboard without destabilizing the price path. + +### 7.4 `GBMSimulator` + +```python +class GBMSimulator: + TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 + DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR + + 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 every tracked ticker by one time step. Hot path — every 500ms.""" + n = len(self._tickers) + if n == 0: + return {} + + z_independent = np.random.standard_normal(n) + z = self._cholesky @ z_independent if self._cholesky is not None else z_independent + + result: dict[str, float] = {} + for i, ticker in enumerate(self._tickers): + mu, sigma = self._params[ticker]["mu"], self._params[ticker]["sigma"] + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z[i] + self._prices[ticker] *= math.exp(drift + diffusion) + + if random.random() < self._event_prob: + shock = random.uniform(0.02, 0.05) * random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock + + result[ticker] = round(self._prices[ticker], 2) + return result + + def add_ticker(self, ticker: str) -> None: + if ticker in self._prices: + return + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + def remove_ticker(self, ticker: str) -> None: + if ticker not in self._prices: + return + self._tickers.remove(ticker) + del self._prices[ticker] + del self._params[ticker] + self._rebuild_cholesky() + + def get_price(self, ticker: str) -> float | None: + return self._prices.get(ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + def _add_ticker_internal(self, ticker: str) -> None: + if ticker in self._prices: + return + self._tickers.append(ticker) + self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) + self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) + + def _rebuild_cholesky(self) -> None: + """O(n^2) rebuild whenever tickers are added/removed. n stays small (<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: + if t1 == "TSLA" or t2 == "TSLA": + return TSLA_CORR + tech, finance = CORRELATION_GROUPS["tech"], CORRELATION_GROUPS["finance"] + 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 +``` + +`step()` is allocation-light on purpose (one vectorized `numpy` draw per tick for *all* tickers, not per-ticker `random.gauss()` calls) since it runs every 500ms for the life of the process. The Cholesky matrix is cached and rebuilt only on `add_ticker`/`remove_ticker`, never inside `step()`. `round(self._prices[ticker], 2)` only rounds the **emitted** value — the internal `self._prices[ticker]` stays full precision so tiny per-tick drifts aren't lost to repeated rounding over a long-running session. + +### 7.5 `SimulatorDataSource` — async adapter + +```python +class SimulatorDataSource(MarketDataSource): + 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) + for ticker in tickers: + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + logger.info("Simulator started with %d tickers", len(tickers)) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("Simulator stopped") + + async def add_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.add_ticker(ticker) + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + + async def remove_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.remove_ticker(ticker) + self._cache.remove(ticker) + + def get_tickers(self) -> list[str]: + return self._sim.get_tickers() if self._sim else [] + + async def _run_loop(self) -> None: + while True: + try: + if self._sim: + for ticker, price in self._sim.step().items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +Two resilience choices worth preserving in any future edit: **seed the cache synchronously inside `start()`**, before the loop task even exists, so an SSE client connecting immediately after boot sees prices on its first read instead of an empty payload; and **wrap each step in `try/except Exception`** so one bad draw logs and continues rather than silently killing the price feed for the rest of the process's life. + +--- + +## 8. Massive API Client — `massive_client.py` + +✅ Implemented, and already reflects the two corrections flagged during research against the live `massive` package (`planning/MASSIVE_API.md` "Discrepancies & Gotchas" — both were live bugs against an *earlier* draft of this file, already fixed in current code): + +- `get_snapshot_all("stocks", tickers)` — positional string + list, not `SnapshotMarketType.STOCKS`. +- `snap.last_trade.sip_timestamp / 1_000_000_000` — Massive's `LastTrade` has no `.timestamp` field; `sip_timestamp` is Unix **nanoseconds**, not milliseconds. + +```python +from __future__ import annotations + +import asyncio +import logging + +from massive import RESTClient + +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 (Basic) tier: does not include this endpoint at all (403) — a + Starter plan or above is required for get_snapshot_all. + - Starter and above: effectively unlimited (soft guidance: stay under + 100 req/s), so the poll interval is a design choice, not a rate-limit + constraint; default here is a conservative 15s. + """ + + 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 isn't empty on connect + self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + self._client = None + + async def add_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + if ticker not in self._tickers: + self._tickers.append(ticker) + + async def remove_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + self._tickers = [t for t in self._tickers if t != ticker] + self._cache.remove(ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + async def _poll_loop(self) -> None: + while True: + await asyncio.sleep(self._interval) + await self._poll_once() + + async def _poll_once(self) -> None: + if not self._tickers or not self._client: + return + try: + # RESTClient is synchronous — run in a thread to avoid blocking the event loop. + snapshots = await asyncio.to_thread(self._fetch_snapshots) + processed = 0 + for snap in snapshots: + try: + price = snap.last_trade.price + timestamp = snap.last_trade.sip_timestamp / 1_000_000_000 # ns -> s + self._cache.update(ticker=snap.ticker, price=price, timestamp=timestamp) + processed += 1 + except (AttributeError, TypeError) as e: + logger.warning("Skipping snapshot for %s: %s", getattr(snap, "ticker", "???"), e) + logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers)) + except Exception as e: + logger.error("Massive poll failed: %s", e) + # Never re-raise — the loop retries on the next interval. + # Common failures: 401 (bad key), 403 (plan doesn't include this endpoint), 429, network errors. + + def _fetch_snapshots(self) -> list: + return self._client.get_snapshot_all("stocks", self._tickers) +``` + +Two resilience choices worth keeping in any future rewrite: an **immediate poll inside `start()`** (so the SSE stream has data on the very first client connection, not just after the first `poll_interval` elapses), and a **per-snapshot `try/except` inside the batch loop** (one malformed ticker in a 20-ticker response doesn't drop the other 19). + +| Error | Behavior | +|---|---| +| 401 Unauthorized | Logged; poller keeps running (user might fix `.env` and restart) | +| 403 Plan doesn't include endpoint | Same — this is the practical failure mode for a free/Basic key on `get_snapshot_all` | +| 429 Rate limited | Logged; next poll retries after `poll_interval` | +| Network timeout | Logged; retries next cycle | +| Malformed single snapshot | That ticker skipped with a warning; others still processed | +| All tickers fail | Cache retains last-known prices — SSE keeps streaming stale data rather than nothing | + +**Open item carried from `planning/MASSIVE_API.md`**: a free (Basic) `MASSIVE_API_KEY` will most likely 403 on `get_snapshot_all` rather than just being rate-limited, since that endpoint family isn't included in the Basic plan. Not a code bug — worth documenting for users ("Massive real-data mode requires Starter tier or above") rather than fixing in code, since `get_previous_close_agg` (which *is* free-tier) can't produce intraday movement. + +--- + +## 9. Factory — `factory.py` + +✅ Implemented. + +```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: + """MASSIVE_API_KEY set and non-empty -> MassiveDataSource, else -> SimulatorDataSource. + + Returns an unstarted source. Caller must await source.start(tickers). + """ + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + + if api_key: + logger.info("Market data source: Massive API (real data)") + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + else: + logger.info("Market data source: GBM Simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +`.strip()` matters: an `.env` with `MASSIVE_API_KEY=` present but empty must fall through to the simulator, not attempt a Massive connection with an empty key. + +--- + +## 10. SSE Streaming Endpoint — `stream.py` + +✅ Implemented. `create_stream_router(price_cache)` is a **factory** returning a FastAPI `APIRouter` — the cache is injected, not reached through a module global. + +```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: + @router.get("/prices") + async def stream_prices(request: Request) -> StreamingResponse: + return StreamingResponse( + _generate_events(price_cache, request), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", # disable nginx buffering if proxied + }, + ) + return router + + +async def _generate_events( + price_cache: PriceCache, request: Request, interval: float = 0.5, +) -> AsyncGenerator[str, None]: + yield "retry: 1000\n\n" + + 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 (as implemented — note this diverges from a literal reading of `planning/PLAN.md` §8's "each SSE event contains ticker, price, previous price, timestamp, direction") + +Each `data:` line is **one batched JSON object keyed by ticker**, covering every currently tracked ticker whose cache entry has changed since the last send — not one event per ticker: + +``` +data: {"AAPL":{"ticker":"AAPL","price":190.50,"previous_price":190.42,"timestamp":1707580800.5,"change":0.08,"change_percent":0.042,"direction":"up"},"GOOGL":{"ticker":"GOOGL","price":175.12,...}} + +``` + +- Gated by `PriceCache.version` — a whole batch is skipped if nothing changed since the last poll (important for the Massive source's 15s cadence vs. the 500ms poll loop). +- No `event:` name is set (defaults to the generic `message` event) and there is no periodic heartbeat/comment line — only the one-time `retry: 1000\n\n` directive at connection start, which sets the browser's `EventSource` reconnect delay to 1s. +- Client: + +```javascript +const eventSource = new EventSource('/api/stream/prices'); +eventSource.onmessage = (event) => { + const prices = JSON.parse(event.data); // { "AAPL": {...}, "GOOGL": {...}, ... } +}; +``` + +The SSE loop **polls** the cache on a fixed interval rather than being notified by the data source — deliberately simpler, and it produces predictable, evenly-spaced updates, which matters for building clean sparklines on the frontend. + +--- + +## 11. History Endpoint 🚧 + +**File: `backend/app/market/history.py` (new).** Depends on the `PriceCache.get_history()` extension in §4.2. Thin by design — the buffer already holds oldest-first `PriceUpdate`s, so the route is a direct read-through: + +```python +from __future__ import annotations + +from fastapi import APIRouter, HTTPException + +from .cache import PriceCache + +router = APIRouter(prefix="/api/market", tags=["market"]) + + +def create_history_router(price_cache: PriceCache) -> APIRouter: + @router.get("/history/{ticker}") + async def get_history(ticker: str) -> list[dict]: + """Recent price history for a ticker, oldest first. + + Used to pre-populate sparklines and the main chart on page load; + the frontend appends subsequent points from the SSE stream. + 404 if the ticker isn't currently tracked (never in the cache, or + removed from both the watchlist and all positions). + """ + ticker = ticker.upper().strip() + if ticker not in price_cache: + raise HTTPException(status_code=404, detail=f"No price data for {ticker}") + return [u.to_dict() for u in price_cache.get_history(ticker)] + + return router +``` + +Notes: + +- **404 vs. empty list**: a ticker that's tracked but has zero history yet (added this instant, before the first `update()` call lands) is a narrow race — both data sources seed the cache synchronously in `start()`/`add_ticker()` before returning, so in practice `ticker in price_cache` and `get_history(ticker)` non-empty happen together. If that race is ever observed, prefer returning `200` with an empty list over `404`, since the ticker *is* validly tracked — reserve `404` for "never heard of this ticker." +- **Uppercasing the path param** mirrors the normalization already applied in `MassiveDataSource.add_ticker/remove_ticker` (`ticker.upper().strip()`), so `GET /api/market/history/aapl` and `GET /api/market/history/AAPL` hit the same cache entry. +- Registered in `main.py` the same way as `create_stream_router` — see §12. + +--- + +## 12. FastAPI Lifecycle Integration 🚧 + +`backend/app/main.py` does not exist yet. This is the wiring that starts/stops the market data subsystem with the app's lifespan and exposes `PriceCache` / `MarketDataSource` to the rest of the backend (trade execution, watchlist routes) via dependency injection. + +```python +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from app.market.cache import PriceCache +from app.market.factory import create_market_data_source +from app.market.history import create_history_router +from app.market.interface import MarketDataSource +from app.market.stream import 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_tracked_tickers() # watchlist ∪ positions.ticker, from SQLite — see §13 + await source.start(initial_tickers) + + app.include_router(create_stream_router(price_cache)) + app.include_router(create_history_router(price_cache)) + + yield # app is running + + # --- shutdown --- + await source.stop() + + +app = FastAPI(title="FinAlly", lifespan=lifespan) + + +def get_price_cache() -> PriceCache: + return app.state.price_cache + + +def get_market_source() -> MarketDataSource: + return app.state.market_source +``` + +Other routes reach the cache/source through FastAPI's `Depends`, never through a module-level global: + +```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(400, f"Price not yet available for {trade.ticker}") + # ... execute trade at current_price ... +``` + +**Startup ordering matters**: `source.start(initial_tickers)` seeds the cache synchronously before either router is registered, so the very first SSE connection and the very first history request both see data immediately — there's no window where the routes exist but the cache is empty. + +--- + +## 13. Watchlist Coordination 🚧 + +Per `planning/PLAN.md` §6, the set passed to `start()` / `add_ticker()` / `remove_ticker()` must always be **`watchlist ∪ positions.ticker`**, not just the watchlist — this keeps a live price available for the positions table and portfolio valuation even after a held ticker is removed from the watchlist. Neither `MarketDataSource` implementation enforces this itself (by design — see §5); it's the responsibility of whichever route mutates the watchlist or a position. + +### Adding a ticker + +``` +POST /api/watchlist {ticker: "PYPL"} + → validate: watchlist not already at the 20-ticker cap (else 400) + → INSERT INTO watchlist (SQLite) + → await source.add_ticker("PYPL") + Simulator: adds to GBMSimulator, rebuilds Cholesky, seeds cache (+ history, once §4.2 lands) + Massive: appends to poll list, appears on the next cycle + → 200 { ticker, price: price_cache.get_price("PYPL") } # may be null if Massive hasn't polled yet +``` + +### Removing a ticker + +```python +@router.delete("/watchlist/{ticker}") +async def remove_from_watchlist( + ticker: str, + source: MarketDataSource = Depends(get_market_source), +): + await db.delete_watchlist_entry(ticker) + + # Only stop tracking if there's no open position — a held ticker must + # keep a live price for the positions table / portfolio valuation + # even once it's off the watchlist. + position = await db.get_position(ticker) + if position is None or position.quantity == 0: + await source.remove_ticker(ticker) + + return {"status": "ok"} +``` + +The inverse case — a trade creates a *new* position in a ticker that isn't on the watchlist (e.g. the LLM buys something not currently watched) — needs the same treatment on the trade path: after a buy that opens a new position, call `source.add_ticker(ticker)` if it isn't already tracked, so the positions table has a live price without requiring the ticker to also be added to the watchlist. + +--- + +## 14. Daily-Change-% Semantics + +`planning/PLAN.md` §10 requires "daily change %" in the watchlist panel and positions table. `PriceUpdate.change_percent` (§3) is **tick-over-tick** — this update vs. the immediately preceding one, not vs. session open or previous close. For the simulator that's ~500ms of movement; for Massive that's ~15s. Labeling that as "daily change" would be misleading — a review flagged this as unresolved in the plan text. + +Two ways to close this, neither requiring a `PriceCache` redesign: + +1. **Relabel in the UI as "change" (drop "daily")** and keep using tick-over-tick `change_percent` as-is. Simplest; accurate; loses the "since market open" framing. +2. **Add a session-open baseline.** Record each ticker's *first* price of the current process/session (or, for Massive, use `TickerSnapshot.todays_change_percent`, which the API already computes against the real previous close — see `planning/MASSIVE_API.md` §1) and compute `%` against that instead of the previous tick. This means the simulator and Massive would compute "daily change" differently (simulator: first price since process start; Massive: real previous-close-relative), which is honest given the simulator has no concept of a trading session, but should be called out explicitly in the frontend contract so it isn't read as identical semantics across both data sources. + +This document doesn't pick one — it's a product decision belonging in `planning/PLAN.md` (already flagged there as open) — but recommends whichever choice is made gets encoded once, in `PriceCache` or a thin wrapper, rather than computed ad hoc in the frontend from `previous_price`. + +--- + +## 15. Testing Strategy + +Existing coverage under `backend/tests/market/` (all passing as of the last review pass against an environment with `massive` installed): `test_models.py`, `test_cache.py`, `test_factory.py`, `test_simulator.py`, `test_simulator_source.py`, `test_massive.py`. Representative cases already in place: + +- **`GBMSimulator`**: `step()` returns all tracked tickers; prices always positive over 10k steps; initial price matches seed; add/remove ticker updates the tracked set and rebuilds correlation; duplicate add / unknown-ticker remove are no-ops; unknown ticker gets a `[$50, $300]` random seed; empty ticker list steps to `{}`; Cholesky is `None` for 1 ticker, non-`None` for 2+. +- **`PriceCache`**: update/get round-trip; first update is `direction="flat"`; up/down direction and `change` sign; `remove()`; `get_all()`; `version` increments once per `update()`; `get_price()` convenience. +- **`SimulatorDataSource`**: `start()` populates the cache before the loop's first tick; prices change over multiple intervals; `stop()` is idempotent (safe to call twice); `add_ticker`/`remove_ticker` reflected in both `get_tickers()` and the cache. +- **`MassiveDataSource`** (mocked, no network): `_poll_once` updates the cache from mocked snapshots; a malformed snapshot (`last_trade=None`) is skipped without dropping other tickers; an exception from `_fetch_snapshots` doesn't propagate out of `_poll_once`. + +### New tests needed for the proposed additions 🚧 + +**`backend/tests/market/test_cache.py`** — extend for the history buffer: + +```python +class TestPriceCacheHistory: + def test_history_starts_empty(self): + cache = PriceCache() + assert cache.get_history("AAPL") == [] + + def test_history_accumulates_in_order(self): + cache = PriceCache() + cache.update("AAPL", 190.00) + cache.update("AAPL", 191.00) + cache.update("AAPL", 192.00) + prices = [u.price for u in cache.get_history("AAPL")] + assert prices == [190.00, 191.00, 192.00] + + def test_history_is_bounded(self): + cache = PriceCache(history_size=3) + for p in [1.0, 2.0, 3.0, 4.0, 5.0]: + cache.update("AAPL", p) + prices = [u.price for u in cache.get_history("AAPL")] + assert prices == [3.0, 4.0, 5.0] # oldest evicted first + + def test_history_independent_per_ticker(self): + cache = PriceCache() + cache.update("AAPL", 190.00) + cache.update("GOOGL", 175.00) + assert len(cache.get_history("AAPL")) == 1 + assert len(cache.get_history("GOOGL")) == 1 + + def test_remove_clears_history(self): + cache = PriceCache() + cache.update("AAPL", 190.00) + cache.remove("AAPL") + assert cache.get_history("AAPL") == [] +``` + +**`backend/tests/market/test_history.py`** — the new route, via FastAPI's test client: + +```python +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.market.cache import PriceCache +from app.market.history import create_history_router + + +@pytest.fixture +def client(): + cache = PriceCache() + cache.update("AAPL", 190.00) + cache.update("AAPL", 191.00) + app = FastAPI() + app.include_router(create_history_router(cache)) + return TestClient(app), cache + + +class TestHistoryEndpoint: + def test_returns_history_oldest_first(self, client): + test_client, _ = client + resp = test_client.get("/api/market/history/AAPL") + assert resp.status_code == 200 + body = resp.json() + assert [p["price"] for p in body] == [190.00, 191.00] + + def test_lowercase_ticker_normalized(self, client): + test_client, _ = client + resp = test_client.get("/api/market/history/aapl") + assert resp.status_code == 200 + assert len(resp.json()) == 2 + + def test_unknown_ticker_404s(self, client): + test_client, _ = client + resp = test_client.get("/api/market/history/ZZZZ") + assert resp.status_code == 404 +``` + +### Still worth adding (carried over from `planning/archive/MARKET_DATA_REVIEW.md`, not yet done) + +- At least one SSE integration test against a running ASGI test client (`stream.py` currently has the lowest coverage of the package — it needs a live server to exercise the generator). +- A `GBMSimulator` test with the full 10-ticker default set, to catch any future correlation-matrix issue that only appears at that size (existing tests use 1–2 tickers). +- A concurrent-write test for `PriceCache` (multiple threads calling `update()` simultaneously) to empirically confirm the lock, not just inspect it. + +--- + +## 16. Error Handling & Edge Cases + +| Scenario | Behavior | +|---|---| +| Empty initial watchlist at startup | `start([])` — simulator produces no prices, Massive poller skips its API call entirely (`if not self._tickers: return`). SSE sends nothing (`if prices:` guard). First ticker added starts tracking immediately. | +| Trade on a ticker with no cached price | `price_cache.get_price(ticker) is None` → route returns `400` with a message telling the user to retry ("Price not yet available for {ticker}"). The simulator avoids this in practice by seeding synchronously in `start()`/`add_ticker()`; Massive may have a real gap between "ticker added" and "first successful poll." | +| Massive API key set but invalid | First poll 401s, logged, poller keeps retrying every `poll_interval`. SSE endpoint is "connected" but streams empty/stale data — a UI relying solely on the SSE connection-status dot would show green with no prices, which is misleading; consider surfacing cache staleness (e.g. via `timestamp` age) in the frontend rather than only connection state. | +| Massive 403 on Basic/free tier | Every poll fails identically; same behavior as an invalid key. Document that real-data mode needs Starter+ (see §8). | +| Ticker removed from watchlist while still held | Must **not** call `source.remove_ticker()` — see §13. Removing the ticker from `PriceCache` would break positions-table valuation. | +| `GET /api/market/history/{ticker}` for an untracked ticker | `404` (§11) — distinguishes "never tracked" from "tracked but momentarily empty." | +| Simulator step raises | Caught per-tick inside `_run_loop`'s `try/except Exception`; logged, loop continues on the next `sleep`. A single bad `numpy` draw does not kill the price feed. | +| `PriceCache` under concurrent load | `threading.Lock` serializes all reads/writes; critical sections are a dict lookup + assignment, negligible contention at the project's scale (≤50 tracked tickers, sub-second cadence). A `ReadWriteLock` would only matter at a scale this project doesn't target. | + +--- + +## 17. Configuration Summary + +| Parameter | Location | Default | Notes | +|---|---|---|---| +| `MASSIVE_API_KEY` | env var | `""` | Non-empty (after `.strip()`) → Massive; else simulator | +| `update_interval` | `SimulatorDataSource.__init__` | `0.5s` | Simulator tick cadence | +| `poll_interval` | `MassiveDataSource.__init__` | `15.0s` | Conservative default; safe even on a rate-limited plan | +| `event_probability` | `GBMSimulator.__init__` | `0.001` | Shock chance per ticker per tick | +| `dt` | `GBMSimulator.__init__` | `≈8.48e-8` | Fraction of a trading year per 500ms tick | +| SSE poll interval | `_generate_events()` | `0.5s` | Independent of the underlying source's own cadence | +| SSE retry directive | `_generate_events()` | `1000ms` | Browser `EventSource` auto-reconnect delay | +| `history_size` 🚧 | `PriceCache.__init__` | `500` | Points retained per ticker (~4 min simulator / ~2 hr Massive) | +| Watchlist cap | route-level validation 🚧 | `20` | `POST /api/watchlist` → `400` past the cap; positions are never capped | + +### `__init__.py` — public API (extend once §11 lands) + +```python +"""Market data subsystem for FinAlly.""" + +from .cache import PriceCache +from .factory import create_market_data_source +from .history import create_history_router +from .interface import MarketDataSource +from .models import PriceUpdate +from .stream import create_stream_router + +__all__ = [ + "PriceUpdate", + "PriceCache", + "MarketDataSource", + "create_market_data_source", + "create_stream_router", + "create_history_router", +] +```