From 09e1857eedd2f93cc4ce3fcbb58d2a1fbb16461b Mon Sep 17 00:00:00 2001 From: Raunak Sachdev Date: Sat, 22 Aug 2026 16:55:40 +0100 Subject: [PATCH 1/7] Rewrite README to reflect actual project state, update PLAN.md, add agent tooling README now describes what's actually built (market data subsystem only) rather than the full target app, with planned features/API/testing sections linked back to PLAN.md. PLAN.md updated to match decisions made since the last pass: Angular over Next.js, free-tier OpenRouter model, ECharts, plus SQLite/avg-cost/ SSE clarifications. Adds project-level agent tooling (change-reviewer subagent, doc-review command, marketplace.json) and its first review output. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019K6SfhrZQx6fCUASEDwJNW --- .claude-plugin/marketplace.json | 8 ++ .claude/agents/reviewer.md | 8 ++ .claude/commands/doc-review.md | 1 + .claude/skills/cerebras/SKILL.md | 2 + .gitignore | 5 + README.md | 153 +++++++++++++++++++++------- planning/PLAN.md | 50 ++++++--- planning/review.md | 170 +++++++++++++++++++++++++++++++ 8 files changed, 345 insertions(+), 52 deletions(-) create mode 100644 .claude-plugin/marketplace.json create mode 100644 .claude/agents/reviewer.md create mode 100644 .claude/commands/doc-review.md create mode 100644 planning/review.md diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json new file mode 100644 index 000000000..80b4e20da --- /dev/null +++ b/.claude-plugin/marketplace.json @@ -0,0 +1,8 @@ +{ + "name": "raunak-tools", + "owner": { + "name": "Raunak Sachdev", + "email": "sachdevraunak1991@gmail.com" + }, + "plugins": [] +} diff --git a/.claude/agents/reviewer.md b/.claude/agents/reviewer.md new file mode 100644 index 000000000..6d169b33c --- /dev/null +++ b/.claude/agents/reviewer.md @@ -0,0 +1,8 @@ +--- +name: change-reviewer +description: carry out comprehensive review of all changes since last commit +tools: Read, Grep, Glob, Bash, Write, Edit +model: inherit +--- + +You are the reviewer agent for the FinAlly project. Your sole job is to review changes since last commit. You only ever write your output to `planning/review.md`. diff --git a/.claude/commands/doc-review.md b/.claude/commands/doc-review.md new file mode 100644 index 000000000..1c2750407 --- /dev/null +++ b/.claude/commands/doc-review.md @@ -0,0 +1 @@ +Review the documentation file in the planning folder called $ARGUMENTS and add questions, clarifications or feedback to a new section at the end, along with any opportunities to simplify \ No newline at end of file diff --git a/.claude/skills/cerebras/SKILL.md b/.claude/skills/cerebras/SKILL.md index 9efd01a38..6e44349ad 100644 --- a/.claude/skills/cerebras/SKILL.md +++ b/.claude/skills/cerebras/SKILL.md @@ -5,6 +5,8 @@ description: Use this to write code to call an LLM using LiteLLM and OpenRouter # Calling an LLM via Cerebras +> **Not for FinAlly's chat feature.** `planning/PLAN.md` §9 specifies `openrouter/nvidia/nemotron-3-ultra-550b-a55b:free` with no Cerebras provider routing for FinAlly's AI chat assistant. Do not use this skill (or the `gpt-oss-120b`/Cerebras example below) to implement that feature — follow PLAN.md §9 instead. This skill remains available for other, unrelated Cerebras-routed LLM calls. + These instructions allow you write code to call an LLM with Cerebras specified as the inference provider. This method uses LiteLLM and OpenRouter. diff --git a/.gitignore b/.gitignore index b7faf403d..230065f81 100644 --- a/.gitignore +++ b/.gitignore @@ -61,6 +61,11 @@ local_settings.py db.sqlite3 db.sqlite3-journal +# FinAlly runtime database (see planning/PLAN.md §4) +db/*.db +db/*.db-journal +!db/.gitkeep + # Flask stuff: instance/ .webassets-cache diff --git a/README.md b/README.md index 3f2582ae2..06ac55b22 100644 --- a/README.md +++ b/README.md @@ -1,62 +1,139 @@ -# FinAlly — AI Trading Workstation +# FinAlly — the Finance Ally -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. +A visually stunning, AI-powered trading workstation: live-streaming market data, a simulated portfolio, and 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. Agents coordinate through documents in [`planning/`](planning/), most importantly [`planning/PLAN.md`](planning/PLAN.md), the full project specification. -## Features +## Status -- **Live price streaming** via SSE with green/red flash animations -- **Simulated portfolio** — $10k virtual cash, market orders, instant fills -- **Portfolio visualizations** — heatmap (treemap), P&L chart, positions table -- **AI chat assistant** — analyzes holdings, suggests and auto-executes trades -- **Watchlist management** — track tickers manually or via AI -- **Dark terminal aesthetic** — Bloomberg-inspired, data-dense layout +🚧 **In progress.** The market data subsystem is complete; the rest of the platform (API, database, frontend, LLM chat, Docker packaging) is still being built. See [`planning/MARKET_DATA_SUMMARY.md`](planning/MARKET_DATA_SUMMARY.md) for what's done. -## Architecture +| Component | Status | +|---|---| +| Market data simulator (GBM, SSE-ready) | ✅ Complete | +| Massive (Polygon.io) live data client | ✅ Complete, unused by default | +| FastAPI app, database, portfolio/trade endpoints | ⏳ Not started | +| LLM chat assistant | ⏳ Not started | +| Angular frontend | ⏳ Not started | +| Docker packaging | ⏳ Not started | -Single Docker container serving everything on port 8000: +## Vision -- **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) +When finished, running a single command will open a browser to a live trading terminal: a watchlist of streaming prices, a $10,000 virtual cash balance, portfolio visualizations (heatmap, P&L chart, positions table), and a docked AI chat assistant that can analyze the portfolio and place trades through natural language. Full UX details are in [`planning/PLAN.md`](planning/PLAN.md) §2. -## Quick Start +### Planned features + +- **Live price streaming** — 10 default tickers (AAPL, GOOGL, MSFT, AMZN, TSLA, NVDA, META, JPM, V, NFLX) updating over SSE, with green/red flash animations on each tick +- **Sparklines** — mini price-history charts next to each watchlist ticker, accumulated client-side since page load +- **Buy/sell** — market orders only, instant fill at current price, no fees or confirmation dialogs +- **Portfolio heatmap** — treemap sized by position weight, colored by P&L +- **P&L chart** — total portfolio value over time +- **Positions table** — ticker, quantity, avg cost, current price, unrealized P&L, % change +- **AI chat assistant** — "FinAlly", backed by an LLM, that can analyze the portfolio and auto-execute trades or watchlist changes it recommends +- **Dark, data-dense terminal UI** — Bloomberg-inspired, accent yellow `#ecad0a`, blue `#209dd7`, purple `#753991` + +### Planned API surface + +| Method | Path | Description | +|---|---|---| +| GET | `/api/stream/prices` | SSE stream of live price updates | +| GET | `/api/portfolio` | 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 (for the P&L chart) | +| GET | `/api/watchlist` | Current watchlist with latest prices | +| POST | `/api/watchlist` | Add a ticker: `{ticker}` | +| DELETE | `/api/watchlist/{ticker}` | Remove a ticker | +| POST | `/api/chat` | Send a chat message, get back a response plus any executed actions | +| GET | `/api/health` | Health check | + +Full request/response contracts and the SQLite schema (`users_profile`, `watchlist`, `positions`, `trades`, `portfolio_snapshots`, `chat_messages`) are in [`planning/PLAN.md`](planning/PLAN.md) §7–8. + +## Architecture (target) + +Everything ships in a single Docker container on one port: + +``` +┌─────────────────────────────────────────────────┐ +│ Docker Container (port 8000) │ +│ │ +│ FastAPI (Python/uv) │ +│ ├── /api/* REST endpoints │ +│ ├── /api/stream/* SSE streaming │ +│ └── /* Static file serving │ +│ (Angular build) │ +│ │ +│ SQLite database (volume-mounted) │ +│ Background task: market data polling/sim │ +└─────────────────────────────────────────────────┘ +``` + +- **Frontend**: Angular + TypeScript, built to static assets and served by FastAPI +- **Backend**: FastAPI (Python), managed as a `uv` project +- **Database**: SQLite, lazily initialized, volume-mounted at `db/finally.db` +- **Real-time data**: Server-Sent Events (`/api/stream/prices`) +- **AI**: LiteLLM → OpenRouter, structured outputs for chat-driven trades +- **Market data**: simulator by default; real data via Massive API if `MASSIVE_API_KEY` is set + +Full rationale for these choices is in [`planning/PLAN.md`](planning/PLAN.md) §3. + +## What's built so far: market data + +A self-contained market data subsystem lives in `backend/app/market/` — a `PriceCache`, a GBM-based simulator with correlated, per-sector price moves, a Massive/Polygon.io REST client behind the same interface, and an SSE stream factory. It's fully tested (73 tests, 91% coverage overall — `stream.py` is the weak spot at 33%, everything else is 94-100%) and has a standalone terminal demo: ```bash -# Clone and configure -cp .env.example .env -# Add your OPENROUTER_API_KEY to .env +cd backend +uv sync --dev +uv run market_data_demo.py +``` + +This runs a live Rich dashboard of all 10 default tickers with sparklines and an event log — no server, database, or frontend required. See [`backend/README.md`](backend/README.md) and [`planning/MARKET_DATA_SUMMARY.md`](planning/MARKET_DATA_SUMMARY.md) for details. -# Run with Docker -docker build -t finally . -docker run -v finally-data:/app/db -p 8000:8000 --env-file .env finally +Run the backend test suite: -# Open http://localhost:8000 +```bash +cd backend +uv run pytest ``` ## 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) | +Create a `.env` file at the project root: -## Project Structure +```bash +# Required for AI chat once it's built +OPENROUTER_API_KEY=your-openrouter-api-key-here + +# Optional: use real market data instead of the simulator +MASSIVE_API_KEY= + +# Optional: deterministic mock LLM responses (for testing) +LLM_MOCK=false +``` + +## Project Layout ``` 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 +├── backend/ # FastAPI uv project (Python) +│ └── app/market/ # Market data subsystem (complete) +├── frontend/ # Angular project (not yet created) +├── planning/ # Shared spec and docs the agents build from +│ ├── PLAN.md +│ └── MARKET_DATA_SUMMARY.md +├── db/ # SQLite volume mount point (runtime) +└── test/ # Playwright E2E tests (not yet created) ``` +See [`planning/PLAN.md`](planning/PLAN.md) §4 for the full target layout and the boundaries between components. + +## Testing Strategy (planned) + +- **Backend (pytest)** — market data math, trade execution and P&L edge cases, LLM structured-output parsing, API route contracts +- **Frontend (Jasmine/Karma via Angular CLI)** — component rendering, price flash animations, watchlist CRUD, chat rendering +- **E2E (Playwright, in `test/`)** — fresh-start flow, watchlist add/remove, buy/sell, portfolio visualizations, mocked AI chat, SSE reconnection. Runs against a container with `LLM_MOCK=true` for speed and determinism. + +Full scenario list is in [`planning/PLAN.md`](planning/PLAN.md) §12. + ## License -See [LICENSE](LICENSE). +MIT — see [LICENSE](LICENSE). diff --git a/planning/PLAN.md b/planning/PLAN.md index bc1811b33..820091ca6 100644 --- a/planning/PLAN.md +++ b/planning/PLAN.md @@ -55,18 +55,18 @@ The user runs a single Docker command (or a provided start script). A browser op │ ├── /api/* REST endpoints │ │ ├── /api/stream/* SSE streaming │ │ └── /* Static file serving │ -│ (Next.js export) │ +│ (Angular build) │ │ │ │ SQLite database (volume-mounted) │ │ Background task: market data polling/sim │ └─────────────────────────────────────────────────┘ ``` -- **Frontend**: Next.js with TypeScript, built as a static export (`output: 'export'`), served by FastAPI as static files +- **Frontend**: Angular with TypeScript, built via the Angular CLI (`ng build`) into static assets, served by FastAPI as static files - **Backend**: FastAPI (Python), managed as a `uv` project - **Database**: SQLite, single file at `db/finally.db`, volume-mounted for persistence - **Real-time data**: Server-Sent Events (SSE) — simpler than WebSockets, one-way server→client push, works everywhere -- **AI integration**: LiteLLM → OpenRouter (Cerebras for fast inference), with structured outputs for trade execution +- **AI integration**: LiteLLM → OpenRouter (free-tier model), with structured outputs for trade execution - **Market data**: Environment-variable driven — simulator by default, real data via Massive API if key provided ### Why These Choices @@ -74,7 +74,7 @@ The user runs a single Docker command (or a provided start script). A browser op | 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 | +| Static Angular build | 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 | @@ -86,7 +86,7 @@ The user runs a single Docker command (or a provided start script). A browser op ``` finally/ -├── frontend/ # Next.js TypeScript project (static export) +├── frontend/ # Angular TypeScript project (static build) ├── backend/ # FastAPI uv project (Python) │ └── db/ # Schema definitions, seed data, migration logic ├── planning/ # Project-wide documentation for agents @@ -108,7 +108,7 @@ finally/ ### Key Boundaries -- **`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. +- **`frontend/`** is a self-contained Angular 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. - **`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. @@ -156,13 +156,14 @@ Both the simulator and the Massive client implement the same abstract interface. - Starts from realistic seed prices (e.g., AAPL ~$190, GOOGL ~$175, etc.) - Runs as an in-process background task — no external dependencies -### Massive API (Optional) +### Massive API (Optional, not used in this project) - REST API polling (not WebSocket) — simpler, works on all tiers - Polls for the union of all watched tickers on a configurable interval - 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 +- This client is already implemented (see `planning/MARKET_DATA_SUMMARY.md`) and kept as a working alternative behind the `MASSIVE_API_KEY` switch, but this project intentionally runs on the simulator only — `MASSIVE_API_KEY` stays unset. No further work should target Massive-specific behavior (e.g. adapting the UI to its slower polling cadence). ### Shared Price Cache @@ -178,6 +179,7 @@ Both the simulator and the Massive client implement the same abstract interface. - 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 - Each SSE event contains ticker, price, previous price, timestamp, and change direction - Client handles reconnection automatically (EventSource has built-in retry) +- Watchlist changes (`POST`/`DELETE /api/watchlist/*`, or an LLM-initiated change) take effect on the existing, already-open SSE connection immediately — the data source's `add_ticker`/`remove_ticker` updates the shared `PriceCache`, which the stream reads from on every tick. The frontend does not need to reconnect when the watchlist changes. --- @@ -191,6 +193,8 @@ The backend checks for the SQLite database on startup (or first request). If the - No manual database setup - Fresh Docker volumes start with a clean, seeded database automatically +Given the single-user scope, a single SQLite connection with straightforward sequential writes is sufficient — no WAL mode, connection pool, or write-queue is required. This is a deliberate simplification, not an oversight. + ### Schema 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. @@ -215,6 +219,8 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - `avg_cost` REAL - `updated_at` TEXT (ISO timestamp) - UNIQUE constraint on `(user_id, ticker)` +- **Buys**: `avg_cost` is recalculated as the weighted average of the existing position and the new fill (`(old_qty * old_avg_cost + fill_qty * fill_price) / (old_qty + fill_qty)`). +- **Sells**: `avg_cost` is unchanged (only realizes P&L against the existing basis); `quantity` decreases. When a sell brings `quantity` to exactly 0, the position row is deleted rather than kept at zero. **trades** — Trade history (append-only log) - `id` TEXT PRIMARY KEY (UUID) @@ -230,6 +236,7 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - `user_id` TEXT (default: `"default"`) - `total_value` REAL - `recorded_at` TEXT (ISO timestamp) +- No retention/pruning logic is needed — this is a short-lived demo app, not a long-running service, so unbounded row growth is not a concern in scope. **chat_messages** — Conversation history with LLM - `id` TEXT PRIMARY KEY (UUID) @@ -281,7 +288,7 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod ## 9. LLM Integration -When writing code to make calls to LLMs, use cerebras-inference skill to use LiteLLM via OpenRouter to the `openrouter/openai/gpt-oss-120b` model with Cerebras as the inference provider. Structured Outputs should be used to interpret the results. +When writing code to make calls to LLMs, use LiteLLM via OpenRouter to the `openrouter/nvidia/nemotron-3-ultra-550b-a55b:free` model — a free-tier model, chosen to keep the project runnable without incurring API costs. Structured Outputs should be used to interpret the results. There is an OPENROUTER_API_KEY in the .env file in the project root. @@ -292,11 +299,11 @@ 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 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 +4. Calls the LLM via LiteLLM → OpenRouter, requesting structured output 5. Parses the complete structured JSON response 6. Auto-executes any trades or watchlist changes specified in the response 7. Stores the message and executed actions in `chat_messages` -8. Returns the complete JSON response to the frontend (no token-by-token streaming — Cerebras inference is fast enough that a loading indicator is sufficient) +8. Returns the complete JSON response to the frontend (no token-by-token streaming — a loading indicator is shown while awaiting the response) ### Structured Output Schema @@ -315,9 +322,16 @@ The LLM is instructed to respond with JSON matching this schema: ``` - `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) +- `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). Concretely: there is exactly one internal `execute_trade()` function, and both `POST /api/portfolio/trade` and the LLM chat action-execution step call it — the chat handler must not reimplement trade validation - `watchlist_changes` (optional): Array of watchlist modifications +### Structured Output Fallback + +Free-tier OpenRouter models can have inconsistent support for JSON-schema-constrained structured outputs / tool calling. If a structured-output call fails (API error, malformed/non-conforming JSON, or the model ignoring the schema): +1. Retry once with a prompt-based approach — include the schema and an example in the prompt text itself, and ask for JSON-only output +2. Attempt to parse/repair the response (e.g., extract the first valid JSON object from the text) +3. If both attempts fail, return a plain-text error message to the user (no trades/watchlist changes executed) rather than crashing the request + ### Auto-Execution Trades specified by the LLM execute automatically — no confirmation dialog. This is a deliberate design choice: @@ -364,7 +378,7 @@ The frontend is a single-page application with a dense, terminal-inspired layout ### Technical Notes - Use `EventSource` for SSE connection to `/api/stream/prices` -- Canvas-based charting library preferred (Lightweight Charts or Recharts) for performance +- **Charting library: ECharts (via `ngx-echarts`)** for all three visualizations — sparklines/main price chart, the P&L line chart, and the portfolio heatmap (`treemap` series). One dependency instead of separate line-chart and treemap libraries, canvas-rendered for performance - Price flash effect: on receiving a new price, briefly apply a CSS class with background color transition, then remove it - All API calls go to the same origin (`/api/*`) — no CORS configuration needed - Tailwind CSS for styling with a custom dark theme @@ -378,7 +392,7 @@ The frontend is a single-page application with a dense, terminal-inspired layout ``` Stage 1: Node 20 slim - Copy frontend/ - - npm install && npm run build (produces static export) + - npm install && npm run build (Angular CLI production build, produces static assets in dist/) Stage 2: Python 3.12 slim - Install uv @@ -391,6 +405,14 @@ Stage 2: Python 3.12 slim FastAPI serves the static frontend files and all API routes on port 8000. +### Local Development + +Rebuilding the Docker image on every code change is too slow for iteration. During development, run frontend and backend as two separate processes instead: + +- **Backend**: `uv run uvicorn app.main:app --reload --port 8000` from `backend/` — serves `/api/*` and `/api/stream/*` with hot reload. +- **Frontend**: `ng serve` from `frontend/` — serves the Angular dev server (default port 4200) with a proxy config (`proxy.conf.json`) forwarding `/api/*` requests to `http://localhost:8000`, so the frontend code always calls the same-origin-relative `/api/*` paths in both dev and production. +- The production Docker build (this section) remains the target for the final `ng build` + FastAPI static-file-serving setup; local dev never needs to go through Docker. + ### Docker Volume The SQLite database persists via a named Docker volume: @@ -433,7 +455,7 @@ The container is designed to deploy to AWS App Runner, Render, or any container - 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)**: +**Frontend (Jasmine/Karma via Angular CLI, or Angular Testing Library)**: - Component rendering with mock data - Price flash animation triggers correctly on price changes - Watchlist CRUD operations diff --git a/planning/review.md b/planning/review.md new file mode 100644 index 000000000..36d0218ac --- /dev/null +++ b/planning/review.md @@ -0,0 +1,170 @@ +# Change Review — since last commit (`14550e1 Ready for Teams`) + +Reviewed: 2026-08-22 + +## Scope + +Nothing is staged or committed — this reviews the working tree only. No +application code (`backend/app/`, no `frontend/`) changed. The diff is +entirely documentation + new agent-orchestration tooling. + +**Modified (tracked):** +- `planning/PLAN.md` (+50/-16) +- `README.md` (near-total rewrite) +- `.gitignore` (+5) +- `.claude/skills/cerebras/SKILL.md` (+2) + +**Untracked (new):** +- `.claude-plugin/marketplace.json` +- `.claude/agents/reviewer.md` +- `.claude/commands/doc-review.md` +- `independent-reviewer/.claude-plugin/plugin.json` +- `independent-reviewer/hooks/hooks.json` +- `planning/review.md` (this file — regenerated each run, not itself a review subject) + +Verified independently: +- `cd backend && uv run pytest -q` → **73 passed**. +- `cd backend && uv run pytest --cov=app --cov-report=term-missing -q` → **91% overall**. Note this is not evenly distributed: `app/market/stream.py` sits at **33%** coverage (lines 26-48, 62-87 uncovered) while everything else is 94-100%. Pre-existing, not touched by this diff, but worth flagging since the README leans on the aggregate "fully tested" framing. +- No `db/` directory exists in the current working tree or on `main`'s history. `db/.gitkeep` and `.env.example` *do* exist, but only on unrelated branches (`basic`, `codex`, etc.) that are **not ancestors of `main`** (confirmed via `git merge-base --is-ancestor`) — so on this branch's actual history, both files have genuinely never existed. Refining last review's phrasing: not "never existed in this repo" but "never existed on `main`." +- `LICENSE` exists at repo root, tracked since the initial commit (`8e9bb6b`). +- `.claude/settings.json` → `enabledPlugins` lists only `frontend-design`, `context7`, `playwright`. `independent-reviewer` is not in that list. + +--- + +## `planning/PLAN.md` + +Brings the spec in line with decisions apparently made elsewhere (Angular +instead of Next.js, a free-tier OpenRouter model instead of Cerebras, ECharts +instead of Lightweight Charts/Recharts) and adds several genuinely useful +clarifications: SQLite single-connection rationale, buy/sell `avg_cost` math, +SSE watchlist-change semantics, structured-output fallback strategy, and a +local-dev workflow subsection. No internal contradictions found — checked +that no stray Cerebras/`gpt-oss-120b` references remain anywhere in the file. + +**Issue:** +- §4's directory listing still states `db/.gitkeep` "Directory exists in + repo; finally.db is gitignored" — this is false on `main` (verified above). + The new `.gitignore` rules added in this same diff (`db/*.db`, + `!db/.gitkeep`) implicitly assume this file exists. Worth creating `db/` + + `db/.gitkeep` in the same pass, since it's a one-line fix and the diff + otherwise treats this as settled. + +**Minor, non-blocking:** +- §9's model id `openrouter/nvidia/nemotron-3-ultra-550b-a55b:free` can't be + verified from this repo — free-tier OpenRouter slugs rotate, worth a sanity + check before backend LLM work starts. +- The new "single `execute_trade()`" constraint (§9, shared by + `POST /api/portfolio/trade` and the chat action executor) is a good, + testable rule — worth enforcing explicitly in code review once backend + trade logic lands. + +## `README.md` + +Rewrite is a real improvement — the old version implied a working +`docker build`/`docker run` quick start that doesn't exist yet (no +Dockerfile, no `frontend/`). New version's status table, directory layout, +and commands all check out against actual repo state, with one exception: + +**Issue:** +- "It's fully tested (73 tests, 84% coverage)" — the 73 is correct, but 84% + is stale. Actual current coverage running the suite is **91%** overall, + though that average masks `app/market/stream.py` at 33% (see Scope + section). The 84% figure matches `planning/MARKET_DATA_SUMMARY.md` (line + 58), which appears to predate later test additions; the README borrowed a + number that was already out of date. One-line fix, or drop the percentage + and cite the test count alone to avoid future drift — and if precision + matters, call out the `stream.py` gap rather than only the headline number. + +No other issues found — `LICENSE` link is valid, layout matches actual +`backend/app/market/` structure, `uv sync --dev` / `uv run pytest` / +`uv run market_data_demo.py` commands match `backend/README.md`. + +## `.gitignore` + +``` +db/*.db +db/*.db-journal +!db/.gitkeep +``` +Correct and harmless in isolation (the negation pattern would preserve a +future `db/.gitkeep` while ignoring `db/finally.db`), but as noted above, +it's added to support a file/directory that doesn't exist yet on `main` — so +right now this rule has nothing to protect. Not wrong, just premature; tie it +to actually creating `db/.gitkeep`. + +## `.claude/skills/cerebras/SKILL.md` + +Adds a clear callout that this skill must not be used for FinAlly's chat +feature, pointing at PLAN.md §9. Good guardrail now that PLAN.md dropped +Cerebras as the mandated provider — prevents a future agent from following +stale skill guidance over current project spec. No issues. + +--- + +## New agent tooling: `independent-reviewer` plugin, `.claude/agents/reviewer.md`, `.claude/commands/doc-review.md` + +This is the meta-tooling that produces this very review. Findings for whoever +owns this setup: + +1. **Plugin not enabled.** `independent-reviewer` (registered via + `.claude-plugin/marketplace.json`) is absent from + `.claude/settings.json`'s `enabledPlugins`. If plugin hooks only fire for + enabled plugins, the Stop hook in `independent-reviewer/hooks/hooks.json` + won't run automatically until it's added there. Confirm this is + intentional staging rather than an oversight. +2. **Plugin isn't self-contained.** `independent-reviewer/` only has + `.claude-plugin/plugin.json` and `hooks/hooks.json` — no `agents/` or + `commands/` of its own. The actual `change-reviewer` agent lives at the + project level (`.claude/agents/reviewer.md`), and the hook depends on it + being present there. Enabling the plugin alone in a different repo (or + without these project-level files) would leave the hook invoking an agent + that doesn't exist. Consider moving the agent definition inside the + plugin directory if portability matters. +3. **Recursion guard looks correct.** The hook checks + `[ -z "$CLAUDE_STOP_HOOK_ACTIVE" ]` before invoking + `claude --agent change-reviewer` and sets the env var when it does, which + should prevent the reviewer's own Stop event from re-triggering itself. +4. **Naming mismatch (cosmetic).** Hook invokes `--agent change-reviewer`; + the agent is defined in a file named `reviewer.md` (not + `change-reviewer.md`) — wired correctly via the `name:` frontmatter field, + but easy to trip over when grepping for the agent by filename. +5. **`doc-review` slash command's relationship to this plugin is unclear.** + `.claude/commands/doc-review.md` reviews a single planning doc in place + (adds a feedback section) — a different workflow from `change-reviewer`'s + "review everything since last commit → `planning/review.md`" job. Both + were added in the same diff; worth confirming these are two intentionally + separate tools rather than one meant to replace the other. +6. **Description text is duplicated across three files** (agent frontmatter, + `marketplace.json`, `plugin.json`), all saying essentially "review changes + since last commit." Not a bug, just three places to keep in sync if scope + changes later. +7. **`planning/review.md` is self-overwriting output, not reviewable input.** + Since this file is regenerated by the very agent this review evaluates, + treat its own historical diffs as tooling exhaust rather than content to + scrutinize — noted here only so a future run doesn't waste effort + re-reviewing its own prior output. + +None of these block anything — this tooling is additive and doesn't touch +the application. + +--- + +## Overall + +No application code changed. Documentation changes are accurate and +internally consistent, with two small, non-blocking staleness issues: + +1. README's "84% coverage" is stale (actual: 91% overall, with a real + coverage gap in `app/market/stream.py` at 33% that the headline number + obscures; test count of 73 is correct). +2. `.gitignore`'s new `!db/.gitkeep` rule, and PLAN.md §4's claim that + `db/.gitkeep` already exists in the repo, both refer to a file that has + never existed on `main` (it exists only on unrelated, non-ancestor + branches) — a pre-existing gap this diff doesn't introduce but also + doesn't close, despite touching the adjacent `.gitignore` rule in the + same commit. + +The new `independent-reviewer` plugin/agent tooling is functionally sound +(correct recursion guard, correctly wired agent name) but is not yet enabled +in `.claude/settings.json` and is not fully self-contained within the plugin +directory. From 91c4696208da49a1b4e4234c34379003d3205b31 Mon Sep 17 00:00:00 2001 From: Raunak Sachdev Date: Sun, 23 Aug 2026 09:56:47 +0100 Subject: [PATCH 2/7] design for the python code --- .claude-plugin/marketplace.json | 8 -- planning/MARKET_DATA_API.md | 220 ++++++++++++++++++++++++++++ planning/MARKET_SUMULATOR.md | 221 ++++++++++++++++++++++++++++ planning/MASSIVE_API.md | 246 ++++++++++++++++++++++++++++++++ 4 files changed, 687 insertions(+), 8 deletions(-) delete mode 100644 .claude-plugin/marketplace.json create mode 100644 planning/MARKET_DATA_API.md create mode 100644 planning/MARKET_SUMULATOR.md create mode 100644 planning/MASSIVE_API.md diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json deleted file mode 100644 index 80b4e20da..000000000 --- a/.claude-plugin/marketplace.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "name": "raunak-tools", - "owner": { - "name": "Raunak Sachdev", - "email": "sachdevraunak1991@gmail.com" - }, - "plugins": [] -} diff --git a/planning/MARKET_DATA_API.md b/planning/MARKET_DATA_API.md new file mode 100644 index 000000000..e65a10e13 --- /dev/null +++ b/planning/MARKET_DATA_API.md @@ -0,0 +1,220 @@ +# Unified Market Data API — Design + +This is the interface design for retrieving stock prices in FinAlly: one +abstraction with two implementations — a live client against the Massive +API (see `planning/MASSIVE_API.md`) and a built-in simulator (see +`planning/MARKET_SUMULATOR.md`) — selected at startup by whether +`MASSIVE_API_KEY` is set. This document describes the interface as already +implemented in `backend/app/market/`; treat it as the contract reference +for that package. + +## 1. Goals + +- Downstream code (SSE stream, portfolio valuation, trade execution) never + knows or cares whether prices come from Massive or the simulator. +- Switching sources is a pure environment-variable toggle — no code change, + no restart-time branching beyond the factory. +- A single shared, thread-safe cache is the only thing downstream code reads + from; data sources only ever write to it. + +## 2. Component diagram + +``` + MASSIVE_API_KEY set? + │ + ┌────────────────┴────────────────┐ + │ yes │ no + ▼ ▼ +MassiveDataSource SimulatorDataSource +(REST poller, Massive API) (GBM simulator, in-process) + │ │ + └────────────────┬─────────────────┘ + ▼ + PriceCache + (thread-safe, in-memory) + │ + ┌───────────────┼───────────────┐ + ▼ ▼ ▼ + SSE stream Portfolio Trade execution + (/api/stream/ valuation (fill price) + prices) +``` + +Both implementations conform to the same `MarketDataSource` abstract base +class, so the factory function is the only place that knows both classes +exist. + +## 3. Core types + +### 3.1 `PriceUpdate` (`app/market/models.py`) + +An immutable snapshot of one ticker's price at a point in time. Frozen so it +can be shared across threads/tasks without defensive copying. + +```python +@dataclass(frozen=True, slots=True) +class PriceUpdate: + ticker: str + price: float + previous_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: ... # price - previous_price + @property + def change_percent(self) -> float: ... # % change vs previous_price + @property + def direction(self) -> str: ... # "up" | "down" | "flat" + + def to_dict(self) -> dict: ... # JSON-serializable for SSE +``` + +### 3.2 `MarketDataSource` (`app/market/interface.py`) + +The abstract contract both implementations satisfy. + +```python +class MarketDataSource(ABC): + @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. + Call exactly once.""" + + @abstractmethod + async def stop(self) -> None: + """Stop the background task. 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. + Takes effect on the next update cycle.""" + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Remove a ticker from the active set and from the PriceCache.""" + + @abstractmethod + def get_tickers(self) -> list[str]: + """Currently tracked tickers.""" +``` + +Design choices baked into this contract: +- **Async lifecycle, sync accessor** — `start`/`stop`/`add_ticker`/`remove_ticker` + are async because both implementations do I/O or task management on those + paths (spawning a poller, spawning the sim loop); `get_tickers()` is sync + because it's just a local list read. +- **No `get_price()` on the interface** — price reads always go through + `PriceCache`, never through the source. This keeps "who writes" (sources) + and "who reads" (everyone else) strictly separated and means adding a + third data source later requires zero changes to any reader. +- **Idempotent `stop()`, no-op `add_ticker`/`remove_ticker` on duplicates** — + callers (route handlers, chat action execution) don't need to + pre-check state before calling. + +### 3.3 `PriceCache` (`app/market/cache.py`) + +The single point of truth. Producers (one at a time — either +`SimulatorDataSource` or `MassiveDataSource`, never both) write; every +reader in the app reads from here, never from the source directly. + +```python +class PriceCache: + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: ... + def get(self, ticker: str) -> PriceUpdate | None: ... + def get_price(self, ticker: str) -> float | None: ... + def get_all(self) -> dict[str, PriceUpdate]: ... + def remove(self, ticker: str) -> None: ... + + @property + def version(self) -> int: ... # monotonically increasing, bumped on every update +``` + +- Guarded by a `threading.Lock` — safe even though writers and the FastAPI + event loop can interleave (`MassiveDataSource` moves its synchronous HTTP + call to a thread via `asyncio.to_thread`). +- `version` exists purely so the SSE endpoint can cheaply detect "did + anything change since I last sent a frame" without diffing the whole + price dict every tick. +- `update()` computes `previous_price` internally from whatever was cached + before — callers only ever supply the new price; this is what makes + `direction`/`change` correct without every writer duplicating that logic. + +### 3.4 Factory (`app/market/factory.py`) + +The only place that imports both concrete implementations and the only +place that reads `MASSIVE_API_KEY`. + +```python +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """MASSIVE_API_KEY set and non-empty -> MassiveDataSource. + Otherwise -> SimulatorDataSource. + Returns an unstarted source; caller must await source.start(tickers).""" + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + if api_key: + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + return SimulatorDataSource(price_cache=price_cache) +``` + +Trimming and checking for non-empty (not just presence) matters: an `.env` +file with `MASSIVE_API_KEY=` (present but blank) must fall back to the +simulator, not attempt a poll with an empty key. + +## 4. Implementations + +### 4.1 `SimulatorDataSource` — default + +Runs an in-process asyncio task that steps a GBM model every ~500ms and +writes results straight into `PriceCache`. No network calls, no external +dependency, no API key. Full design in `planning/MARKET_SUMULATOR.md`. + +### 4.2 `MassiveDataSource` — optional, when `MASSIVE_API_KEY` is set + +Polls Massive's multi-ticker snapshot endpoint (`get_snapshot_all`) on a +timer — 15s on the free tier, 2–5s on paid tiers — fetching all watched +tickers in a single REST call, then writes `last_trade.price` / +`last_trade.timestamp` into the cache. Full research and code examples in +`planning/MASSIVE_API.md`. + +Per `planning/PLAN.md` §6 and `CLAUDE.md`, this project deliberately ships +with `MASSIVE_API_KEY` unset — the simulator is the supported path — and +this implementation is kept working but not actively extended. + +## 5. Wiring into the app + +```python +from app.market import PriceCache, create_market_data_source, create_stream_router + +# App startup +price_cache = PriceCache() +market_source = create_market_data_source(price_cache) +await market_source.start(initial_watchlist_tickers) + +app.include_router(create_stream_router(price_cache)) + +# Watchlist mutation (REST route or LLM tool call) — same call either way +await market_source.add_ticker("TSLA") +await market_source.remove_ticker("GOOGL") + +# App shutdown +await market_source.stop() +``` + +`create_stream_router(price_cache)` (`app/market/stream.py`) builds the SSE +endpoint (`GET /api/stream/prices`) as a closure over the cache, so the +endpoint has no dependency on which `MarketDataSource` is active. The +streaming loop polls `price_cache.version` every 500ms and only emits a +frame when it has advanced — this is what lets `add_ticker`/`remove_ticker` +apply to an already-open SSE connection without a reconnect: the next tick +after the cache changes just includes (or drops) that ticker. + +Portfolio valuation and trade execution read current prices the same way — +`price_cache.get_price(ticker)` — so they too are agnostic to the data +source. + +## 6. Extending this later + +Adding a third source (a different vendor, a WebSocket-based feed, etc.) +requires only: implement `MarketDataSource`, and extend the factory's +branch on the relevant env var. No other module changes, because every +consumer already goes through `PriceCache`. diff --git a/planning/MARKET_SUMULATOR.md b/planning/MARKET_SUMULATOR.md new file mode 100644 index 000000000..c34d839c6 --- /dev/null +++ b/planning/MARKET_SUMULATOR.md @@ -0,0 +1,221 @@ +# Market Simulator — Approach & Code Structure + +The default market data source (no `MASSIVE_API_KEY` required). Generates +realistic, correlated, live-updating stock prices in-process. This document +describes the design implemented in `backend/app/market/simulator.py` and +`backend/app/market/seed_prices.py`, and how it plugs into the unified +interface described in `planning/MARKET_DATA_API.md`. + +## 1. Why a simulator + +Per `planning/PLAN.md` §6, the simulator is the primary, supported data +source for this project — not a fallback bolted on for when a paid API key +is missing. It needs no external dependency, no network calls, no rate +limits, and produces the ~500ms-cadence price action the frontend's flash +animations and sparklines are built around, which a 15s-polled free-tier +Massive feed cannot deliver on its own. + +## 2. The model: Geometric Brownian Motion + +Each ticker's price evolves under GBM: + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +Where: +- `S(t)` — current price +- `mu` — annualized drift (expected return) +- `sigma` — annualized volatility +- `dt` — time step, expressed as a fraction of a trading year +- `Z` — a (correlated) standard normal random draw + +### Choosing `dt` + +Ticks happen every 500ms, but `mu`/`sigma` are annualized, so `dt` has to +convert "half a second" into "fraction of a trading 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` is what keeps individual ticks to realistic sub-cent moves +that accumulate into believable multi-minute price action, rather than +each tick looking like a full day's move. + +### Correlated moves via Cholesky decomposition + +Real markets don't move ticker-by-ticker independently — tech stocks tend +to move together, as do financials. To reproduce that, draw `n` independent +standard normals, then multiply by the Cholesky factor of a correlation +matrix built from sector groupings: + +```python +z_independent = np.random.standard_normal(n) +z_correlated = cholesky_factor @ z_independent # correlated draws +``` + +Correlation structure (`app/market/seed_prices.py`): + +| Pair | Correlation | +|---|---| +| Two tech tickers (AAPL, GOOGL, MSFT, AMZN, META, NVDA, NFLX) | 0.6 | +| Two finance tickers (JPM, V) | 0.5 | +| Either ticker is TSLA | 0.3 (TSLA "does its own thing") | +| Cross-sector / unknown ticker | 0.3 | + +The correlation matrix — and its Cholesky factor — is rebuilt whenever a +ticker is added or removed (O(n²), fine for n < ~50 watchlist tickers). + +### Random shock events + +Independently of the GBM step, each ticker has a small per-tick chance +(`event_probability`, default 0.1%) of a sudden 2–5% move in either +direction — this is what produces the occasional dramatic single-ticker +spike/drop for visual interest, layered on top of the smooth GBM walk. At +10 tickers and 2 ticks/sec, expect roughly one event every ~50 seconds. + +```python +if random.random() < self._event_prob: + shock_magnitude = random.uniform(0.02, 0.05) + shock_sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock_magnitude * shock_sign +``` + +## 3. Seed data (`app/market/seed_prices.py`) + +Starting prices and per-ticker GBM parameters for the default watchlist — +volatility (`sigma`) and drift (`mu`) are hand-tuned per ticker to feel +representative (e.g. TSLA/NVDA high-vol, JPM/V low-vol): + +```python +SEED_PRICES = { + "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 = { + "AAPL": {"sigma": 0.22, "mu": 0.05}, + "TSLA": {"sigma": 0.50, "mu": 0.03}, # high volatility + "NVDA": {"sigma": 0.40, "mu": 0.08}, # high volatility, strong drift + "JPM": {"sigma": 0.18, "mu": 0.04}, # low volatility (bank) + # ... +} + +DEFAULT_PARAMS = {"sigma": 0.25, "mu": 0.05} # used for dynamically added tickers +``` + +A ticker added later that isn't in `SEED_PRICES`/`TICKER_PARAMS` (e.g. a +user adds an arbitrary symbol via the watchlist or chat) gets a random seed +price in `[$50, $300]` and `DEFAULT_PARAMS` — it still participates fully in +the simulation, just without hand-tuned realism. + +## 4. Code structure + +### 4.1 `GBMSimulator` — pure simulation state, no I/O + +The math and per-ticker state live in a plain class with no asyncio, no +cache, no network — this keeps the numerically interesting part unit +testable in isolation. + +```python +class GBMSimulator: + def __init__(self, tickers: list[str], dt: float = DEFAULT_DT, + event_probability: float = 0.001) -> None: ... + + def step(self) -> dict[str, float]: + """Advance all tickers by one time step. Hot path — called every + 500ms, so keep it allocation-light.""" + + def add_ticker(self, ticker: str) -> None: + """Add a ticker; rebuilds the Cholesky decomposition.""" + + def remove_ticker(self, ticker: str) -> None: + """Remove a ticker; rebuilds the Cholesky decomposition.""" + + def get_price(self, ticker: str) -> float | None: ... + def get_tickers(self) -> list[str]: ... +``` + +`step()` returns `{ticker: new_price}` for every tracked ticker on every +call — the caller decides what to do with that (write to a cache, print to +a terminal demo, feed a test assertion). + +### 4.2 `SimulatorDataSource` — the `MarketDataSource` adapter + +Wraps `GBMSimulator` in the async lifecycle the unified interface expects +(see `planning/MARKET_DATA_API.md` §3.2), and owns writing results into the +shared `PriceCache`. + +```python +class SimulatorDataSource(MarketDataSource): + def __init__(self, price_cache: PriceCache, + update_interval: float = 0.5, + event_probability: float = 0.001) -> None: ... + + async def start(self, tickers: list[str]) -> None: + """Build the GBMSimulator, seed the cache immediately (so SSE has + data on the very first frame), then spawn the update loop task.""" + + async def stop(self) -> None: + """Cancel the update loop task; safe to call more than once.""" + + async def add_ticker(self, ticker: str) -> None: + """Delegate to GBMSimulator.add_ticker, then seed the cache with + its starting price immediately rather than waiting for the next + tick.""" + + async def remove_ticker(self, ticker: str) -> None: + """Delegate to GBMSimulator.remove_ticker, then remove from cache.""" + + def get_tickers(self) -> list[str]: ... + + async def _run_loop(self) -> None: + """while True: step the simulator, write every result to the + cache, sleep(update_interval). Wrapped in try/except so one bad + step (should never happen, but) doesn't kill the background task — + it logs and continues on the next tick.""" +``` + +Two "seed immediately" touches matter for UX: on `start()` and on +`add_ticker()`, the cache gets a value before the first scheduled tick +fires, so a newly opened SSE connection or a freshly added watchlist ticker +never shows as blank/missing for up to 500ms. + +## 5. Example usage + +```python +from app.market import PriceCache, create_market_data_source + +cache = PriceCache() +source = create_market_data_source(cache) # SimulatorDataSource, since + # MASSIVE_API_KEY is unset +await source.start(["AAPL", "GOOGL", "MSFT", "TSLA"]) + +# ... 500ms later ... +update = cache.get("TSLA") +print(update.price, update.direction, update.change_percent) + +await source.add_ticker("NVDA") # immediately visible in the cache +await source.remove_ticker("MSFT") + +await source.stop() +``` + +A standalone terminal visualization of this exact simulator is available at +`backend/market_data_demo.py` (`uv run market_data_demo.py` from +`backend/`) — a live Rich dashboard with sparklines, direction arrows, and +an event log, useful for eyeballing whether tuning changes to `sigma`/`mu` +or the correlation groups still feel realistic. + +## 6. Testing notes + +Because `GBMSimulator` has no I/O, its statistical properties (drift over +many steps trends toward `mu`, correlated tickers actually correlate, an +added/removed ticker rebuilds the Cholesky matrix without crashing) are +directly testable with straightforward `pytest` assertions over many +`step()` calls — see `backend/tests/market/test_simulator.py` for the +existing suite (17 tests, 98% coverage per +`planning/MARKET_DATA_SUMMARY.md`). diff --git a/planning/MASSIVE_API.md b/planning/MASSIVE_API.md new file mode 100644 index 000000000..850081f6d --- /dev/null +++ b/planning/MASSIVE_API.md @@ -0,0 +1,246 @@ +# Massive API Research (formerly Polygon.io) + +Research notes on the Massive.com API for retrieving real-time and end-of-day +prices for multiple tickers, as used by `MassiveDataSource` in +`backend/app/market/massive_client.py`. This is background/reference +documentation — per `CLAUDE.md`, FinAlly intentionally runs on the built-in +simulator by default and does not currently target further Massive-specific +work, but this doc keeps the integration accurate if `MASSIVE_API_KEY` is +ever set. + +## 1. The rebrand + +Polygon.io rebranded to **Massive** on **October 30, 2025**. Existing API +keys, accounts, and integrations continued to work without interruption. +The REST surface stayed at `api.polygon.io`-compatible routes, now also +served from `api.massive.com`; the Python package was renamed from +`polygon-api-client` to **`massive`** on PyPI. +[Source: Polygon.io is Now Massive](https://massive.com/blog/polygon-is-now-massive) + +## 2. Installation & authentication + +```bash +pip install -U massive +# or, in this project: +uv add massive +``` + +Requires Python 3.9+. + +```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") +``` + +Auth is a bearer token derived from the API key; the client attaches it to +every request automatically — callers never build the `Authorization` +header by hand. +[Source: massive-com/client-python README](https://github.com/massive-com/client-python/blob/master/README.md) + +## 3. Rate limits & plans + +| Plan | Requests | Data | +|---|---|---| +| Free | 5 requests/minute | 15-minute delayed | +| Starter / Developer / Advanced / Business (paid, from $199/mo) | No hard cap — Massive monitors usage rather than enforcing a fixed ceiling; stay under ~100 req/s to avoid throttling | Real-time or 15-min delayed depending on tier | + +[Source: What is the request limit for Massive's RESTful APIs?](https://massive.com/knowledge-base/article/what-is-the-request-limit-for-massives-restful-apis) · +[Source: Massive Pricing](https://massive.com/pricing) + +This confirms the polling cadence already specified in `planning/PLAN.md` §6 +and implemented in `MassiveDataSource`: **free tier → poll every 15s** +(5 req/min ceiling with margin), **paid tiers → poll every 2–5s**. + +## 4. Endpoints relevant to multi-ticker real-time + EOD prices + +### 4.1 Full Market / Multi-Ticker Snapshot (primary endpoint used by this project) + +Returns a snapshot (last trade, last quote, day OHLC, previous day OHLC) for +many tickers in **one API call** — this is what makes polling on the free +tier viable, since watching 10 tickers still costs only 1 request. + +``` +GET /v2/snapshot/locale/us/markets/stocks/tickers?tickers=AAPL,GOOGL,MSFT +``` + +Query parameters: +- `tickers` (optional, comma-separated, case-sensitive) — restrict to specific symbols, e.g. `AAPL,TSLA,GOOG`. Omitting it returns the entire market (10,000+ tickers). +- `include_otc` (optional, bool) — include OTC securities; default `false`. + +Snapshot data resets daily at 3:30 AM ET and starts repopulating as +exchanges report, from as early as 4:00 AM ET. + +Response shape (per ticker, camelCase over the wire): + +```json +{ + "ticker": "AAPL", + "day": { "o": 129.61, "h": 130.15, "l": 125.07, "c": 125.07, "v": 111237700, "vw": 127.35 }, + "prevDay": { "o": 128.4, "h": 129.95, "l": 127.8, "c": 129.61, "v": 98765400, "vw": 128.9 }, + "lastTrade": { "p": 125.07, "s": 100, "x": 11, "t": 1675190399000 }, + "lastQuote": { "p": 125.06, "P": 125.08, "s": 500, "S": 1000, "t": 1675190399500 }, + "min": { "o": 125.0, "h": 125.1, "l": 124.95, "c": 125.07, "v": 12000 }, + "todaysChange": -4.54, + "todaysChangePerc": -3.50, + "updated": 1675190399500000000 +} +``` + +Python client (models expose the same fields as snake_case attributes): + +```python +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +client = RESTClient() + +snapshots = client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"], +) + +for snap in snapshots: + print(f"{snap.ticker}: ${snap.last_trade.price}") + print(f" Day change: {snap.day.change_percent}%") + print(f" Day OHLC: O={snap.day.open} H={snap.day.high} L={snap.day.low} C={snap.day.close}") + print(f" Prev close: {snap.prev_daily_bar.close}") +``` + +Requires Starter plan or above (not available on the bare free tier for +delayed-only access in some configurations — verify against the account's +actual plan before relying on it in a paid deployment). +[Source: Full Market Snapshot docs](https://massive.com/docs/rest/stocks/snapshots/full-market-snapshot) + +### 4.2 Single Ticker Snapshot + +Same shape as above, scoped to one ticker — used for a detail view when a +user clicks a specific ticker. + +```python +snapshot = client.get_snapshot_ticker( + market_type=SnapshotMarketType.STOCKS, + ticker="AAPL", +) +print(f"Price: ${snapshot.last_trade.price}") +print(f"Bid/Ask: ${snapshot.last_quote.bid_price} / ${snapshot.last_quote.ask_price}") +``` +[Source: Single Ticker Snapshot docs](https://massive.com/docs/rest/stocks/snapshots/single-ticker-snapshot) + +### 4.3 Previous Close (end-of-day) + +``` +GET /v2/aggs/ticker/{stocksTicker}/prev +``` + +Returns the prior trading day's OHLCV for one ticker — useful for seeding a +simulator with realistic starting prices, or for computing day-over-day +change independent of the snapshot endpoint. + +```python +prev = client.get_previous_close_agg(ticker="AAPL") +for agg in prev: + print(f"Previous close: ${agg.close} O={agg.open} H={agg.high} L={agg.low} V={agg.volume}") +``` + +Response (raw JSON): +```json +{ + "ticker": "AAPL", + "results": [ + {"o": 150.0, "h": 155.0, "l": 149.0, "c": 154.5, "v": 1000000, "t": 1672531200000} + ] +} +``` +[Source: Previous Day Bar (OHLC) docs](https://massive.com/docs/rest/stocks/aggregates/previous-day-bar) + +### 4.4 Aggregates / Custom Bars (historical, for charts) + +Not needed for live polling, but the natural source for a "main chart" +history view beyond what's accumulated client-side from the SSE stream. + +``` +GET /v2/aggs/ticker/{ticker}/range/{multiplier}/{timespan}/{from}/{to} +``` + +```python +aggs = [] +for a in client.list_aggs( + ticker="AAPL", + multiplier=1, + timespan="day", + from_="2024-01-01", + to="2024-01-31", + limit=50000, +): + aggs.append(a) +``` +[Source: Custom Bars (OHLC) docs](https://massive.com/docs/rest/stocks/aggregates/custom-bars) + +### 4.5 Last Trade / Last Quote (single values) + +```python +trade = client.get_last_trade(ticker="AAPL") +print(f"Last trade: ${trade.price} x {trade.size}") + +quote = client.get_last_quote(ticker="AAPL") +print(f"Bid: ${quote.bid_price} x {quote.bid_size} Ask: ${quote.ask_price} x {quote.ask_size}") +``` +[Source: massive-com/client-python README](https://github.com/massive-com/client-python/blob/master/README.md) + +## 5. Pagination + +The client paginates automatically by default for list-style endpoints +(`list_aggs`, `list_trades`, `list_quotes`), transparently fetching +subsequent pages as the generator is iterated. Disable with +`RESTClient(api_key=..., pagination=False)` if manual paging is preferred. + +## 6. Error handling + +The client raises typed exceptions rather than returning error payloads: +- **401** — invalid or missing API key +- **403** — valid key, but the current plan doesn't include this endpoint/data tier +- **429** — rate limit exceeded (free tier: 5 req/min) +- **5xx** — server error; treat as transient + +`MassiveDataSource._poll_once()` (see `backend/app/market/massive_client.py`) +already wraps each poll cycle in a broad `try/except`, logs the failure, and +lets the next scheduled poll retry — it does not propagate exceptions out of +the background task, so a transient 429 or network blip does not crash the +poller. + +## 7. Timestamps & data-freshness notes + +- All timestamps from the API are **Unix milliseconds** (some fields, like + snapshot `updated`, are nanoseconds — check the specific field before + dividing). +- During closed-market hours, `last_trade.price` reflects the last traded + price and may include after-hours activity. +- The `day` object resets at market open; during pre-market it may still + reflect the previous session until new trades post. + +## 8. How this project uses it + +`MassiveDataSource` (see `backend/app/market/massive_client.py`) polls +`get_snapshot_all()` once per interval for the full watchlist in a single +call, runs the synchronous client in a thread via `asyncio.to_thread` to +avoid blocking the event loop, and writes `last_trade.price` + +`last_trade.timestamp` into the shared `PriceCache`. This is selected by +`create_market_data_source()` (`backend/app/market/factory.py`) only when +`MASSIVE_API_KEY` is non-empty; see `planning/MARKET_DATA_API.md` for the +full interface this implementation conforms to. + +## Sources + +- [Polygon.io is Now Massive](https://massive.com/blog/polygon-is-now-massive) +- [massive-com/client-python README](https://github.com/massive-com/client-python/blob/master/README.md) +- [Full Market Snapshot | Stocks REST API](https://massive.com/docs/rest/stocks/snapshots/full-market-snapshot) +- [Single Ticker Snapshot | Stocks REST API](https://massive.com/docs/rest/stocks/snapshots/single-ticker-snapshot) +- [Previous Day Bar (OHLC) | Stocks REST API](https://massive.com/docs/rest/stocks/aggregates/previous-day-bar) +- [Custom Bars (OHLC) | Stocks REST API](https://massive.com/docs/rest/stocks/aggregates/custom-bars) +- [What is the request limit for Massive's RESTful APIs?](https://massive.com/knowledge-base/article/what-is-the-request-limit-for-massives-restful-apis) +- [Massive Pricing](https://massive.com/pricing) From 8b3860522800b356031b7ecb6f4decf6e29a0c09 Mon Sep 17 00:00:00 2001 From: raunaksachdev Date: Sun, 23 Aug 2026 10:21:28 +0100 Subject: [PATCH 3/7] "Update Claude PR Assistant workflow" --- .github/workflows/claude.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/claude.yml b/.github/workflows/claude.yml index d300267f1..6b15fac7a 100644 --- a/.github/workflows/claude.yml +++ b/.github/workflows/claude.yml @@ -46,5 +46,5 @@ jobs: # Optional: Add claude_args to customize behavior and configuration # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md # or https://code.claude.com/docs/en/cli-reference for available options - # claude_args: '--allowed-tools Bash(gh pr:*)' + # claude_args: '--allowed-tools Bash(gh pr *)' From cba70e7d6e484ecce24aead75d753486cf1b61c5 Mon Sep 17 00:00:00 2001 From: raunaksachdev Date: Sun, 23 Aug 2026 10:21:29 +0100 Subject: [PATCH 4/7] "Update Claude Code Review workflow" --- .github/workflows/claude-code-review.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/claude-code-review.yml b/.github/workflows/claude-code-review.yml index b5e8cfd4d..37e66f3fd 100644 --- a/.github/workflows/claude-code-review.yml +++ b/.github/workflows/claude-code-review.yml @@ -38,7 +38,8 @@ jobs: claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }} plugin_marketplaces: 'https://github.com/anthropics/claude-code.git' plugins: 'code-review@claude-code-plugins' - prompt: '/code-review:code-review ${{ github.repository }}/pull/${{ github.event.pull_request.number }}' + prompt: '/code-review:code-review --comment ${{ github.repository }}/pull/${{ github.event.pull_request.number }}' + claude_args: '--allowedTools "mcp__github_inline_comment__create_inline_comment"' # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md # or https://code.claude.com/docs/en/cli-reference for available options From 327aa453c1705e4239600c03422fed34eac47ac0 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Aug 2026 08:21:26 +0000 Subject: [PATCH 5/7] Add consolidated market data backend design doc Writes planning/MARKET_DATA_DESIGN.md as an implementation-ready reference covering the unified MarketDataSource interface, PriceCache, GBM simulator, Massive API client, SSE streaming, and FastAPI lifecycle wiring. Code snippets are verified against the current backend/app/market/ implementation, correcting drift present in the older planning/archive/MARKET_DATA_DESIGN.md (lazy-import framing for the massive client, GBMSimulator.get_tickers() visibility, stream.py's generator return type). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01RyKr4TS8gMZPNU8roFiYUT --- planning/MARKET_DATA_DESIGN.md | 1557 ++++++++++++++++++++++++++++++++ 1 file changed, 1557 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..89cf5d7e3 --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,1557 @@ +# Market Data Backend — Detailed Design + +Implementation-ready design for the FinAlly market data subsystem: the +unified `MarketDataSource` interface, the GBM simulator (default source), +the Massive (Polygon.io) REST client (optional source), the shared price +cache, the SSE streaming endpoint, and how all of it wires into the FastAPI +app. + +**Status:** this subsystem is already built and tested — see +`planning/MARKET_DATA_SUMMARY.md` for the test/coverage summary. This +document is the implementation-ready reference: every code block below is +taken from (or matches) the real source under `backend/app/market/`, so it +doubles as an onboarding doc and as a spec a fresh implementation could be +rebuilt from. It supersedes `planning/archive/MARKET_DATA_DESIGN.md`, whose +code had drifted slightly from the implementation after later fixes +(notably: `massive_client.py` no longer lazy-imports `massive`, and +`GBMSimulator.get_tickers()` is now a public method). + +Everything below lives under `backend/app/market/`. + +--- + +## Table of Contents + +1. [Goals & Component Diagram](#1-goals--component-diagram) +2. [Data Model — `models.py`](#2-data-model) +3. [Price Cache — `cache.py`](#3-price-cache) +4. [Abstract Interface — `interface.py`](#4-abstract-interface) +5. [Seed Prices & Ticker Parameters — `seed_prices.py`](#5-seed-prices--ticker-parameters) +6. [GBM Simulator — `simulator.py`](#6-gbm-simulator) +7. [Massive API Client — `massive_client.py`](#7-massive-api-client) +8. [Factory — `factory.py`](#8-factory) +9. [SSE Streaming Endpoint — `stream.py`](#9-sse-streaming-endpoint) +10. [FastAPI Lifecycle Integration](#10-fastapi-lifecycle-integration) +11. [Watchlist Coordination](#11-watchlist-coordination) +12. [Testing Strategy](#12-testing-strategy) +13. [Error Handling & Edge Cases](#13-error-handling--edge-cases) +14. [Configuration Summary](#14-configuration-summary) + +--- + +## 1. Goals & Component Diagram + +- Downstream code (SSE stream, portfolio valuation, trade execution) never + knows or cares whether prices come from Massive or the simulator. +- Switching sources is a pure environment-variable toggle (`MASSIVE_API_KEY`) + — no code change, no restart-time branching beyond the factory function. +- A single shared, thread-safe cache is the only thing downstream code reads + from; data sources only ever write to it. + +``` + MASSIVE_API_KEY set (non-empty)? + │ + ┌────────────────┴────────────────┐ + │ yes │ no + ▼ ▼ +MassiveDataSource SimulatorDataSource +(REST poller, Massive API) (GBM simulator, in-process) + │ │ + └────────────────┬─────────────────┘ + ▼ + PriceCache + (thread-safe, in-memory) + │ + ┌───────────────┼───────────────┐ + ▼ ▼ ▼ + SSE stream Portfolio Trade execution + (/api/stream/ valuation (fill price) + prices) +``` + +Both implementations conform to the same `MarketDataSource` abstract base +class (strategy pattern), so `factory.py` is the only module that imports +both concrete classes and the only one that reads `MASSIVE_API_KEY`. + +### File layout + +``` +backend/ + app/ + market/ + __init__.py # Re-exports: PriceUpdate, PriceCache, MarketDataSource, + # create_market_data_source, create_stream_router + models.py # PriceUpdate dataclass + cache.py # PriceCache (thread-safe in-memory store) + interface.py # MarketDataSource ABC + seed_prices.py # SEED_PRICES, TICKER_PARAMS, DEFAULT_PARAMS, CORRELATION_GROUPS + simulator.py # GBMSimulator + SimulatorDataSource + massive_client.py # MassiveDataSource + factory.py # create_market_data_source() + stream.py # SSE endpoint (FastAPI router factory) + tests/ + market/ # test_models.py, test_cache.py, test_simulator.py, + # test_simulator_source.py, test_factory.py, test_massive.py + market_data_demo.py # Rich terminal demo of the live simulator +``` + +--- + +## 2. Data Model + +**File: `backend/app/market/models.py`** + +`PriceUpdate` is the only data structure that leaves the market data layer. +Every downstream consumer — SSE streaming, portfolio valuation, trade +execution — works exclusively with this type. + +```python +"""Data models for market data.""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass(frozen=True, slots=True) +class PriceUpdate: + """Immutable snapshot of a single ticker's price at a point in time.""" + + ticker: str + price: float + previous_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: + """Absolute price change from previous update.""" + return round(self.price - self.previous_price, 4) + + @property + def change_percent(self) -> float: + """Percentage change from previous update.""" + if self.previous_price == 0: + return 0.0 + return round((self.price - self.previous_price) / self.previous_price * 100, 4) + + @property + def direction(self) -> str: + """'up', 'down', or 'flat'.""" + if self.price > self.previous_price: + return "up" + elif self.price < self.previous_price: + return "down" + return "flat" + + def to_dict(self) -> dict: + """Serialize for JSON / SSE transmission.""" + return { + "ticker": self.ticker, + "price": self.price, + "previous_price": self.previous_price, + "timestamp": self.timestamp, + "change": self.change, + "change_percent": self.change_percent, + "direction": self.direction, + } +``` + +### Design decisions + +- **`frozen=True`**: price updates are immutable value objects — safe to + share across threads/async tasks without defensive copying. +- **`slots=True`**: memory optimization; many of these are created per second. +- **Computed properties** (`change`, `direction`, `change_percent`): derived + from `price`/`previous_price` so they can never drift out of sync — there + is no stale `direction` field to accidentally forget to update. +- **`to_dict()`**: single serialization point used by both the SSE endpoint + and any REST API response that needs a price. + +--- + +## 3. Price Cache + +**File: `backend/app/market/cache.py`** + +The central data hub. Data sources write to it; SSE streaming, portfolio +valuation, and trade execution read from it. It must be thread-safe because +`MassiveDataSource` runs its synchronous HTTP call via `asyncio.to_thread` +(a real OS thread), while everything else touches it from the asyncio event +loop. + +```python +"""Thread-safe in-memory price cache.""" + +from __future__ import annotations + +import time +from threading import Lock + +from .models import PriceUpdate + + +class PriceCache: + """Thread-safe in-memory cache of the latest price for each ticker. + + Writers: SimulatorDataSource or MassiveDataSource (one at a time). + Readers: SSE streaming endpoint, portfolio valuation, trade execution. + """ + + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._lock = Lock() + self._version: int = 0 # Monotonically increasing; bumped on every update + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + """Record a new price for a ticker. Returns the created PriceUpdate. + + Automatically computes direction and change from the previous price. + If this is the first update for the ticker, previous_price == price + (direction='flat'). + """ + with self._lock: + ts = timestamp or time.time() + prev = self._prices.get(ticker) + previous_price = prev.price if prev else price + + update = PriceUpdate( + ticker=ticker, + price=round(price, 2), + previous_price=round(previous_price, 2), + timestamp=ts, + ) + self._prices[ticker] = update + self._version += 1 + return update + + def get(self, ticker: str) -> PriceUpdate | None: + """Get the latest price for a single ticker, or None if unknown.""" + with self._lock: + return self._prices.get(ticker) + + def get_all(self) -> dict[str, PriceUpdate]: + """Snapshot of all current prices. Returns a shallow copy.""" + with self._lock: + return dict(self._prices) + + def get_price(self, ticker: str) -> float | None: + """Convenience: get just the price float, or None.""" + update = self.get(ticker) + return update.price if update else None + + def remove(self, ticker: str) -> None: + """Remove a ticker from the cache (e.g., when removed from watchlist).""" + with self._lock: + self._prices.pop(ticker, None) + + @property + def version(self) -> int: + """Current version counter. Useful for SSE change detection.""" + return self._version + + def __len__(self) -> int: + with self._lock: + return len(self._prices) + + def __contains__(self, ticker: str) -> bool: + with self._lock: + return ticker in self._prices +``` + +### Why a version counter? + +The SSE streaming loop polls the cache every ~500ms. Without a version +counter it would serialize and send every price on every tick even when +nothing changed (e.g. Massive only updates every 15s). The counter lets the +loop skip a send when nothing is new: + +```python +last_version = -1 +while True: + if price_cache.version != last_version: + last_version = price_cache.version + yield format_sse(price_cache.get_all()) + await asyncio.sleep(0.5) +``` + +`update()` computes `previous_price` internally from whatever was cached +before — callers only ever supply the new price. That's what makes +`direction`/`change` correct without every writer duplicating that logic. + +--- + +## 4. Abstract Interface + +**File: `backend/app/market/interface.py`** + +```python +"""Abstract interface for market data sources.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class MarketDataSource(ABC): + """Contract for market data providers. + + Implementations push price updates into a shared PriceCache on their own + schedule. Downstream code never calls the data source directly for prices — + it reads from the cache. + + Lifecycle: + source = create_market_data_source(cache) + await source.start(["AAPL", "GOOGL", ...]) + # ... app runs ... + await source.add_ticker("TSLA") + await source.remove_ticker("GOOGL") + # ... app shutting down ... + await source.stop() + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing price updates for the given tickers. + + Starts a background task that periodically writes to the PriceCache. + Must be called exactly once. Calling start() twice is undefined behavior. + """ + + @abstractmethod + async def stop(self) -> None: + """Stop the background task and release resources. + + Safe to call multiple times. After stop(), the source will not write + to the cache again. + """ + + @abstractmethod + async def add_ticker(self, ticker: str) -> None: + """Add a ticker to the active set. No-op if already present. + + The next update cycle will include this ticker. + """ + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Remove a ticker from the active set. No-op if not present. + + Also removes the ticker from the PriceCache. + """ + + @abstractmethod + def get_tickers(self) -> list[str]: + """Return the current list of actively tracked tickers.""" +``` + +### Design choices baked into this contract + +- **Async lifecycle, sync accessor** — `start`/`stop`/`add_ticker`/ + `remove_ticker` are `async` because both implementations do I/O or task + management on those paths (spawning a poller / the sim loop); + `get_tickers()` is sync because it's just a local list read. +- **No `get_price()` on the interface** — price reads always go through + `PriceCache`, never through the source. This keeps "who writes" (sources) + and "who reads" (everyone else) strictly separated; adding a third data + source later requires zero changes to any reader. +- **Idempotent `stop()`, no-op `add_ticker`/`remove_ticker` on duplicates** + — callers (route handlers, chat action execution) don't need to + pre-check state before calling. +- **Push model, not pull** — the source decides its own timing internally + (simulator ticks every 500ms, Massive polls every 15s) and just writes to + the cache whenever it has something. The SSE layer never needs to know + which source is active or how often it updates. + +--- + +## 5. Seed Prices & Ticker Parameters + +**File: `backend/app/market/seed_prices.py`** + +Constants only — no logic, no imports beyond stdlib types. Shared by the +simulator (initial prices + GBM parameters) and available as sane fallback +seed prices for anything that needs one. + +```python +"""Seed prices and per-ticker parameters for the market simulator.""" + +# Realistic starting prices for the default watchlist (as of project creation) +SEED_PRICES: dict[str, float] = { + "AAPL": 190.00, + "GOOGL": 175.00, + "MSFT": 420.00, + "AMZN": 185.00, + "TSLA": 250.00, + "NVDA": 800.00, + "META": 500.00, + "JPM": 195.00, + "V": 280.00, + "NFLX": 600.00, +} + +# Per-ticker GBM parameters +# sigma: annualized volatility (higher = more price movement) +# mu: annualized drift / expected return +TICKER_PARAMS: dict[str, dict[str, float]] = { + "AAPL": {"sigma": 0.22, "mu": 0.05}, + "GOOGL": {"sigma": 0.25, "mu": 0.05}, + "MSFT": {"sigma": 0.20, "mu": 0.05}, + "AMZN": {"sigma": 0.28, "mu": 0.05}, + "TSLA": {"sigma": 0.50, "mu": 0.03}, # High volatility + "NVDA": {"sigma": 0.40, "mu": 0.08}, # High volatility, strong drift + "META": {"sigma": 0.30, "mu": 0.05}, + "JPM": {"sigma": 0.18, "mu": 0.04}, # Low volatility (bank) + "V": {"sigma": 0.17, "mu": 0.04}, # Low volatility (payments) + "NFLX": {"sigma": 0.35, "mu": 0.05}, +} + +# Default parameters for tickers not in the list above (dynamically added) +DEFAULT_PARAMS: dict[str, float] = {"sigma": 0.25, "mu": 0.05} + +# Correlation groups for the simulator's Cholesky decomposition +# Tickers in the same group have higher intra-group correlation +CORRELATION_GROUPS: dict[str, set[str]] = { + "tech": {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"}, + "finance": {"JPM", "V"}, +} + +# Correlation coefficients +INTRA_TECH_CORR = 0.6 # Tech stocks move together +INTRA_FINANCE_CORR = 0.5 # Finance stocks move together +CROSS_GROUP_CORR = 0.3 # Between sectors / unknown tickers +TSLA_CORR = 0.3 # TSLA does its own thing +``` + +A ticker added later that isn't in `SEED_PRICES`/`TICKER_PARAMS` (a user +adds an arbitrary symbol via the watchlist or chat) gets a random seed price +in `[$50, $300]` and `DEFAULT_PARAMS` — it still participates fully in the +simulation, just without hand-tuned realism (see `_add_ticker_internal` +below). + +--- + +## 6. GBM Simulator + +**File: `backend/app/market/simulator.py`** + +The default, primary data source — not a fallback bolted on for when a paid +API key is missing (see `planning/PLAN.md` §6). It needs no external +dependency beyond `numpy`, no network calls, no rate limits, and produces +the ~500ms-cadence price action the frontend's flash animations and +sparklines are built around, which a 15s-polled free-tier Massive feed +cannot deliver on its own. + +Two classes live here: +- `GBMSimulator` — pure math engine, no I/O. Stateful: holds current prices + and advances them one step at a time. +- `SimulatorDataSource` — the `MarketDataSource` adapter that wraps + `GBMSimulator` in an async loop and writes results into the `PriceCache`. + +### 6.1 The model — Geometric Brownian Motion + +Each ticker's price evolves under GBM: + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +Where `S(t)` is the current price, `mu` is annualized drift, `sigma` is +annualized volatility, `dt` is the time step as a fraction of a trading +year, and `Z` is a (correlated) standard normal random draw. + +**Choosing `dt`**: ticks happen every 500ms, but `mu`/`sigma` are +annualized, so `dt` has to convert "half a second" into "fraction of a +trading 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` is what keeps individual ticks to realistic sub-cent moves +that accumulate into believable multi-minute price action, rather than each +tick looking like a full day's move. + +**Correlated moves via Cholesky decomposition**: real markets don't move +ticker-by-ticker independently — tech stocks tend to move together, as do +financials. Draw `n` independent standard normals, then multiply by the +Cholesky factor of a correlation matrix built from sector groupings: + +```python +z_independent = np.random.standard_normal(n) +z_correlated = cholesky_factor @ z_independent # correlated draws +``` + +| Pair | Correlation | +|---|---| +| Two tech tickers (AAPL, GOOGL, MSFT, AMZN, META, NVDA, NFLX) | 0.6 | +| Two finance tickers (JPM, V) | 0.5 | +| Either ticker is TSLA | 0.3 (TSLA "does its own thing") | +| Cross-sector / unknown ticker | 0.3 | + +The correlation matrix — and its Cholesky factor — is rebuilt whenever a +ticker is added or removed (O(n²), fine for n < ~50 watchlist tickers). + +**Random shock events**: independently of the GBM step, each ticker has a +small per-tick chance (`event_probability`, default 0.1%) of a sudden 2–5% +move in either direction — the occasional dramatic single-ticker spike/drop +for visual interest, layered on top of the smooth GBM walk. At 10 tickers +and 2 ticks/sec, expect roughly one event every ~50 seconds. + +### 6.2 `GBMSimulator` — the math engine + +```python +"""GBM-based market simulator.""" + +from __future__ import annotations + +import asyncio +import logging +import math +import random + +import numpy as np + +from .cache import PriceCache +from .interface import MarketDataSource +from .seed_prices import ( + CORRELATION_GROUPS, + CROSS_GROUP_CORR, + DEFAULT_PARAMS, + INTRA_FINANCE_CORR, + INTRA_TECH_CORR, + SEED_PRICES, + TICKER_PARAMS, + TSLA_CORR, +) + +logger = logging.getLogger(__name__) + + +class GBMSimulator: + """Geometric Brownian Motion simulator for correlated stock prices. + + Math: + S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) + + The tiny dt (~8.5e-8 for 500ms ticks over 252 trading days * 6.5h/day) + produces sub-cent moves per tick that accumulate naturally over time. + """ + + # 500ms expressed as a fraction of a trading year + # 252 trading days * 6.5 hours/day * 3600 seconds/hour = 5,896,800 seconds + TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 + DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8 + + def __init__( + self, + tickers: list[str], + dt: float = DEFAULT_DT, + event_probability: float = 0.001, + ) -> None: + self._dt = dt + self._event_prob = event_probability + + # Per-ticker state + self._tickers: list[str] = [] + self._prices: dict[str, float] = {} + self._params: dict[str, dict[str, float]] = {} + + # Cholesky decomposition of the correlation matrix (for correlated moves) + self._cholesky: np.ndarray | None = None + + # Initialize all starting tickers + for ticker in tickers: + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + # --- Public API --- + + def step(self) -> dict[str, float]: + """Advance all tickers by one time step. Returns {ticker: new_price}. + + This is the hot path — called every 500ms. Keep it fast. + """ + n = len(self._tickers) + if n == 0: + return {} + + # Generate n independent standard normal draws + z_independent = np.random.standard_normal(n) + + # Apply Cholesky to get correlated draws + if self._cholesky is not None: + z_correlated = self._cholesky @ z_independent + else: + z_correlated = z_independent + + result: dict[str, float] = {} + for i, ticker in enumerate(self._tickers): + params = self._params[ticker] + mu = params["mu"] + sigma = params["sigma"] + + # GBM: S(t+dt) = S(t) * exp((mu - 0.5*sigma^2)*dt + sigma*sqrt(dt)*Z) + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z_correlated[i] + self._prices[ticker] *= math.exp(drift + diffusion) + + # Random event: ~0.1% chance per tick per ticker + # With 10 tickers at 2 ticks/sec, expect an event ~every 50 seconds + if random.random() < self._event_prob: + shock_magnitude = random.uniform(0.02, 0.05) + shock_sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock_magnitude * shock_sign + logger.debug( + "Random event on %s: %.1f%% %s", + ticker, + shock_magnitude * 100, + "up" if shock_sign > 0 else "down", + ) + + result[ticker] = round(self._prices[ticker], 2) + + return result + + def add_ticker(self, ticker: str) -> None: + """Add a ticker to the simulation. Rebuilds the correlation matrix.""" + if ticker in self._prices: + return + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + def remove_ticker(self, ticker: str) -> None: + """Remove a ticker from the simulation. Rebuilds the correlation matrix.""" + if ticker not in self._prices: + return + self._tickers.remove(ticker) + del self._prices[ticker] + del self._params[ticker] + self._rebuild_cholesky() + + def get_price(self, ticker: str) -> float | None: + """Current price for a ticker, or None if not tracked.""" + return self._prices.get(ticker) + + def get_tickers(self) -> list[str]: + """Return the list of currently tracked tickers.""" + return list(self._tickers) + + # --- Internals --- + + def _add_ticker_internal(self, ticker: str) -> None: + """Add a ticker without rebuilding Cholesky (for batch initialization).""" + if ticker in self._prices: + return + self._tickers.append(ticker) + self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) + self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) + + def _rebuild_cholesky(self) -> None: + """Rebuild the Cholesky decomposition of the ticker correlation matrix. + + Called whenever tickers are added or removed. O(n^2) but n < 50. + """ + n = len(self._tickers) + if n <= 1: + self._cholesky = None + return + + # Build the correlation matrix + 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] = rho + corr[j, i] = rho + + self._cholesky = np.linalg.cholesky(corr) + + @staticmethod + def _pairwise_correlation(t1: str, t2: str) -> float: + """Determine correlation between two tickers based on sector grouping.""" + tech = CORRELATION_GROUPS["tech"] + finance = CORRELATION_GROUPS["finance"] + + # TSLA is in the tech set but behaves independently + if t1 == "TSLA" or t2 == "TSLA": + return TSLA_CORR + + if t1 in tech and t2 in tech: + return INTRA_TECH_CORR + if t1 in finance and t2 in finance: + return INTRA_FINANCE_CORR + + return CROSS_GROUP_CORR +``` + +### 6.3 `SimulatorDataSource` — async wrapper + +```python +class SimulatorDataSource(MarketDataSource): + """MarketDataSource backed by the GBM simulator. + + Runs a background asyncio task that calls GBMSimulator.step() every + `update_interval` seconds and writes results to the PriceCache. + """ + + def __init__( + self, + price_cache: PriceCache, + update_interval: float = 0.5, + event_probability: float = 0.001, + ) -> None: + self._cache = price_cache + self._interval = update_interval + self._event_prob = event_probability + self._sim: GBMSimulator | None = None + self._task: asyncio.Task | None = None + + async def start(self, tickers: list[str]) -> None: + self._sim = GBMSimulator( + tickers=tickers, + event_probability=self._event_prob, + ) + # Seed the cache with initial prices so SSE has data immediately + for ticker in tickers: + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + logger.info("Simulator started with %d tickers", len(tickers)) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("Simulator stopped") + + async def add_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.add_ticker(ticker) + # Seed cache immediately so the ticker has a price right away + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + logger.info("Simulator: added ticker %s", ticker) + + async def remove_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.remove_ticker(ticker) + self._cache.remove(ticker) + logger.info("Simulator: removed ticker %s", ticker) + + def get_tickers(self) -> list[str]: + return self._sim.get_tickers() if self._sim else [] + + async def _run_loop(self) -> None: + """Core loop: step the simulation, write to cache, sleep.""" + while True: + try: + if self._sim: + prices = self._sim.step() + for ticker, price in prices.items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +### Key behaviors + +- **Immediate seeding**: `start()` and `add_ticker()` both write to the + cache *before* the loop's next tick, so the SSE endpoint (or a freshly + added watchlist ticker) never shows blank/missing for up to 500ms. +- **Graceful cancellation**: `stop()` cancels the task and awaits it, + catching `CancelledError` — clean shutdown during FastAPI lifespan + teardown. +- **Exception resilience**: `_run_loop` catches exceptions per-step so one + bad tick doesn't kill the whole background task. +- **Public accessor, not private-attribute reach-through**: + `get_tickers()` delegates to `GBMSimulator.get_tickers()` rather than + reading `self._sim._tickers` directly, keeping the class boundary clean. + +### Example usage + +```python +from app.market import PriceCache, create_market_data_source + +cache = PriceCache() +source = create_market_data_source(cache) # SimulatorDataSource, since + # MASSIVE_API_KEY is unset +await source.start(["AAPL", "GOOGL", "MSFT", "TSLA"]) + +# ... 500ms later ... +update = cache.get("TSLA") +print(update.price, update.direction, update.change_percent) + +await source.add_ticker("NVDA") # immediately visible in the cache +await source.remove_ticker("MSFT") + +await source.stop() +``` + +A standalone terminal visualization of this exact simulator is available at +`backend/market_data_demo.py` (`uv run market_data_demo.py` from +`backend/`) — a live Rich dashboard with sparklines, direction arrows, and +an event log; useful for eyeballing whether tuning changes to `sigma`/`mu` +or the correlation groups still feel realistic. + +--- + +## 7. Massive API Client + +**File: `backend/app/market/massive_client.py`** + +Polls Massive's (formerly Polygon.io) multi-ticker snapshot endpoint on a +timer. Per `planning/PLAN.md` §6 and `CLAUDE.md`, this project deliberately +ships with `MASSIVE_API_KEY` unset — the simulator is the supported, +primary path — but this client is kept working as a real alternative. + +### 7.1 Endpoint used + +``` +GET /v2/snapshot/locale/us/markets/stocks/tickers?tickers=AAPL,GOOGL,MSFT +``` + +Returns last trade, last quote, and day/previous-day OHLC for many tickers +in **one API call** — this is what makes polling viable even on the free +tier (5 req/min): a 10-ticker watchlist still costs one request per poll. + +```python +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +client = RESTClient() # reads MASSIVE_API_KEY from the environment + +snapshots = client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"], +) +for snap in snapshots: + print(f"{snap.ticker}: ${snap.last_trade.price}") +``` + +Rate limits: **free tier → 5 req/min → poll every 15s** (default); +**paid tiers → poll every 2–5s**. See `planning/MASSIVE_API.md` for the full +endpoint catalog (single-ticker snapshot, previous close, custom bars, +last trade/quote) and error-code reference. + +### 7.2 `MassiveDataSource` + +```python +"""Massive (Polygon.io) API client for real market data.""" + +from __future__ import annotations + +import asyncio +import logging + +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +from .cache import PriceCache +from .interface import MarketDataSource + +logger = logging.getLogger(__name__) + + +class MassiveDataSource(MarketDataSource): + """MarketDataSource backed by the Massive (Polygon.io) REST API. + + Polls GET /v2/snapshot/locale/us/markets/stocks/tickers for all watched + tickers in a single API call, then writes results to the PriceCache. + + Rate limits: + - Free tier: 5 req/min → poll every 15s (default) + - Paid tiers: higher limits → poll every 2-5s + """ + + def __init__( + self, + api_key: str, + price_cache: PriceCache, + poll_interval: float = 15.0, + ) -> None: + self._api_key = api_key + self._cache = price_cache + self._interval = poll_interval + self._tickers: list[str] = [] + self._task: asyncio.Task | None = None + self._client: RESTClient | None = None + + async def start(self, tickers: list[str]) -> None: + self._client = RESTClient(api_key=self._api_key) + self._tickers = list(tickers) + + # Do an immediate first poll so the cache has data right away + await self._poll_once() + + self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") + logger.info( + "Massive poller started: %d tickers, %.1fs interval", + len(tickers), + self._interval, + ) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + self._client = None + logger.info("Massive poller stopped") + + async def add_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + if ticker not in self._tickers: + self._tickers.append(ticker) + logger.info("Massive: added ticker %s (will appear on next poll)", ticker) + + async def remove_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + self._tickers = [t for t in self._tickers if t != ticker] + self._cache.remove(ticker) + logger.info("Massive: removed ticker %s", ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + # --- Internal --- + + async def _poll_loop(self) -> None: + """Poll on interval. First poll already happened in start().""" + while True: + await asyncio.sleep(self._interval) + await self._poll_once() + + async def _poll_once(self) -> None: + """Execute one poll cycle: fetch snapshots, update cache.""" + if not self._tickers or not self._client: + return + + try: + # The Massive RESTClient is synchronous — run in a thread to + # avoid blocking the event loop. + snapshots = await asyncio.to_thread(self._fetch_snapshots) + processed = 0 + for snap in snapshots: + try: + price = snap.last_trade.price + # Massive timestamps are Unix milliseconds → convert to seconds + timestamp = snap.last_trade.timestamp / 1000.0 + self._cache.update( + ticker=snap.ticker, + price=price, + timestamp=timestamp, + ) + processed += 1 + except (AttributeError, TypeError) as e: + logger.warning( + "Skipping snapshot for %s: %s", + getattr(snap, "ticker", "???"), + e, + ) + logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers)) + + except Exception as e: + logger.error("Massive poll failed: %s", e) + # Don't re-raise — the loop will retry on the next interval. + # Common failures: 401 (bad key), 429 (rate limit), network errors. + + def _fetch_snapshots(self) -> list: + """Synchronous call to the Massive REST API. Runs in a thread.""" + return self._client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=self._tickers, + ) +``` + +`massive` (the `polygon-api-client` package's successor on PyPI, per the +October 2025 rebrand — see `planning/MASSIVE_API.md` §1) is imported at +**module level**, not lazily inside a method. It is declared as a core +backend dependency in `pyproject.toml`, so it is always installed +regardless of which data source ends up active at runtime; the +`MASSIVE_API_KEY` env var, not import structure, is what decides whether +`MassiveDataSource` is actually instantiated (see §8, Factory). + +### 7.3 Error handling philosophy + +The poller is intentionally resilient — it never lets a bad HTTP response +kill the background task: + +| Error | Behavior | +|-------|----------| +| **401 Unauthorized** | Logged as error. Poller keeps running (user might fix `.env` and restart). | +| **429 Rate Limited** | Logged as error. Next poll retries after `poll_interval` seconds. | +| **Network timeout** | Logged as error. Retries automatically on next cycle. | +| **Malformed snapshot** | Individual ticker skipped with a warning; other tickers still processed. | +| **All tickers fail** | Cache retains last-known prices. SSE keeps streaming stale data (better than no data). | + +--- + +## 8. Factory + +**File: `backend/app/market/factory.py`** + +The only module that imports both concrete `MarketDataSource` +implementations and the only one that reads `MASSIVE_API_KEY`. + +```python +"""Factory for creating market data sources.""" + +from __future__ import annotations + +import logging +import os + +from .cache import PriceCache +from .interface import MarketDataSource +from .massive_client import MassiveDataSource +from .simulator import SimulatorDataSource + +logger = logging.getLogger(__name__) + + +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """Create the appropriate market data source based on environment variables. + + - MASSIVE_API_KEY set and non-empty → MassiveDataSource (real market data) + - Otherwise → SimulatorDataSource (GBM simulation) + + Returns an unstarted source. Caller must await source.start(tickers). + """ + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + + if api_key: + logger.info("Market data source: Massive API (real data)") + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + else: + logger.info("Market data source: GBM Simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +Trimming and checking for non-empty (not just presence) matters: an `.env` +file with `MASSIVE_API_KEY=` (present but blank) must fall back to the +simulator, not attempt a poll with an empty key. + +### Usage at app startup + +```python +price_cache = PriceCache() +source = create_market_data_source(price_cache) +await source.start(initial_tickers) # e.g. ["AAPL", "GOOGL", ...] +``` + +### Extending this later + +Adding a third source (a different vendor, a WebSocket-based feed, etc.) +requires only: implement `MarketDataSource`, and extend the factory's +branch on the relevant env var. No other module changes, because every +consumer already goes through `PriceCache`. + +--- + +## 9. SSE Streaming Endpoint + +**File: `backend/app/market/stream.py`** + +A FastAPI route that holds open a long-lived HTTP connection and pushes +price updates to the client as `text/event-stream`. + +```python +"""SSE streaming endpoint for live price updates.""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import AsyncGenerator + +from fastapi import APIRouter, Request +from fastapi.responses import StreamingResponse + +from .cache import PriceCache + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/stream", tags=["streaming"]) + + +def create_stream_router(price_cache: PriceCache) -> APIRouter: + """Create the SSE streaming router with a reference to the price cache. + + This factory pattern lets us inject the PriceCache without globals. + """ + + @router.get("/prices") + async def stream_prices(request: Request) -> StreamingResponse: + """SSE endpoint for live price updates. + + Streams all tracked ticker prices every ~500ms. The client connects + with EventSource and receives events in the format: + + data: {"AAPL": {"ticker": "AAPL", "price": 190.50, ...}, ...} + + Includes a retry directive so the browser auto-reconnects on + disconnection (EventSource built-in behavior). + """ + return StreamingResponse( + _generate_events(price_cache, request), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", # Disable nginx buffering if proxied + }, + ) + + return router + + +async def _generate_events( + price_cache: PriceCache, + request: Request, + interval: float = 0.5, +) -> AsyncGenerator[str, None]: + """Async generator that yields SSE-formatted price events. + + Sends all prices every `interval` seconds. Stops when the client + disconnects (detected via request.is_disconnected()). + """ + # Tell the client to retry after 1 second if the connection drops + 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: + # Check for client disconnect + 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()} + payload = json.dumps(data) + yield f"data: {payload}\n\n" + + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("SSE stream cancelled for: %s", client_ip) +``` + +### SSE wire format + +``` +data: {"AAPL":{"ticker":"AAPL","price":190.50,"previous_price":190.42,"timestamp":1707580800.5,"change":0.08,"change_percent":0.042,"direction":"up"},"GOOGL":{"ticker":"GOOGL","price":175.12,...}} + +``` + +The frontend parses this with the native `EventSource` API: + +```javascript +const eventSource = new EventSource('/api/stream/prices'); +eventSource.onmessage = (event) => { + const prices = JSON.parse(event.data); + // prices is { "AAPL": { ticker, price, previous_price, change, change_percent, direction, timestamp }, ... } +}; +``` + +### Why poll-and-push instead of event-driven? + +The SSE endpoint polls the cache on a fixed interval rather than being +notified by the data source. This is simpler and produces predictable, +evenly-spaced updates for the frontend, which accumulates them client-side +into sparkline charts — regular spacing matters for a clean line. + +### Why watchlist changes need no reconnect + +Because `add_ticker`/`remove_ticker` write straight into the same +`PriceCache` the stream reads from, the next tick after a watchlist change +just includes (or drops) that ticker — the SSE loop and the mutation are +fully decoupled through the cache, with no direct call path between them. + +--- + +## 10. FastAPI Lifecycle Integration + +The market data system starts and stops with the app via the `lifespan` +context manager. + +**In `backend/app/main.py`:** + +```python +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from app.market import PriceCache, create_market_data_source, create_stream_router + + +@asynccontextmanager +async def lifespan(app: FastAPI): + """Manage startup and shutdown of background services.""" + + # --- STARTUP --- + + # 1. Create the shared price cache + price_cache = PriceCache() + app.state.price_cache = price_cache + + # 2. Create and start the market data source + source = create_market_data_source(price_cache) + app.state.market_source = source + + # 3. Load initial tickers from the database watchlist (lazy-init happens here too) + initial_tickers = await load_watchlist_tickers() # reads from SQLite + await source.start(initial_tickers) + + # 4. Register the SSE streaming router + app.include_router(create_stream_router(price_cache)) + + yield # App is running + + # --- SHUTDOWN --- + await source.stop() + + +app = FastAPI(title="FinAlly", lifespan=lifespan) + + +def get_price_cache() -> PriceCache: + return app.state.price_cache + + +def get_market_source(): + return app.state.market_source +``` + +### Accessing market data from other routes + +Trade execution, portfolio valuation, and watchlist management access the +cache and data source via FastAPI dependency injection: + +```python +from fastapi import APIRouter, Depends, HTTPException + +router = APIRouter(prefix="/api") + + +@router.post("/portfolio/trade") +async def execute_trade(trade: TradeRequest, price_cache: PriceCache = Depends(get_price_cache)): + current_price = price_cache.get_price(trade.ticker) + if current_price is None: + raise HTTPException(404, f"No price available for {trade.ticker}") + # ... execute trade at current_price via the single execute_trade() function + # (see planning/PLAN.md §9 — the chat action executor calls the same function) ... + + +@router.post("/watchlist") +async def add_to_watchlist(payload: WatchlistAdd, source=Depends(get_market_source)): + # Insert into the watchlist table ... + await source.add_ticker(payload.ticker) + + +@router.delete("/watchlist/{ticker}") +async def remove_from_watchlist(ticker: str, source=Depends(get_market_source)): + # Delete from the watchlist table ... + await source.remove_ticker(ticker) +``` + +--- + +## 11. Watchlist Coordination + +### Flow: adding a ticker + +``` +User (or LLM) → POST /api/watchlist {ticker: "PYPL"} + → Insert into watchlist table (SQLite) + → await source.add_ticker("PYPL") + Simulator: adds to GBMSimulator, rebuilds Cholesky, seeds cache immediately + Massive: appends to ticker list, appears on next poll + → Return success (ticker + current price if already available) +``` + +### Flow: removing a ticker + +``` +User (or LLM) → DELETE /api/watchlist/PYPL + → Delete from watchlist table (SQLite) + → await source.remove_ticker("PYPL") + Simulator: removes from GBMSimulator, rebuilds Cholesky, removes from cache + Massive: removes from ticker list, removes from cache + → Return success +``` + +### Edge case: ticker has an open position + +If the user removes a ticker from the watchlist but still holds shares, the +data source should keep tracking it so portfolio valuation stays accurate: + +```python +@router.delete("/watchlist/{ticker}") +async def remove_from_watchlist(ticker: str, source=Depends(get_market_source)): + await db.delete_watchlist_entry(ticker) + + # Only stop tracking if there's no open position + position = await db.get_position(ticker) + if position is None or position.quantity == 0: + await source.remove_ticker(ticker) + + return {"status": "ok"} +``` + +--- + +## 12. Testing Strategy + +The existing suite lives in `backend/tests/market/` — 73 tests, all +passing, 91% overall coverage (per `planning/MARKET_DATA_SUMMARY.md` and +the independent verification in `planning/review.md`). Coverage is uneven: +`stream.py` sits around 33% because exercising the SSE generator properly +needs a running ASGI test client, not just unit tests — a real gap worth +closing, not a rounding error to wave off. + +| Module | Tests | What it covers | +|--------|-------|-----------------| +| `test_models.py` | 11 | `PriceUpdate` properties (`change`, `direction`, `change_percent`), `to_dict()` | +| `test_cache.py` | 13 | `PriceCache` update/get/get_all/remove, version increments, first-update-is-flat | +| `test_simulator.py` | 17 | `GBMSimulator` math: positive prices, drift over many steps, add/remove ticker rebuilds Cholesky, unknown-ticker random seed | +| `test_simulator_source.py` | 10 | `SimulatorDataSource` async lifecycle: start seeds cache, prices update over time, clean stop, add/remove ticker | +| `test_factory.py` | 7 | env var branching (`MASSIVE_API_KEY` set/unset/blank) | +| `test_massive.py` | 13 | `MassiveDataSource` with the Massive client mocked — poll success, malformed snapshot skip, API error doesn't crash the loop | + +### 12.1 Representative example — `GBMSimulator` math properties + +```python +from app.market.simulator import GBMSimulator +from app.market.seed_prices import SEED_PRICES + + +class TestGBMSimulator: + def test_prices_are_positive(self): + """GBM prices can never go negative (exp() is always positive).""" + sim = GBMSimulator(tickers=["AAPL"]) + for _ in range(10_000): + prices = sim.step() + assert prices["AAPL"] > 0 + + def test_add_ticker_rebuilds_correlation(self): + sim = GBMSimulator(tickers=["AAPL"]) + assert sim._cholesky is None # only 1 ticker, no correlation matrix needed + sim.add_ticker("GOOGL") + assert sim._cholesky is not None + + def test_unknown_ticker_gets_random_seed_price(self): + sim = GBMSimulator(tickers=["ZZZZ"]) + price = sim.get_price("ZZZZ") + assert 50.0 <= price <= 300.0 +``` + +### 12.2 Representative example — `PriceCache` + +```python +from app.market.cache import PriceCache + + +class TestPriceCache: + def test_first_update_is_flat(self): + cache = PriceCache() + update = cache.update("AAPL", 190.50) + assert update.direction == "flat" + assert update.previous_price == 190.50 + + def test_version_increments(self): + cache = PriceCache() + v0 = cache.version + cache.update("AAPL", 190.00) + assert cache.version == v0 + 1 +``` + +### 12.3 Representative example — `MassiveDataSource` (mocked) + +```python +from unittest.mock import MagicMock, patch +import pytest +from app.market.cache import PriceCache +from app.market.massive_client import MassiveDataSource + + +def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock: + snap = MagicMock() + snap.ticker = ticker + snap.last_trade.price = price + snap.last_trade.timestamp = timestamp_ms + return snap + + +@pytest.mark.asyncio +async def test_poll_updates_cache(): + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL", "GOOGL"] + + mock_snapshots = [ + _make_snapshot("AAPL", 190.50, 1707580800000), + _make_snapshot("GOOGL", 175.25, 1707580800000), + ] + with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): + await source._poll_once() + + assert cache.get_price("AAPL") == 190.50 + assert cache.get_price("GOOGL") == 175.25 +``` + +Because `_fetch_snapshots` is a plain instance method (not a lazily-imported +free function), tests patch it directly with `patch.object(source, ...)` +rather than patching a module-level `RESTClient` name — this is simpler now +that `massive` is a top-level import (see §7.2) and avoids the mock-target +fragility that an earlier revision of this design had. + +### 12.4 Closing the `stream.py` gap + +The recommended way to push `stream.py` coverage up is an ASGI-level +integration test using `httpx.ASGITransport`, reading a few chunks off the +response and asserting on the decoded SSE payload: + +```python +import httpx +import pytest +from app.main import app + + +@pytest.mark.asyncio +async def test_sse_stream_emits_prices(): + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with client.stream("GET", "/api/stream/prices") as response: + assert response.status_code == 200 + async for line in response.aiter_lines(): + if line.startswith("data: "): + break +``` + +--- + +## 13. Error Handling & Edge Cases + +### 13.1 Startup: empty watchlist + +If the database has no watchlist entries, `start()` receives an empty list. +Both sources handle this gracefully — the simulator produces no prices, the +Massive poller skips its API call (`if not self._tickers: return`). The SSE +endpoint sends nothing until a ticker is added, at which point the source +starts tracking it immediately. + +### 13.2 Price cache miss during trade + +If a user tries to trade a ticker with no cached price yet (just added, +Massive hasn't polled): + +```python +price = price_cache.get_price(ticker) +if price is None: + raise HTTPException( + status_code=400, + detail=f"Price not yet available for {ticker}. Please wait a moment and try again.", + ) +``` + +The simulator avoids this in practice by seeding the cache synchronously +inside `add_ticker()`. The Massive client may have a brief gap between +`add_ticker()` returning and the next poll landing — the 400 with a clear +message is the correct response for that window. + +### 13.3 Massive API key invalid + +If the key is set but wrong, the first poll fails with 401. The poller logs +the error and keeps retrying every `poll_interval`. SSE keeps streaming +(connection healthy) but with no data. Fix: correct `.env` and restart — +there's no in-process key-reload mechanism, by design (this is a demo app, +not a service needing hot config reload). + +### 13.4 Thread safety under load + +`PriceCache` uses `threading.Lock` — a real mutex, correct across both the +event loop and the `asyncio.to_thread` worker thread the Massive client +runs on. Under normal load (≤50 tickers, 2 updates/sec) lock contention is +negligible; the critical section is a dict lookup and assignment. This is +intentionally not optimized further (e.g. no `ReadWriteLock`) — unneeded at +this project's scale. + +### 13.5 Simulator numerical stability + +- Prices are `round()`ed to 2 decimals in both `GBMSimulator.step()` and + `PriceCache.update()`. +- The exponential formulation (`exp(drift + diffusion)`) guarantees prices + stay positive — no explicit floor/clamp is needed. +- `dt` is tiny enough that even the shock-event path (`* (1 ± 0.02..0.05)`) + can't push a price to zero or overflow in any realistic run length. + +--- + +## 14. Configuration Summary + +| Parameter | Location | Default | Description | +|-----------|----------|---------|-------------| +| `MASSIVE_API_KEY` | Environment variable | `""` (unset) | If set and non-empty, use Massive; otherwise use the simulator | +| `update_interval` | `SimulatorDataSource.__init__` | `0.5` (seconds) | Time between simulator ticks | +| `poll_interval` | `MassiveDataSource.__init__` | `15.0` (seconds) | Time between Massive API polls (free-tier default) | +| `event_probability` | `GBMSimulator.__init__` | `0.001` | Chance of a random shock event per ticker per tick | +| `dt` | `GBMSimulator.__init__` | `~8.5e-8` | GBM time step (fraction of a trading year) | +| SSE push interval | `_generate_events()` | `0.5` (seconds) | Time between SSE cache polls / pushes | +| SSE retry directive | `_generate_events()` | `1000` (ms) | Browser `EventSource` reconnection delay | + +### `__init__.py` — public API surface + +**File: `backend/app/market/__init__.py`** + +```python +"""Market data subsystem for FinAlly. + +Public API: + PriceUpdate - Immutable price snapshot dataclass + PriceCache - Thread-safe in-memory price store + MarketDataSource - Abstract interface for data providers + create_market_data_source - Factory that selects simulator or Massive + create_stream_router - FastAPI router factory for SSE endpoint +""" + +from .cache import PriceCache +from .factory import create_market_data_source +from .interface import MarketDataSource +from .models import PriceUpdate +from .stream import create_stream_router + +__all__ = [ + "PriceUpdate", + "PriceCache", + "MarketDataSource", + "create_market_data_source", + "create_stream_router", +] +``` + +All downstream backend code — trade execution, watchlist routes, LLM chat +action execution, portfolio valuation — should import exclusively from +`app.market` (this `__init__.py`), never reach into submodules like +`app.market.simulator` directly. That's what keeps the rest of the backend +fully agnostic to which concrete data source is active. From d94e7a71b15894a44036d61a94b51ce1ff434f88 Mon Sep 17 00:00:00 2001 From: Raunak Sachdev Date: Tue, 25 Aug 2026 09:04:22 +0100 Subject: [PATCH 6/7] Add independent code review of the market data backend Fresh pass over backend/app/market/ and its test suite: confirms all 7 issues from the prior archived review are genuinely fixed, verifies 73/73 tests pass at 91% coverage, and surfaces a few new low-severity findings (a dormant falsy-timestamp edge case in PriceCache, two test-quality gaps, and stale coverage figures in MARKET_DATA_SUMMARY.md). Co-Authored-By: Claude Sonnet 5 --- planning/MARKET_DATA_REVIEW.md | 150 +++++++++++++++++++++++++++++++++ 1 file changed, 150 insertions(+) create mode 100644 planning/MARKET_DATA_REVIEW.md diff --git a/planning/MARKET_DATA_REVIEW.md b/planning/MARKET_DATA_REVIEW.md new file mode 100644 index 000000000..cb344802e --- /dev/null +++ b/planning/MARKET_DATA_REVIEW.md @@ -0,0 +1,150 @@ +# Market Data Backend — Code Review + +**Date:** 2026-08-25 +**Scope:** `backend/app/market/` (9 source files) and `backend/tests/market/` (6 test files, 73 tests) +**Reviewer note:** This supersedes `planning/archive/MARKET_DATA_REVIEW.md` (2026-02-10). All "must fix" and "should fix" items from that earlier review have been verified as resolved (see §4). This is a fresh, independent pass looking for anything new. + +--- + +## 1. Test Results + +Ran from a clean `uv sync --extra dev`: + +``` +uv run pytest -v → 73 passed, 0 failed +uv run pytest --cov=app → 91% overall +uv run ruff check app/ tests/ → All checks passed +uv run ruff format --check → 3 test files would be reformatted (cosmetic only, see §3.4) +``` + +**Coverage by module:** + +| Module | Coverage | Missing | +|---|---|---| +| `models.py` | 100% | — | +| `cache.py` | 100% | — | +| `interface.py` | 100% | — | +| `factory.py` | 100% | — | +| `seed_prices.py` | 100% | — | +| `__init__.py` | 100% | — | +| `simulator.py` | 98% | L149 (duplicate-add guard), L268-269 (exception-handler branch in `_run_loop`) | +| `massive_client.py` | 94% | L85-87 (`_poll_loop` body), L125 (`_fetch_snapshots` real body) | +| `stream.py` | 33% | L26-48, 62-87 — essentially the whole SSE generator and route handler | +| **Total** | **91%** | | + +This is a real improvement over what `planning/MARKET_DATA_SUMMARY.md` documents (see §5.1 — that doc is now stale). With the `massive` package actually installed, `massive_client.py` tests exercise real code paths and coverage rose from the previously-recorded 56% to 94%. + +`stream.py` at 33% remains the one genuine, unaddressed gap — no test exercises `_generate_events` end-to-end. This was flagged in the prior review and in `planning/MARKET_DATA_DESIGN.md` §12.4 (which even supplies the `httpx.ASGITransport` recipe to close it) but the test still doesn't exist. Since this endpoint is the only consumer-facing piece of the whole subsystem, it's worth adding before this is called fully tested. + +--- + +## 2. Architecture Assessment + +Confirmed by reading all 9 source modules end-to-end: the design holds up. Strategy pattern (`MarketDataSource` ABC with `SimulatorDataSource`/`MassiveDataSource`), single-writer `PriceCache` as the only read path for downstream consumers, and a factory that's the sole place branching on `MASSIVE_API_KEY` — all exactly as documented in `planning/MARKET_DATA_API.md` and `planning/MARKET_DATA_DESIGN.md`, and the docs' code samples match the real source verbatim (no drift found). + +Strengths worth calling out: +- `PriceUpdate` (`frozen=True, slots=True`) makes `direction`/`change`/`change_percent` computed properties instead of stored fields — they can't drift out of sync with `price`/`previous_price`. +- Both background loops (`SimulatorDataSource._run_loop`, `MassiveDataSource._poll_loop`) wrap their step logic in try/except so one bad tick/poll can't kill the task. +- `add_ticker` seeds the cache synchronously before returning (simulator) so a freshly-added ticker is never blank for up to 500ms — a real UX detail, correctly implemented. +- GBM math is textbook-correct: `S(t+dt) = S(t) * exp((mu - sigma²/2)*dt + sigma*sqrt(dt)*Z)`, with `dt` correctly derived from a 500ms tick against a 252-day/6.5h trading year. + +### 2.1 Verified: Cholesky correlation matrix never fails for the actual ticker universe + +The correlation-matrix construction in `_rebuild_cholesky`/`_pairwise_correlation` assigns different fixed correlations depending on pair type (intra-tech 0.6, intra-finance 0.5, TSLA-anything 0.3, cross-sector/unknown 0.3). This kind of ad-hoc, non-block-consistent correlation assignment isn't *mathematically* guaranteed to produce a positive-semi-definite matrix in general (which `np.linalg.cholesky` requires), so I stress-tested it directly: + +```python +# All subsets (size 2..12) of {7 tech tickers, 2 finance tickers, TSLA, 3 unknown tickers} +# → 4,083 combinations tested, 0 Cholesky failures +``` + +For the actual ticker set this project uses (the 10 defaults plus arbitrary user-added symbols, which all fall into the "unknown/cross-sector" 0.3 bucket), this is safe. Flagging only so it's understood as empirically-verified-safe rather than mathematically-guaranteed-safe — if the correlation constants in `seed_prices.py` are ever tuned upward (e.g., intra-tech pushed to 0.9+ while cross-sector stays low), it would be worth re-running a check like this before shipping the change, since a `LinAlgError` there would crash `add_ticker`/`remove_ticker` (and thus a live watchlist mutation) with no handling for it anywhere in the call chain. + +--- + +## 3. New Findings (not in the prior review) + +### 3.1 `PriceCache.update()`: a timestamp of exactly `0.0` is silently replaced (Severity: Low) + +```python +ts = timestamp or time.time() +``` + +`0.0` is falsy in Python, so a caller that explicitly passes `timestamp=0.0` gets `time.time()` instead — confirmed by direct test: + +``` +cache.update('AAPL', 190.0, timestamp=0.0) → update.timestamp == time.time(), not 0.0 +``` + +In practice this can't currently be hit by production code paths (`MassiveDataSource` converts real millisecond epoch timestamps, which are never 0; the simulator never passes `timestamp` at all), so this is dormant rather than an active bug. The fix, if it's worth making, is `ts = timestamp if timestamp is not None else time.time()`. + +### 3.2 `test_exception_resilience` doesn't actually test exception resilience (Severity: Low, test-quality) + +`tests/market/test_simulator_source.py::test_exception_resilience` starts a normal simulator, sleeps, and asserts the background task is still running — it never injects a failure into `GBMSimulator.step()`. This is consistent with the coverage report: `simulator.py` L268-269, the `except Exception: logger.exception(...)` branch inside `_run_loop`, is genuinely uncovered. The resilience behavior is real (verified by reading the code — a bare `try/except Exception` around the step call, which does what's claimed) but the test doesn't exercise the failure path it's named for. A tightened version would monkeypatch `sim.step` to raise once and assert the loop survives and later ticks still land. + +### 3.3 `test_custom_update_interval` is timing-dependent and could flake under load (Severity: Low, test-quality) + +```python +source = SimulatorDataSource(price_cache=cache, update_interval=0.01) +await source.start(["AAPL"]) +initial_version = cache.version +await asyncio.sleep(0.05) # Should get ~5 updates +assert cache.version > initial_version + 2 +``` + +This assumes at least 3 ticks land inside a 50ms window against a 10ms interval. On a loaded CI runner or under GIL contention from other tests running in parallel, this margin is thin enough to occasionally fail without any actual regression. Not currently flaky in this environment (ran the suite multiple times without failure), but worth a wider margin (e.g., `update_interval=0.02`, `sleep(0.15)`, assert `> initial_version`) if it's ever seen to flake in CI. + +### 3.4 Three test files are not `ruff format`-clean (Severity: Trivial) + +`ruff check` (the linter) passes clean, but `ruff format --check` flags `test_models.py`, `test_simulator.py`, and `test_simulator_source.py` — a handful of `PriceUpdate(...)` constructor calls exceed the 88-char wrap width ruff's formatter prefers, even though the project's own `line-length = 100` / `ignore = ["E501"]` lint config doesn't care. Cosmetic only; `uv run ruff format app/ tests/` would silently fix it. Not worth blocking on, but noting since "clean lint" and "clean format" aren't the same claim. + +--- + +## 4. Status of the Prior Review's Findings — all resolved, verified against current source + +| # | Prior finding | Verified status | +|---|---|---| +| 1 | Missing `[tool.hatch.build.targets.wheel]` in `pyproject.toml` — broke `uv sync`/Docker builds | **Fixed.** Present in `pyproject.toml`; `uv sync --extra dev` succeeds cleanly. | +| 2 | `massive` lazy-imported inside methods, breaking `patch("...RESTClient")` | **Fixed.** `massive_client.py` imports `RESTClient`/`SnapshotMarketType` at module level (lines 8-9). | +| 3 | `_generate_events` annotated `-> None` despite being an async generator | **Fixed.** Now `-> AsyncGenerator[str, None]` (`stream.py:55`). | +| 4 | `SimulatorDataSource.get_tickers()` reached into `GBMSimulator._tickers` (private) | **Fixed.** `GBMSimulator.get_tickers()` is now a public method; the adapter delegates to it (`simulator.py:140-142`, `257-258`). | +| 5 | Unused `DEFAULT_CORR` constant, confusingly separate from `CROSS_GROUP_CORR` | **Fixed.** `seed_prices.py` only defines `CROSS_GROUP_CORR`; no `DEFAULT_CORR` remains. | +| 6 | Unused imports (`pytest`, `math`, `asyncio`) in 4 test files | **Fixed.** `ruff check` reports zero warnings across `app/` and `tests/`. | +| 7 | 5 `test_massive.py` tests failed without the `massive` package installed | **Fixed.** All 73 tests pass with `massive` installed as a real dependency (declared in `pyproject.toml`, confirmed via `uv sync`); mocks now use `patch.object(source, "_fetch_snapshots", ...)` and direct `source._client = MagicMock()` assignment rather than patching a module-level name that didn't exist. | + +No regressions were introduced while fixing these — all corresponding tests still pass and the surrounding code reads cleanly. + +--- + +## 5. Documentation Accuracy + +### 5.1 `planning/MARKET_DATA_SUMMARY.md` is stale (Severity: Low, docs-only) + +This file (last describing itself as the authoritative summary) states: +- "84% overall coverage" — actual is **91%**. +- `massive_client.py`: "56% (expected — API methods mocked)" — actual is **94%**, because `massive` is now a real installed dependency and the module-level-import fix (§4, item 2) means tests actually exercise real code paths instead of failing before they get there. + +Both numbers were accurate as of the file's writing but predate the fixes in §4. The top-level `README.md` (`"73 tests, 91% coverage overall — stream.py is the weak spot at 33%..."`) already has the correct, current numbers — `planning/MARKET_DATA_SUMMARY.md` is the one document left with the old figures. Worth a one-line update so a future reader doesn't cite the wrong number. + +### 5.2 Everything else checked out + +- `planning/MARKET_DATA_API.md` and `planning/MARKET_DATA_DESIGN.md` code samples were diffed by eye against the real source files in `backend/app/market/` — no drift found; both documents accurately describe the shipped implementation, including the specific fixes from §4 (they explicitly call out that they supersede `planning/archive/MARKET_DATA_DESIGN.md` for exactly this reason). +- `planning/MASSIVE_API.md` accurately describes the `MassiveDataSource` implementation and correctly scopes itself as reference-only per `CLAUDE.md`/`PLAN.md` §6 (simulator is the supported path; `MASSIVE_API_KEY` stays unset). +- `backend/CLAUDE.md` and `backend/README.md` match the actual public API surface (`app.market` re-exports) and test/lint commands — both run clean as documented. + +--- + +## 6. Verdict + +The market data subsystem remains solid, well-tested, and accurately documented (with the one stale-numbers exception in §5.1). All issues from the prior review are genuinely fixed, not just marked fixed. Nothing found in this pass rises above "low severity" / "nice to have": + +**Worth doing, not urgent:** +1. Add an ASGI-level integration test for `stream.py` (recipe already exists in `planning/MARKET_DATA_DESIGN.md` §12.4) — this is the one real, still-open coverage gap, and it covers the subsystem's only externally-facing endpoint. +2. Update the stale coverage figures in `planning/MARKET_DATA_SUMMARY.md` (§5.1). + +**Nice to have:** +3. Fix the falsy-`0.0`-timestamp edge case in `PriceCache.update()` (§3.1) — dormant today, cheap one-line fix (`if timestamp is not None else`). +4. Make `test_exception_resilience` actually inject a failure (§3.2). +5. Widen the timing margin in `test_custom_update_interval` to remove latent flakiness risk (§3.3). +6. Run `ruff format` on the 3 flagged test files (§3.4). + +No blockers. This module is ready to be built on by the rest of the platform as-is. From 7e7069b97a70cfcb0c41aff368ac81d9969a63cd Mon Sep 17 00:00:00 2001 From: raunaksachdev Date: Tue, 25 Aug 2026 09:33:56 +0100 Subject: [PATCH 7/7] Address MARKET_DATA_REVIEW.md findings: close stream.py coverage gap, fix flaky/weak tests, fix falsy-timestamp bug - Add ASGI-route-level and generator-level tests for stream.py (0% -> 100% coverage on that module). Note: the httpx ASGITransport / FastAPI TestClient recipe suggested in MARKET_DATA_DESIGN.md deadlocks against _generate_events's infinite loop in the installed dependency versions (both buffer the full response before returning), so the tests instead drive the real generator and route handler coroutine directly. - Fix PriceCache.update() treating an explicit timestamp=0.0 as falsy and silently replacing it with time.time(). - test_exception_resilience now actually injects a step() failure and asserts the loop recovers, instead of just asserting the task is alive. - Widen the timing margin in test_custom_update_interval to remove latent flakiness risk. - ruff format the 3 flagged test files. - Add httpx as a dev dependency (needed for the new stream tests). - Update the stale coverage figures in planning/MARKET_DATA_SUMMARY.md and README.md to the current 79 tests / 99% overall. Co-Authored-By: Claude Sonnet 5 --- README.md | 2 +- backend/app/market/cache.py | 2 +- backend/pyproject.toml | 1 + backend/tests/market/test_models.py | 44 ++++-- backend/tests/market/test_simulator.py | 4 +- backend/tests/market/test_simulator_source.py | 35 +++-- backend/tests/market/test_stream.py | 144 ++++++++++++++++++ backend/uv.lock | 30 ++++ planning/MARKET_DATA_SUMMARY.md | 9 +- 9 files changed, 239 insertions(+), 32 deletions(-) create mode 100644 backend/tests/market/test_stream.py diff --git a/README.md b/README.md index 06ac55b22..d74b64c79 100644 --- a/README.md +++ b/README.md @@ -78,7 +78,7 @@ Full rationale for these choices is in [`planning/PLAN.md`](planning/PLAN.md) § ## What's built so far: market data -A self-contained market data subsystem lives in `backend/app/market/` — a `PriceCache`, a GBM-based simulator with correlated, per-sector price moves, a Massive/Polygon.io REST client behind the same interface, and an SSE stream factory. It's fully tested (73 tests, 91% coverage overall — `stream.py` is the weak spot at 33%, everything else is 94-100%) and has a standalone terminal demo: +A self-contained market data subsystem lives in `backend/app/market/` — a `PriceCache`, a GBM-based simulator with correlated, per-sector price moves, a Massive/Polygon.io REST client behind the same interface, and an SSE stream factory. It's fully tested (79 tests, 99% coverage overall — every module is 94-100%, `stream.py` included) and has a standalone terminal demo: ```bash cd backend diff --git a/backend/app/market/cache.py b/backend/app/market/cache.py index 4d0215778..03370e717 100644 --- a/backend/app/market/cache.py +++ b/backend/app/market/cache.py @@ -27,7 +27,7 @@ def update(self, ticker: str, price: float, timestamp: float | None = None) -> P If this is the first update for the ticker, previous_price == price (direction='flat'). """ with self._lock: - ts = timestamp or time.time() + ts = timestamp if timestamp is not None else time.time() prev = self._prices.get(ticker) previous_price = prev.price if prev else price diff --git a/backend/pyproject.toml b/backend/pyproject.toml index e172cca22..40dcabf23 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -18,6 +18,7 @@ dev = [ "pytest-asyncio>=0.24.0", "pytest-cov>=5.0.0", "ruff>=0.7.0", + "httpx>=0.27.0", ] [build-system] diff --git a/backend/tests/market/test_models.py b/backend/tests/market/test_models.py index 21600dfd6..1e0d3042d 100644 --- a/backend/tests/market/test_models.py +++ b/backend/tests/market/test_models.py @@ -10,7 +10,9 @@ class TestPriceUpdate: def test_price_update_creation(self): """Test basic PriceUpdate creation.""" - update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0 + ) assert update.ticker == "AAPL" assert update.price == 190.50 assert update.previous_price == 190.00 @@ -18,47 +20,65 @@ def test_price_update_creation(self): def test_change_calculation(self): """Test price change calculation.""" - update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0 + ) assert update.change == 0.50 def test_change_negative(self): """Test negative price change.""" - update = PriceUpdate(ticker="AAPL", price=189.50, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=189.50, previous_price=190.00, timestamp=1234567890.0 + ) assert update.change == -0.50 def test_change_percent_up(self): """Test percentage change calculation (up).""" - update = PriceUpdate(ticker="AAPL", price=190.00, previous_price=100.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=190.00, previous_price=100.00, timestamp=1234567890.0 + ) assert update.change_percent == 90.0 def test_change_percent_down(self): """Test percentage change calculation (down).""" - update = PriceUpdate(ticker="AAPL", price=100.00, previous_price=200.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=100.00, previous_price=200.00, timestamp=1234567890.0 + ) assert update.change_percent == -50.0 def test_change_percent_zero_previous(self): """Test percentage change with zero previous price.""" - update = PriceUpdate(ticker="AAPL", price=100.00, previous_price=0.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=100.00, previous_price=0.00, timestamp=1234567890.0 + ) assert update.change_percent == 0.0 def test_direction_up(self): """Test direction calculation (up).""" - update = PriceUpdate(ticker="AAPL", price=191.00, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=191.00, previous_price=190.00, timestamp=1234567890.0 + ) assert update.direction == "up" def test_direction_down(self): """Test direction calculation (down).""" - update = PriceUpdate(ticker="AAPL", price=189.00, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=189.00, previous_price=190.00, timestamp=1234567890.0 + ) assert update.direction == "down" def test_direction_flat(self): """Test direction calculation (flat).""" - update = PriceUpdate(ticker="AAPL", price=190.00, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=190.00, previous_price=190.00, timestamp=1234567890.0 + ) assert update.direction == "flat" def test_to_dict(self): """Test serialization to dictionary.""" - update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0 + ) result = update.to_dict() assert result["ticker"] == "AAPL" @@ -71,7 +91,9 @@ def test_to_dict(self): def test_immutability(self): """Test that PriceUpdate is immutable.""" - update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0 + ) with pytest.raises(AttributeError): update.price = 200.00 # Should raise error diff --git a/backend/tests/market/test_simulator.py b/backend/tests/market/test_simulator.py index 1845ec16b..02f7f8a89 100644 --- a/backend/tests/market/test_simulator.py +++ b/backend/tests/market/test_simulator.py @@ -126,6 +126,6 @@ def test_prices_rounded_to_two_decimals(self): result = sim.step() price_str = str(result["AAPL"]) # Check that we have at most 2 decimal places - if '.' in price_str: - decimal_part = price_str.split('.')[1] + if "." in price_str: + decimal_part = price_str.split(".")[1] assert len(decimal_part) <= 2 diff --git a/backend/tests/market/test_simulator_source.py b/backend/tests/market/test_simulator_source.py index 515ce7290..a720026e1 100644 --- a/backend/tests/market/test_simulator_source.py +++ b/backend/tests/market/test_simulator_source.py @@ -94,33 +94,44 @@ async def test_empty_start(self): await source.stop() async def test_exception_resilience(self): - """Test that simulator continues running after errors.""" + """Test that the loop survives a step() failure and keeps ticking.""" cache = PriceCache() - source = SimulatorDataSource(price_cache=cache, update_interval=0.05) - - # Start with a valid ticker + source = SimulatorDataSource(price_cache=cache, update_interval=0.02) await source.start(["AAPL"]) - # Wait for some updates + real_step = source._sim.step + calls = {"count": 0} + + def flaky_step(): + calls["count"] += 1 + if calls["count"] == 1: + raise RuntimeError("simulated step failure") + return real_step() + + source._sim.step = flaky_step + + # Wait long enough for the failing tick plus subsequent successful ticks await asyncio.sleep(0.15) - # Task should still be running + # Task survived the exception and kept running assert source._task is not None assert not source._task.done() + # And it kept producing updates after the injected failure + assert calls["count"] > 1 await source.stop() async def test_custom_update_interval(self): """Test using a custom update interval.""" cache = PriceCache() - source = SimulatorDataSource(price_cache=cache, update_interval=0.01) + source = SimulatorDataSource(price_cache=cache, update_interval=0.02) await source.start(["AAPL"]) initial_version = cache.version - await asyncio.sleep(0.05) # Should get ~5 updates + await asyncio.sleep(0.15) - # Should have multiple updates with fast interval - assert cache.version > initial_version + 2 + # Should have gotten at least one update with the fast interval + assert cache.version > initial_version await source.stop() @@ -128,9 +139,7 @@ async def test_custom_event_probability(self): """Test creating source with custom event probability.""" cache = PriceCache() # Very high event probability for testing - source = SimulatorDataSource( - price_cache=cache, update_interval=0.1, event_probability=1.0 - ) + source = SimulatorDataSource(price_cache=cache, update_interval=0.1, event_probability=1.0) await source.start(["AAPL"]) # Just verify it starts and stops cleanly diff --git a/backend/tests/market/test_stream.py b/backend/tests/market/test_stream.py new file mode 100644 index 000000000..fdf47504b --- /dev/null +++ b/backend/tests/market/test_stream.py @@ -0,0 +1,144 @@ +"""Tests for the SSE streaming endpoint (app/market/stream.py). + +Note: full HTTP round-trip testing via httpx's ASGITransport (or FastAPI's +TestClient, which is built on the same buffering transport in the installed +dependency versions) is not viable here — both drive the ASGI app to +completion inside a single `await` before returning any response to the +caller, which deadlocks against `_generate_events`'s infinite loop. Instead +these tests drive the real generator and route-handler coroutine directly, +which exercises the same code paths without relying on a transport that +can stream partial responses. +""" + +import asyncio +import json + +import pytest +from fastapi.responses import StreamingResponse + +from app.market.cache import PriceCache +from app.market.stream import _generate_events, create_stream_router + + +class _FakeClient: + def __init__(self, host: str = "test-client") -> None: + self.host = host + + +class _FakeRequest: + """Minimal stand-in for fastapi.Request exposing what _generate_events uses.""" + + def __init__(self, disconnect_after: int | None = None) -> None: + self.client = _FakeClient() + self._calls = 0 + self._disconnect_after = disconnect_after + + async def is_disconnected(self) -> bool: + self._calls += 1 + if self._disconnect_after is not None and self._calls > self._disconnect_after: + return True + return False + + +@pytest.mark.asyncio +async def test_generate_events_yields_retry_directive_first(): + cache = PriceCache() + request = _FakeRequest(disconnect_after=0) + gen = _generate_events(cache, request, interval=0.01) + + first = await gen.__anext__() + assert first == "retry: 1000\n\n" + + with pytest.raises(StopAsyncIteration): + await gen.__anext__() + + +@pytest.mark.asyncio +async def test_generate_events_emits_prices_on_version_change(): + cache = PriceCache() + cache.update("AAPL", 190.50) + request = _FakeRequest(disconnect_after=1) + gen = _generate_events(cache, request, interval=0.01) + + await gen.__anext__() # retry directive + data_event = await gen.__anext__() + + assert data_event.startswith("data: ") + payload = json.loads(data_event.removeprefix("data: ").strip()) + assert payload["AAPL"]["price"] == 190.50 + + with pytest.raises(StopAsyncIteration): + await gen.__anext__() + + +@pytest.mark.asyncio +async def test_generate_events_sends_no_data_event_when_cache_empty(): + cache = PriceCache() + request = _FakeRequest(disconnect_after=2) + gen = _generate_events(cache, request, interval=0.01) + + events = [event async for event in gen] + + # Only the retry directive - no data event, since the cache never had prices + assert events == ["retry: 1000\n\n"] + + +@pytest.mark.asyncio +async def test_generate_events_skips_unchanged_version(): + """A second tick with no cache write should not repeat the data event.""" + cache = PriceCache() + cache.update("AAPL", 100.0) + request = _FakeRequest(disconnect_after=2) + gen = _generate_events(cache, request, interval=0.01) + + events = [event async for event in gen] + + # retry directive + exactly one data event (version unchanged on 2nd tick) + assert len(events) == 2 + assert events[0] == "retry: 1000\n\n" + assert events[1].startswith("data: ") + + +@pytest.mark.asyncio +async def test_generate_events_stops_on_cancellation(): + """_generate_events catches CancelledError internally (to log a clean + disconnect) rather than propagating it, so the consuming task finishes + normally instead of raising or hanging.""" + cache = PriceCache() + cache.update("AAPL", 100.0) + request = _FakeRequest() # never disconnects on its own + + async def consume(): + async for _ in _generate_events(cache, request, interval=0.05): + pass + + task = asyncio.create_task(consume()) + await asyncio.sleep(0.05) + task.cancel() + + await asyncio.wait_for(task, timeout=1) + assert task.done() + assert not task.cancelled() + + +@pytest.mark.asyncio +async def test_stream_prices_route_returns_streaming_response(): + """create_stream_router wires the route to a StreamingResponse over _generate_events.""" + cache = PriceCache() + cache.update("AAPL", 190.50) + router = create_stream_router(cache) + endpoint = router.routes[-1].endpoint + + response = await endpoint(_FakeRequest()) + try: + assert isinstance(response, StreamingResponse) + assert response.media_type == "text/event-stream" + assert response.headers["cache-control"] == "no-cache" + assert response.headers["x-accel-buffering"] == "no" + + first = await response.body_iterator.__anext__() + assert first == "retry: 1000\n\n" + second = await response.body_iterator.__anext__() + assert second.startswith("data: ") + finally: + await response.body_iterator.aclose() diff --git a/backend/uv.lock b/backend/uv.lock index 67d471b2d..fd4977954 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -177,6 +177,7 @@ dependencies = [ [package.optional-dependencies] dev = [ + { name = "httpx" }, { name = "pytest" }, { name = "pytest-asyncio" }, { name = "pytest-cov" }, @@ -186,6 +187,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "fastapi", specifier = ">=0.115.0" }, + { name = "httpx", marker = "extra == 'dev'", specifier = ">=0.27.0" }, { name = "massive", specifier = ">=1.0.0" }, { name = "numpy", specifier = ">=2.0.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.3.0" }, @@ -206,6 +208,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" }, ] +[[package]] +name = "httpcore" +version = "1.0.9" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "certifi" }, + { name = "h11" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/94/82699a10bca87a5556c9c59b5963f2d039dbd239f25bc2a63907a05a14cb/httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8", size = 85484, upload-time = "2025-04-24T22:06:22.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" }, +] + [[package]] name = "httptools" version = "0.7.1" @@ -235,6 +250,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/53/cf/878f3b91e4e6e011eff6d1fa9ca39f7eb17d19c9d7971b04873734112f30/httptools-0.7.1-cp314-cp314-win_amd64.whl", hash = "sha256:cfabda2a5bb85aa2a904ce06d974a3f30fb36cc63d7feaddec05d2050acede96", size = 88205, upload-time = "2025-10-10T03:55:00.389Z" }, ] +[[package]] +name = "httpx" +version = "0.28.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "certifi" }, + { name = "httpcore" }, + { name = "idna" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b1/df/48c586a5fe32a0f01324ee087459e112ebb7224f646c0b5023f5e79e9956/httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc", size = 141406, upload-time = "2024-12-06T15:37:23.222Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" }, +] + [[package]] name = "idna" version = "3.11" diff --git a/planning/MARKET_DATA_SUMMARY.md b/planning/MARKET_DATA_SUMMARY.md index ae518283a..b23f5cefa 100644 --- a/planning/MARKET_DATA_SUMMARY.md +++ b/planning/MARKET_DATA_SUMMARY.md @@ -44,18 +44,19 @@ MarketDataSource (ABC) ## Test Suite -**73 tests, all passing.** 6 test modules in `backend/tests/market/`. +**79 tests, all passing.** 7 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.py | 19 | simulator.py: 99% | | 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) | +| test_massive.py | 13 | massive_client.py: 94% (real `massive` package installed; only the real-API-call bodies are unmocked/uncovered) | +| test_stream.py | 6 | stream.py: 100% | -Overall coverage: 84%. +Overall coverage: 99%. ## Code Review & Fixes Applied