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/settings.json b/.claude/settings.json index aa06f43dc..c72c6b731 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -1,7 +1,3 @@ { - "enabledPlugins": { - "frontend-design@claude-plugins-official": true, - "context7@claude-plugins-official": true, - "playwright@claude-plugins-official": true - } + "enabledPlugins": {} } 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 *)' diff --git a/README.md b/README.md index 3f2582ae2..0bd3cd4e8 100644 --- a/README.md +++ b/README.md @@ -1,62 +1,82 @@ # FinAlly — AI Trading Workstation -A visually stunning AI-powered trading workstation that streams live market data, simulates portfolio trading, and integrates an LLM chat assistant that can analyze positions and execute trades via natural language. +A Bloomberg-style trading terminal with an AI copilot: live streaming prices, a simulated +$10k portfolio, and an LLM chat assistant that can analyse positions and execute trades from +natural language. -Built entirely by coding agents as a capstone project for an agentic AI coding course. +Built entirely by coding agents as the capstone project for an agentic AI coding course. +Agents coordinate through the shared documents in `planning/`. -## 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 +Market data subsystem is complete and tested (73 tests). The rest of the platform is in +development. + +| Component | State | +|---|---| +| Market data (simulator, Massive client, price cache, SSE stream) | Built | +| Database, portfolio and watchlist APIs | Planned | +| Next.js frontend, charts, chat panel | Planned | +| Docker image and start/stop scripts | Planned | ## Architecture -Single Docker container serving everything on port 8000: +One container, one port. FastAPI serves the REST API, the SSE price stream, and the exported +Next.js frontend as static files on port 8000. + +- **Backend** — FastAPI, Python 3.12, managed with `uv` +- **Frontend** — Next.js static export, TypeScript, Tailwind, Recharts +- **Database** — SQLite at `db/finally.db`, lazily created and seeded +- **Real-time** — Server-Sent Events, one event per tick carrying every tracked ticker +- **AI** — LiteLLM to OpenRouter (Cerebras inference) with structured outputs +- **Market data** — built-in GBM simulator by default, Massive (Polygon.io) if a key is set -- **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) +## Running the Market Data Demo -## Quick Start +A Rich terminal dashboard of the live simulated price stream: ```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 +``` -# Run with Docker -docker build -t finally . -docker run -v finally-data:/app/db -p 8000:8000 --env-file .env finally +## Tests -# Open http://localhost:8000 +```bash +cd backend +uv run pytest +uv run ruff check . ``` ## 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 `.env` in the project root. Every variable is optional — with none set, the app runs on +the simulator and chat reports that no API key is configured. + +| Variable | Description | +|---|---| +| `OPENROUTER_API_KEY` | Enables the AI chat panel | +| `MASSIVE_API_KEY` | Use real market data instead of the simulator | +| `LLM_MOCK` | Set `true` for deterministic mock LLM responses in tests | ## Project Structure ``` finally/ +├── backend/ # FastAPI uv project (app/market/ is built) ├── 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 +├── planning/ # PLAN.md and agent reference docs +├── test/ # Playwright E2E tests, run on the host +├── scripts/ # Docker start/stop helpers +└── db/ # SQLite bind mount at runtime ``` +## Documentation + +- `planning/PLAN.md` — the full specification and the authority for all agent work +- `planning/MARKET_DATA_SUMMARY.md` — what the market data subsystem does and how + ## License See [LICENSE](LICENSE). diff --git a/planning/MARKET_DATA_DESIGN.md b/planning/MARKET_DATA_DESIGN.md new file mode 100644 index 000000000..0299bb598 --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,1949 @@ +# Market Data Backend — Detailed Design + +The implementation-level design for `backend/app/market/`: the subsystem that produces live +prices for FinAlly, whether they come from the built-in simulator or the Massive REST API, and +publishes them to every consumer through one cache and one SSE stream. + +This document is the buildable form of three companion documents — `MARKET_INTERFACE.md` (the +contract), `MARKET_SIMULATOR.md` (the default provider), `MASSIVE_API.md` (the real provider) — +plus sections 6 and 13 of `PLAN.md`. Where they describe *what* and *why*, this describes *what +the code is*. Every snippet below is intended to be typed into the repository as written. + +**Status.** Most of this module already exists and is tested (73 tests, 84% coverage — see +`MARKET_DATA_SUMMARY.md`). What is missing is the session baseline, sparkline history, and a +handful of correctness fixes. Section 15 is the precise delta from the code currently on disk; +if you are implementing rather than reading, start there and use sections 4–13 as the target +state. + +--- + +## Table of Contents + +1. [Scope and responsibilities](#1-scope-and-responsibilities) +2. [Architecture](#2-architecture) +3. [Module map](#3-module-map) +4. [`models.py` — PriceUpdate](#4-modelspy--priceupdate) +5. [`cache.py` — PriceCache](#5-cachepy--pricecache) +6. [`tickers.py` — validation](#6-tickerspy--validation) +7. [`interface.py` — MarketDataSource](#7-interfacepy--marketdatasource) +8. [`seed_prices.py` — parameters](#8-seed_pricespy--parameters) +9. [`simulator.py` — the default source](#9-simulatorpy--the-default-source) +10. [`massive_client.py` — the real source](#10-massive_clientpy--the-real-source) +11. [`factory.py` — the switch](#11-factorypy--the-switch) +12. [`stream.py` — SSE](#12-streampy--sse) +13. [Wiring into FastAPI](#13-wiring-into-fastapi) +14. [Consumer recipes](#14-consumer-recipes) +15. [Delta from the current implementation](#15-delta-from-the-current-implementation) +16. [Testing](#16-testing) +17. [Failure modes](#17-failure-modes) + +--- + +## 1. Scope and responsibilities + +**In scope for `app/market/`:** + +- Producing a current price for every tracked ticker, from one of two sources +- Holding those prices, their session baselines, and ~60 points of recent history +- Validating ticker symbols (one rule, shared by every caller) +- Streaming price changes to browsers over SSE +- Deciding which source runs, from one environment variable + +**Explicitly out of scope** — these belong to the portfolio, watchlist, and chat routers: + +- Persistence. Nothing in this module touches SQLite. Prices are process state and are meant to + be lost on restart. +- The watchlist itself. The database owns the list of tickers; this module is *told* which + tickers to track, at startup and on change. +- Trade execution, valuation, P&L. Those read the cache; the cache does not know they exist. + +The load-bearing rule: **no code outside `app/market/` knows which source is running.** A +consumer holds a `PriceCache` and reads it. It cannot tell a 500ms simulator from a 15-second +poller, and it has no API through which to ask. + +--- + +## 2. Architecture + +``` + MASSIVE_API_KEY set? + │ + ┌─────────────────┴─────────────────┐ + │ yes no │ + ▼ ▼ + MassiveDataSource SimulatorDataSource + REST snapshot poll, 15s GBM step, 500ms, in-process + asyncio.to_thread (SDK is sync) pure numpy, no I/O + │ │ + └─────────────────┬─────────────────┘ + │ cache.update(ticker, price, timestamp) + ▼ + ┌───────────────────────┐ + │ PriceCache │ latest / previous / open price + │ threading.Lock │ version counter + │ version: int │ deque(maxlen=60) history + └───────────┬───────────┘ + │ read-only + ┌──────────────┬────────┴────────┬──────────────────┐ + ▼ ▼ ▼ ▼ + SSE stream portfolio trade fill /api/watchlist + /api/stream/… valuation price lookup price + history +``` + +Three properties fall out of this shape, and each is worth stating because breaking any one of +them breaks the module: + +**Producers push, consumers pull.** The source never returns a price to a caller; it writes to +the cache on its own schedule. That inversion is what makes a 500ms producer and a 15-second +producer interchangeable — the reader always finds a current price and never learns how old it +is or where it came from. + +**One writer at a time, many readers, always.** The lock is a `threading.Lock` rather than an +`asyncio.Lock` because the Massive poller writes from a worker thread (`asyncio.to_thread`) +while the simulator writes from the event loop. An `asyncio.Lock` would be silently wrong for +the threaded path. + +**Change is a number, not an event.** The cache carries a monotonic `version`. The SSE generator +remembers the version it last sent. No pub/sub, no per-client queues, no fan-out bookkeeping — +adding a client costs one integer comparison every 500ms. + +--- + +## 3. Module map + +``` +backend/app/market/ +├── __init__.py Public API re-exports +├── models.py PriceUpdate ~70 lines +├── cache.py PriceCache, wait_for_price ~150 lines +├── tickers.py TICKER_PATTERN, normalize_ticker ~25 lines +├── interface.py MarketDataSource (ABC) ~70 lines +├── seed_prices.py Seed prices, GBM params, groups, synthesize ~90 lines +├── simulator.py GBMSimulator + SimulatorDataSource ~230 lines +├── massive_client.py MassiveDataSource ~200 lines +├── factory.py create_market_data_source() ~35 lines +└── stream.py create_stream_router() ~90 lines +``` + +```python +# app/market/__init__.py +"""Market data subsystem for FinAlly.""" + +from .cache import PriceCache, wait_for_price +from .factory import create_market_data_source +from .interface import MarketDataSource +from .models import PriceUpdate +from .stream import create_stream_router +from .tickers import TICKER_PATTERN, normalize_ticker + +__all__ = [ + "MarketDataSource", + "PriceCache", + "PriceUpdate", + "TICKER_PATTERN", + "create_market_data_source", + "create_stream_router", + "normalize_ticker", + "wait_for_price", +] +``` + +Concrete source classes (`SimulatorDataSource`, `MassiveDataSource`, `GBMSimulator`) are +deliberately *not* re-exported. Application code should never name them — it calls the factory +and holds a `MarketDataSource`. Tests import them from their modules directly. + +--- + +## 4. `models.py` — PriceUpdate + +One ticker, one moment, immutable. + +```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. + + Constructed only by PriceCache.update(), which supplies previous_price and + open_price from the entry it is replacing. Sources never build one directly. + """ + + ticker: str + price: float + previous_price: float + open_price: float + timestamp: float = field(default_factory=time.time) # Unix epoch seconds + + # --- Derived: tick over tick --- + + @property + def change(self) -> float: + """Absolute price change since the previous tick.""" + return round(self.price - self.previous_price, 4) + + @property + def change_percent(self) -> float: + """Percent change since the previous tick. Drives the flash animation only.""" + 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' since the previous tick.""" + if self.price > self.previous_price: + return "up" + if self.price < self.previous_price: + return "down" + return "flat" + + # --- Derived: against the session baseline --- + + @property + def change_from_open(self) -> float: + """Absolute price change since the session open.""" + return round(self.price - self.open_price, 4) + + @property + def change_from_open_percent(self) -> float: + """Percent change since the session open. This is the user-facing 'change %'.""" + if self.open_price == 0: + return 0.0 + return round((self.price - self.open_price) / self.open_price * 100, 4) + + def to_dict(self) -> dict: + """Serialise for JSON / SSE. Keys match the payload in PLAN.md section 6.""" + return { + "ticker": self.ticker, + "price": self.price, + "previous_price": self.previous_price, + "open_price": self.open_price, + "timestamp": self.timestamp, + "change": self.change, + "change_percent": self.change_percent, + "change_from_open_percent": self.change_from_open_percent, + "direction": self.direction, + } +``` + +### Two "change" numbers, and which one the UI uses + +`change_percent` compares against the previous tick. At 500ms that number flickers around zero +and is meaningless as a daily-change column — it reports how the last half-second went. + +`open_price` is the **session baseline**: the first price seen for a ticker after process start, +or after the ticker was added to a running system. `change_from_open_percent` measures against +it, which is the number a user recognises as "how is it doing today". + +| Consumer | Field | +|---|---| +| Watchlist "change %" column | `change_from_open_percent` | +| Price colour (green/red text) | `change_from_open_percent` | +| Flash animation on tick | `direction` / `change_percent` | +| Main chart, sparkline | `price` | + +The baseline survives page reloads and SSE reconnects because it lives on the server, and resets +on container restart. For a simulation that is the right trade: no persistence, no market +calendar, no timezone logic. + +### Why frozen and slotted + +A `PriceUpdate` is handed to JSON serialisation on one task and to portfolio arithmetic on +another, concurrently. Nothing should be able to mutate one after the cache published it — +`frozen=True` makes that a `FrozenInstanceError` instead of a heisenbug. `slots=True` drops the +per-instance `__dict__`, which matters when 10–50 of these are allocated twice a second for the +life of the container. + +### Rounding convention + +`price` and `previous_price` and `open_price` are rounded to 2dp **by the cache** before the +dataclass is built, so every consumer sees the same cent value and no consumer needs to round. +Derived percentages round to 4dp — enough to show `+0.687%` without a wall of float noise. + +--- + +## 5. `cache.py` — PriceCache + +The single point of truth. Everything else in the app reads prices from here. + +```python +"""Thread-safe in-memory price cache.""" + +from __future__ import annotations + +import asyncio +import time +from collections import deque +from threading import Lock + +from .models import PriceUpdate + +HISTORY_POINTS = 60 # Points retained per ticker for sparklines +HISTORY_INTERVAL_SECONDS = 60.0 # Minimum spacing between recorded points + + +class PriceCache: + """Latest price, session baseline and recent history for each tracked ticker. + + Writers: exactly one MarketDataSource (simulator or Massive poller). + Readers: SSE stream, portfolio valuation, trade execution, watchlist. + """ + + def __init__( + self, + history_points: int = HISTORY_POINTS, + history_interval: float = HISTORY_INTERVAL_SECONDS, + ) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._history: dict[str, deque[float]] = {} + self._history_at: dict[str, float] = {} + self._history_points = history_points + self._history_interval = history_interval + self._lock = Lock() + self._version = 0 + + # --- Writing --- + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + """Record a new price. Returns the stored PriceUpdate. + + Derives previous_price and open_price from the entry being replaced, so + sources stay dumb: they produce a number, the cache supplies the meaning. + The first update for a ticker pins its session baseline — + previous_price == open_price == price, direction 'flat', both changes 0. + """ + with self._lock: + ts = time.time() if timestamp is None else timestamp + price = round(price, 2) + previous = self._prices.get(ticker) + + update = PriceUpdate( + ticker=ticker, + price=price, + previous_price=previous.price if previous else price, + open_price=previous.open_price if previous else price, + timestamp=ts, + ) + self._prices[ticker] = update + + # Version tracks *visible* change. A repeated price refreshes the + # timestamp (so /api/health still sees a live feed) without waking + # every SSE client to re-send an identical payload. + if previous is None or previous.price != price: + self._version += 1 + + self._record_history(ticker, update) + return update + + def seed_history( + self, ticker: str, prices: list[float], timestamp: float | None = None + ) -> None: + """Install backfilled history for a ticker, replacing anything present. + + Called by a source at startup and when a ticker is added, so sparklines + are populated on first paint rather than filling in over 30 seconds. + """ + with self._lock: + self._history[ticker] = deque( + (round(p, 2) for p in prices[-self._history_points :]), + maxlen=self._history_points, + ) + self._history_at[ticker] = time.time() if timestamp is None else timestamp + + def remove(self, ticker: str) -> None: + """Forget a ticker entirely — price, baseline and history.""" + with self._lock: + self._prices.pop(ticker, None) + self._history.pop(ticker, None) + self._history_at.pop(ticker, None) + + # --- Reading --- + + def get(self, ticker: str) -> PriceUpdate | None: + with self._lock: + return self._prices.get(ticker) + + def get_price(self, ticker: str) -> float | None: + update = self.get(ticker) + return update.price if update else None + + def get_all(self) -> dict[str, PriceUpdate]: + """Shallow copy of every current price. Safe to iterate without the lock.""" + with self._lock: + return dict(self._prices) + + def get_history(self, ticker: str) -> list[float]: + """Recent prices, oldest first, up to history_points. Empty if unknown.""" + with self._lock: + history = self._history.get(ticker) + return list(history) if history else [] + + def newest_timestamp(self) -> float | None: + """Timestamp of the most recently written price, for /api/health.""" + with self._lock: + if not self._prices: + return None + return max(update.timestamp for update in self._prices.values()) + + @property + def version(self) -> int: + """Monotonic counter, bumped whenever a price actually changes.""" + 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 + + # --- Internal (callers already hold the lock) --- + + def _record_history(self, ticker: str, update: PriceUpdate) -> None: + history = self._history.get(ticker) + if history is None: + history = self._history[ticker] = deque(maxlen=self._history_points) + + last_at = self._history_at.get(ticker) + if last_at is None or update.timestamp - last_at >= self._history_interval: + history.append(update.price) + self._history_at[ticker] = update.timestamp +``` + +### Why `update()` computes the derived fields + +Callers pass a raw float. The cache looks up the previous entry, carries `open_price` forward, +and constructs the `PriceUpdate`. Sources stay dumb — they fetch or generate a number and hand +it over — and all the semantics live in exactly one function. The alternative, sources building +their own `PriceUpdate`s, means the simulator and the Massive client each have their own copy of +the baseline rule and they drift within a month. + +### Why the history cadence is one point per minute + +The sparkline holds 60 points. If every 500ms tick appended one, the entire series would span 30 +seconds and would overwrite the backfill within half a minute — the sparkline would show noise +instead of shape, and the backfill would have been pointless. + +`HISTORY_INTERVAL_SECONDS = 60.0` matches the cadence the simulator's backfill generates (see +§9), so the seeded points and the live points are on the same time axis and a 60-point series +always means "the last hour". The frontend extends its own copy live from the SSE stream, which +is what makes the sparkline move between reloads; the server's copy exists only so a fresh page +paints something real. + +Tests pass `history_interval=0.0` to record every update. + +### Why `version` only moves on a real change + +`PLAN.md` requires that a quiet market produce no price events. Off-hours on the Massive path, +every 15-second poll returns the same last trade; bumping the version on those writes would push +an identical payload to every client four times a minute and make the "emit only on change" +guarantee a lie. Comparing the rounded price is the cheapest honest test — sub-cent simulator +jitter on a low-volatility name like V is also correctly treated as no change. + +The timestamp is still refreshed on a repeated price, so `/api/health` can distinguish "the feed +is running and the price is flat" from "the feed died twenty minutes ago". + +### `wait_for_price` — the one asynchronous helper + +A just-added ticker has no price for a few hundred milliseconds. Failing the trade would be a +poor experience for the headline demo ("ask the AI to buy something it just added"), so trade +execution waits briefly instead. + +```python +async def wait_for_price(cache: PriceCache, ticker: str, timeout: float = 2.0) -> float: + """Return the current price, waiting up to `timeout` for a first tick. + + Raises ValueError with a user-facing message if no price arrives. Callers + translate that into a 400 (PLAN.md section 8). + """ + deadline = time.monotonic() + timeout + while True: + price = cache.get_price(ticker) + if price is not None: + return price + if time.monotonic() >= deadline: + raise ValueError(f"No price available for {ticker} yet, please try again") + await asyncio.sleep(0.2) +``` + +On the simulator path this never expires: `add_ticker()` seeds the cache synchronously before it +returns. On the Massive path with a 15-second poll it genuinely can, which is why the message is +retryable rather than an error about an unknown symbol. + +--- + +## 6. `tickers.py` — validation + +Manual adds, LLM-driven adds, watchlist deletes and trades all funnel through one function. + +```python +"""Ticker symbol validation — one rule, shared by every caller.""" + +from __future__ import annotations + +import re + +TICKER_PATTERN = re.compile(r"^[A-Z]{1,5}$") + + +def normalize_ticker(raw: str) -> str: + """Uppercase, strip, and validate a ticker symbol. + + Raises ValueError if the symbol is not 1-5 A-Z characters. Callers turn + that into a 400 with the message shown to the user verbatim. + """ + ticker = raw.strip().upper() + if not TICKER_PATTERN.match(ticker): + raise ValueError(f"Invalid ticker symbol: {raw!r}") + return ticker +``` + +Uppercase first, then match — so `aapl` is accepted and normalised, while `hello world`, `12345` +and `""` are rejected. One function, one regex, one error message, so the manual path and the LLM +path cannot drift apart. + +This validates **shape, not existence**. `ZZZZZ` is accepted and gets a synthesised price (§8). +The simulator has no universe of real symbols and could not do otherwise; rejecting unknown +symbols would dead-end "ask the AI to watch a new stock" on the first plausible thing anyone +tries. + +--- + +## 7. `interface.py` — MarketDataSource + +```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 prices into a shared PriceCache on their own schedule. + Downstream code never asks a source for a price — it reads the cache. + + Lifecycle: + source = create_market_data_source(cache) + await source.start(["AAPL", "GOOGL", ...]) # cache populated on return + await source.add_ticker("TSLA") + await source.remove_ticker("GOOGL") + await source.stop() # idempotent + """ + + @property + @abstractmethod + def source_name(self) -> str: + """Short identifier for logs and /api/health: 'simulator' or 'massive'.""" + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing prices for `tickers`. + + Must populate the cache (prices and seeded history) before returning, so + the first HTTP request never sees an empty cache. Called exactly once. + """ + + @abstractmethod + async def stop(self) -> None: + """Stop the background task. Idempotent, and never writes afterwards.""" + + @abstractmethod + async def add_ticker(self, ticker: str) -> None: + """Track a ticker. No-op if already tracked. Seeds price and history.""" + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Stop tracking a ticker and remove it from the cache. No-op if absent.""" + + @abstractmethod + def get_tickers(self) -> list[str]: + """Currently tracked tickers. Synchronous — called from request handlers.""" +``` + +Five methods and one property. Notice what is missing: **there is no `get_price()`**. A source +cannot be asked for a price. Omitting the read method is what enforces the architecture — a +consumer that tries to bypass the cache finds there is no way to. + +`add_ticker` and `remove_ticker` are `async` because the Massive implementation awaits a history +backfill request; the simulator's are trivially async. `get_tickers` is synchronous because it +reads local state from inside request handlers. + +`source_name` exists so `/api/health` can report which provider is running without an `isinstance` +check leaking concrete classes into the router — and because "why are prices not moving" is the +most likely support question in this project. + +--- + +## 8. `seed_prices.py` — parameters + +```python +"""Seed prices, GBM parameters and correlation groups for the simulator.""" + +from __future__ import annotations + +import hashlib + +# Recognisable rather than current. This is a simulation with pretend money. +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, +} + +# sigma: annualised volatility. mu: annualised drift. +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}, +} + +# Correlation groups. TSLA is deliberately in neither: nominally tech, famously +# does its own thing, and its independence gives the watchlist texture. +CORRELATION_GROUPS: dict[str, set[str]] = { + "tech": {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"}, + "finance": {"JPM", "V"}, +} + +INTRA_TECH_CORR = 0.6 # Tech names move together +INTRA_FINANCE_CORR = 0.5 # Finance names move together +CROSS_GROUP_CORR = 0.3 # Across sectors, TSLA, and synthesised tickers + + +def synthesize_params(ticker: str) -> tuple[float, dict[str, float]]: + """Derive a stable seed price and GBM parameters from the symbol itself. + + Deterministic by construction: SHA-256 of the symbol, never random. A user + holding 10 shares of PYPL bought at $73 must not restart the container and + find PYPL trading at $412 — their P&L would be nonsense. + + Ranges are chosen so every synthesised ticker looks like an ordinary + large-cap: $20-$500, sigma 0.15-0.50, mu 0.02-0.08. + """ + digest = hashlib.sha256(ticker.encode()).digest() + price = 20.0 + (int.from_bytes(digest[0:4], "big") % 48_000) / 100.0 + sigma = 0.15 + (digest[4] / 255.0) * 0.35 + mu = 0.02 + (digest[5] / 255.0) * 0.06 + return round(price, 2), {"sigma": round(sigma, 4), "mu": round(mu, 4)} + + +def params_for(ticker: str) -> tuple[float, dict[str, float]]: + """Seed price and GBM parameters for any well-formed ticker.""" + if ticker in SEED_PRICES: + return SEED_PRICES[ticker], dict(TICKER_PARAMS[ticker]) + return synthesize_params(ticker) +``` + +### The spread in sigma is the point + +TSLA at 0.50 against V at 0.17 means TSLA visibly jumps while V barely moves — the watchlist has +texture instead of ten lines doing the same thing. NVDA carries the strongest drift so something +in the portfolio tends to trend upward, which makes the P&L chart more interesting than a random +walk around zero. + +### Positive definiteness + +`np.linalg.cholesky` raises on a matrix that is not positive definite, and correlations assigned +pairwise by ad-hoc rules are not guaranteed to be. The structure above is safe, and it is worth +recording why so that a future edit can check itself. + +With TSLA excluded from both groups, the matrix decomposes as + +``` +C = 0.3·J + 0.3·J_tech + 0.2·J_finance + diag(residual) +``` + +where `J` is the all-ones matrix and `J_tech` / `J_finance` are all-ones on their block. Each +term is positive semi-definite, and every residual on the diagonal is strictly positive +(0.4 for tech, 0.5 for finance, 0.7 for TSLA and synthesised names), so `C` is positive definite. + +The invariant to preserve: **within-group correlation ≥ cross-group correlation, and the sum of +the block values on any diagonal entry < 1.** Anything more elaborate — high correlations, more +groups, exceptions layered on exceptions — needs either a check or a nearest-positive-definite +repair. §9 keeps a defensive fallback for the day someone tries. + +--- + +## 9. `simulator.py` — the default source + +The simulator is not a fallback for people without an API key. It is the better demo: prices +always move, there are no market hours, no latency and no signup. A student running the app at +21:00 on a Sunday against real data sees ten flat prices and reasonably concludes it is broken. + +The file splits into the maths (pure, synchronous, testable without a clock) and the plumbing +(asyncio, cache writes, no maths). + +### 9.1 `GBMSimulator` — the maths + +``` +S(t+dt) = S(t) · exp( (mu − sigma²/2)·dt + sigma·√dt·Z ) +``` + +The `−sigma²/2` correction is not cosmetic. Without it `mu` is the drift of *log* price and the +expected price grows faster than `mu`, so prices inflate visibly over a long session. + +`dt` must be a fraction of a trading year because `mu` and `sigma` are annualised. A trading year +is 252 days of 6.5 hours; a 500ms tick is therefore ~8.48e-8 of one. That tiny `dt` is what makes +the output look right: a $190 stock at sigma 0.22 moves on the order of a cent per tick and +wanders a few tenths of a percent over a minute. Using `dt = 0.5` (half a *year* per tick) sends +prices to five figures inside a minute — the classic failure here. + +```python +"""GBM-based market simulator.""" + +from __future__ import annotations + +import asyncio +import logging +import math +import random + +import numpy as np + +from .cache import HISTORY_POINTS, PriceCache +from .interface import MarketDataSource +from .seed_prices import ( + CORRELATION_GROUPS, + CROSS_GROUP_CORR, + INTRA_FINANCE_CORR, + INTRA_TECH_CORR, + params_for, +) + +logger = logging.getLogger(__name__) + +TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 +DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8 for a 500ms tick +HISTORY_STEP_TICKS = 120 # One backfill point per minute + + +class GBMSimulator: + """Correlated geometric Brownian motion over a set of tickers. + + Pure and synchronous: holds prices, parameters and the Cholesky factor, and + knows nothing about the cache, asyncio or FastAPI. Tests drive step() + directly with a large dt instead of sleeping. + """ + + def __init__( + self, + tickers: list[str], + dt: float = DEFAULT_DT, + event_probability: float = 0.001, + ) -> None: + self._dt = dt + self._event_prob = event_probability + self._tickers: list[str] = [] + self._prices: dict[str, float] = {} + self._params: dict[str, dict[str, float]] = {} + self._cholesky: np.ndarray | None = None + + for ticker in tickers: + self._add_internal(ticker) + self._rebuild_cholesky() + + # --- Public API --- + + def step(self) -> dict[str, float]: + """Advance every ticker one tick. Returns {ticker: rounded price}. + + The hot path: called every 500ms forever. All n normals are drawn in one + numpy call and correlated with one matrix multiply rather than looping. + """ + n = len(self._tickers) + if n == 0: + return {} + + z = np.random.standard_normal(n) + if self._cholesky is not None: + z = self._cholesky @ z + + prices: dict[str, float] = {} + for i, ticker in enumerate(self._tickers): + params = self._params[ticker] + mu, sigma = params["mu"], params["sigma"] + + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z[i] + self._prices[ticker] *= math.exp(drift + diffusion) + + # Random shock: GBM alone is smooth, real markets jump. ~0.1% per + # tick per ticker is an event every ~50s across ten tickers — often + # enough to see in a minute, rare enough to stay an event. + if random.random() < self._event_prob: + magnitude = random.uniform(0.02, 0.05) + sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + magnitude * sign + logger.debug( + "Shock event on %s: %.1f%% %s", + ticker, + magnitude * 100, + "up" if sign > 0 else "down", + ) + + prices[ticker] = round(self._prices[ticker], 2) + + return prices + + def add_ticker(self, ticker: str) -> None: + if ticker in self._prices: + return + self._add_internal(ticker) + self._rebuild_cholesky() + + def remove_ticker(self, ticker: str) -> None: + if ticker not in self._prices: + return + self._tickers.remove(ticker) + del self._prices[ticker] + del self._params[ticker] + self._rebuild_cholesky() + + def get_price(self, ticker: str) -> float | None: + price = self._prices.get(ticker) + return round(price, 2) if price is not None else None + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + def backfill_history(self, ticker: str, points: int = HISTORY_POINTS) -> list[float]: + """Manufacture plausible prior history ending at the current price. + + Runs the GBM recurrence *backwards* — dividing rather than multiplying — + so the series ends at the live price and joins the stream continuously. + The coarser dt gives a per-minute cadence, so 60 points is an hour of + price action with visible shape rather than 30 seconds of flat line. + """ + params = self._params.get(ticker) + if params is None or points < 1: + return [] + + mu, sigma = params["mu"], params["sigma"] + dt = self._dt * HISTORY_STEP_TICKS + drift = (mu - 0.5 * sigma**2) * dt + scale = sigma * math.sqrt(dt) + + price = self._prices[ticker] + history = [round(price, 2)] + for z in np.random.standard_normal(points - 1): + price /= math.exp(drift + scale * z) + history.append(round(price, 2)) + + history.reverse() # Oldest first, ending at the current price + return history + + # --- Internals --- + + def _add_internal(self, ticker: str) -> None: + """Add without rebuilding Cholesky, for batch initialisation.""" + if ticker in self._prices: + return + price, params = params_for(ticker) + self._tickers.append(ticker) + self._prices[ticker] = price + self._params[ticker] = params + + def _rebuild_cholesky(self) -> None: + """Refactor the correlation matrix. O(n^3), called only on add/remove.""" + n = len(self._tickers) + if n <= 1: + self._cholesky = None + return + + corr = np.eye(n) + for i in range(n): + for j in range(i + 1, n): + rho = self._pairwise_correlation(self._tickers[i], self._tickers[j]) + corr[i, j] = corr[j, i] = rho + + try: + self._cholesky = np.linalg.cholesky(corr) + except np.linalg.LinAlgError: + # Should be unreachable with the block structure in seed_prices.py + # (see section 8). Degrade to independent draws rather than taking + # the price feed down over a correlation constant. + logger.error("Correlation matrix not positive definite; using independent draws") + self._cholesky = None + + @staticmethod + def _pairwise_correlation(a: str, b: str) -> float: + """Sector-based correlation: tech 0.6, finance 0.5, everything else 0.3.""" + if a in CORRELATION_GROUPS["tech"] and b in CORRELATION_GROUPS["tech"]: + return INTRA_TECH_CORR + if a in CORRELATION_GROUPS["finance"] and b in CORRELATION_GROUPS["finance"]: + return INTRA_FINANCE_CORR + return CROSS_GROUP_CORR +``` + +Independent random walks look wrong. Real markets move together — when tech sells off, it sells +off broadly. Ten independently wandering lines read as noise; correlated ones read as a market. +Cholesky is how that is imposed: given `C = L·Lᵀ` and independent standard normals `z`, the +product `L @ z` has exactly the correlation structure of `C`. + +The shock line is the single highest-value line in the simulator for demo purposes. It is what +makes the flash animations fire and the P&L chart do something worth looking at. + +### 9.2 `SimulatorDataSource` — the plumbing + +```python +class SimulatorDataSource(MarketDataSource): + """MarketDataSource backed by the GBM simulator. + + Owns one asyncio task that steps the simulation every `update_interval` + seconds and writes each price to the cache. No maths, no I/O. + """ + + 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 + + @property + def source_name(self) -> str: + return "simulator" + + async def start(self, tickers: list[str]) -> None: + self._sim = GBMSimulator(tickers, event_probability=self._event_prob) + + # Populate the cache *before* the loop starts, so start() returns with + # prices and sparklines already available and the first HTTP request + # never sees an empty cache. + for ticker in tickers: + self._seed(ticker) + + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + logger.info("Simulator started: %d tickers, %.0fms interval", + len(tickers), self._interval * 1000) + + 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 is None: + return + if ticker in self._sim.get_tickers(): + return + self._sim.add_ticker(ticker) + self._seed(ticker) + logger.info("Simulator: tracking %s", ticker) + + async def remove_ticker(self, ticker: str) -> None: + if self._sim is not None: + self._sim.remove_ticker(ticker) + self._cache.remove(ticker) + logger.info("Simulator: dropped %s", ticker) + + def get_tickers(self) -> list[str]: + return self._sim.get_tickers() if self._sim else [] + + # --- Internals --- + + def _seed(self, ticker: str) -> None: + """Backfill history then publish the first price. Order matters: the + history must be in place before the price the sparkline ends at.""" + assert self._sim is not None + price = self._sim.get_price(ticker) + if price is None: + return + self._cache.seed_history(ticker, self._sim.backfill_history(ticker)) + self._cache.update(ticker, price) + + async def _run_loop(self) -> None: + while True: + try: + if self._sim is not None: + for ticker, price in self._sim.step().items(): + self._cache.update(ticker, price) + except Exception: + # A background task that raises dies silently and takes the whole + # price feed with it, leaving a UI that looks connected and + # frozen. Log and take the next tick. + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +Two behaviours carry weight: + +**The cache is seeded before the task starts.** `start()` and `add_ticker()` both return with a +price already published, which is what stops `wait_for_price` from ever mattering on the +simulator path. + +**The loop catches and continues.** It gets another chance in 500ms; dying is never the better +option. + +--- + +## 10. `massive_client.py` — the real source + +Massive (formerly Polygon.io) rebranded on 30 October 2025; existing keys and `api.polygon.io` +URLs still work. The SDK is the `massive` PyPI package, **synchronous urllib3 under the hood**. + +Three constraints shape this file: + +1. **The free tier allows 5 requests/minute.** One request per ticker is not viable. The poller + fetches every watched ticker in a *single* snapshot request and defaults to a 15-second + interval (4 requests/minute, leaving headroom). +2. **The SDK blocks.** Every call goes through `asyncio.to_thread`. A blocking HTTP call on the + event loop freezes every SSE connection and every API request for its duration. +3. **A market data failure must never take down the app.** The loop catches broadly, logs, backs + off, and lives to poll again. Stale prices are strictly better than a 500 on every request. + +```python +"""Massive (Polygon.io) REST client for real market data.""" + +from __future__ import annotations + +import asyncio +import datetime as dt +import logging +import time + +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +from .cache import HISTORY_POINTS, PriceCache +from .interface import MarketDataSource + +logger = logging.getLogger(__name__) + +MAX_BACKOFF_MULTIPLIER = 8.0 +BACKFILL_LOOKBACK_DAYS = 7 + + +class MassiveDataSource(MarketDataSource): + """MarketDataSource backed by the Massive REST API. + + Polls /v2/snapshot/locale/us/markets/stocks/tickers for the union of watched + tickers in one request, then writes each result to the PriceCache. + + Rate limits: + Basic (free): 5 req/min -> poll_interval 15.0 (default) + Paid tiers: unlimited -> poll_interval 2.0-5.0 + """ + + def __init__( + self, + api_key: str, + price_cache: PriceCache, + poll_interval: float = 15.0, + backfill_history: bool = True, + ) -> None: + self._api_key = api_key + self._cache = price_cache + self._interval = poll_interval + self._backfill_enabled = backfill_history + self._tickers: list[str] = [] + self._client: RESTClient | None = None + self._task: asyncio.Task | None = None + self._backfill_task: asyncio.Task | None = None + self._backoff = 1.0 + + @property + def source_name(self) -> str: + return "massive" + + # --- Lifecycle --- + + async def start(self, tickers: list[str]) -> None: + self._client = RESTClient(api_key=self._api_key) + self._tickers = [t.upper() for t in tickers] + + await self._log_market_status() + await self._poll_once() # Cache has prices before start() returns + + self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") + if self._backfill_enabled: + self._backfill_task = asyncio.create_task( + self._backfill_all(list(self._tickers)), name="massive-backfill" + ) + logger.info( + "Massive poller started: %d tickers, %.1fs interval", len(self._tickers), self._interval + ) + + async def stop(self) -> None: + for task in (self._task, self._backfill_task): + if task and not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self._task = self._backfill_task = None + self._client = None + logger.info("Massive poller stopped") + + async def add_ticker(self, ticker: str) -> None: + ticker = ticker.upper() + if ticker in self._tickers: + return + self._tickers.append(ticker) + logger.info("Massive: tracking %s (price arrives on the next poll)", ticker) + if self._backfill_enabled: + await self._backfill_one(ticker) + + async def remove_ticker(self, ticker: str) -> None: + ticker = ticker.upper() + self._tickers = [t for t in self._tickers if t != ticker] + self._cache.remove(ticker) + logger.info("Massive: dropped %s", ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + # --- Polling --- + + async def _poll_loop(self) -> None: + while True: + await asyncio.sleep(self._interval * self._backoff) + await self._poll_once() + + async def _poll_once(self) -> None: + """One poll cycle. Never raises — the loop must survive every failure.""" + if not self._tickers or self._client is None: + return + + try: + snapshots = await asyncio.to_thread(self._fetch_snapshots) + except Exception as exc: + # 401 bad key, 403 not in plan, 429 rate limited, network, timeout. + self._backoff = min(self._backoff * 2, MAX_BACKOFF_MULTIPLIER) + logger.error( + "Massive poll failed (%s); backing off to %.1fs", + exc, + self._interval * self._backoff, + ) + return + + self._backoff = 1.0 + updated = 0 + for snap in snapshots: + quote = self._extract_quote(snap) + if quote is None: + logger.warning("No usable price in snapshot for %s", getattr(snap, "ticker", "???")) + continue + price, timestamp = quote + self._cache.update(ticker=snap.ticker, price=price, timestamp=timestamp) + updated += 1 + + logger.debug("Massive poll: updated %d/%d tickers", updated, len(self._tickers)) + + def _fetch_snapshots(self) -> list: + """Synchronous SDK call. Runs on a worker thread.""" + assert self._client is not None + return self._client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=self._tickers, + ) + + @staticmethod + def _extract_quote(snap) -> tuple[float, float] | None: + """Best available price and its Unix-seconds timestamp, or None. + + TIMESTAMP UNITS ARE NOT UNIFORM IN THIS API: + last_trade.timestamp, last_quote.timestamp, snapshot.updated -> NANOseconds + Agg.timestamp (aggregate bars), min.timestamp -> MILLIseconds + + Massive's own sample lastTrade.t of 1605195918306274000 is 2020-11-12 + when divided by 1e9, and out of range for any other unit. Dividing by + 1e3 puts every price roughly 50,000 years in the future, silently. + """ + trade = getattr(snap, "last_trade", None) + if trade is not None and getattr(trade, "price", None): + return float(trade.price), float(trade.timestamp) / 1_000_000_000.0 + + minute = getattr(snap, "min", None) + if minute is not None and getattr(minute, "close", None): + return float(minute.close), float(minute.timestamp) / 1_000.0 + + day = getattr(snap, "day", None) + if day is not None and getattr(day, "close", None): + return float(day.close), time.time() + + return None + + # --- History backfill --- + + async def _backfill_all(self, tickers: list[str]) -> None: + """Seed sparkline history, one ticker at a time, spaced to respect the + rate limit. Off the poll loop: once at startup, once per added ticker. + + On the free tier this shares a 5 req/min budget with the poller, so + sparklines fill in over the first minutes rather than instantly. That is + the correct trade — a burst of ten requests at startup earns a 429 and + no history at all. + """ + for ticker in tickers: + await self._backfill_one(ticker) + await asyncio.sleep(self._interval) + + async def _backfill_one(self, ticker: str) -> None: + if self._client is None: + return + try: + history = await asyncio.to_thread(self._fetch_history, ticker) + except Exception as exc: + logger.warning("History backfill failed for %s: %s", ticker, exc) + return + if history: + self._cache.seed_history(ticker, history) + logger.debug("Backfilled %d history points for %s", len(history), ticker) + + def _fetch_history(self, ticker: str) -> list[float]: + """Most recent ~60 one-minute closes. Synchronous; runs on a thread.""" + assert self._client is not None + today = dt.date.today() + bars = self._client.get_aggs( + ticker=ticker, + multiplier=1, + timespan="minute", + from_=(today - dt.timedelta(days=BACKFILL_LOOKBACK_DAYS)).isoformat(), + to=today.isoformat(), + limit=HISTORY_POINTS, + sort="desc", + ) + return [float(bar.close) for bar in reversed(list(bars))] # Oldest first + + # --- Diagnostics --- + + async def _log_market_status(self) -> None: + """Log whether the market is open. 'Prices are not moving' is the most + likely support question on this path, and this line is the answer.""" + if self._client is None: + return + try: + status = await asyncio.to_thread(self._client.get_market_status) + logger.info("Massive market status: %s", getattr(status, "market", "unknown")) + except Exception as exc: + logger.warning("Could not read market status: %s", exc) +``` + +### Behaviour outside market hours is not a bug + +US equities trade 09:30–16:00 ET on weekdays. Outside that window `last_trade` holds the final +trade of the previous session and never changes. Consequently: + +- Every price is flat and no flash animation fires +- `change_from_open_percent` sits at zero +- The cache version never advances, so the SSE stream emits only heartbeats + +This is correct, and it is the main reason the simulator stays the recommended default. +`/api/health` reporting `market_source` and `newest_price_age_seconds` is what lets a user tell +this apart from a dead backend. + +Free-tier data is additionally 15 minutes delayed, and real-time snapshot access requires Starter +or above; on the free tier the aggregate endpoints are the dependable ones. + +### Optional: a real daily open on the Massive path + +The cache's session baseline is "the first price seen after start", which is uniform across both +sources and needs no market calendar. On real data a truer baseline is available — +`snap.day.open` — and can be adopted without disturbing the abstraction by adding one write-once +method to the cache: + +```python +class PriceCache: + def __init__(self, ...): + ... + self._pending_opens: dict[str, float] = {} # additional field + + def seed_open_price(self, ticker: str, open_price: float) -> None: + """Supply a real session open. Write-once, before the first update().""" + with self._lock: + if ticker in self._prices: + return # Baseline already pinned; do not move it under the UI + self._pending_opens[ticker] = round(open_price, 2) + + # and inside update(), for a ticker with no previous entry: + # open_price = self._pending_opens.pop(ticker, price) +``` + +The Massive poller would then call `cache.seed_open_price(snap.ticker, snap.day.open)` on its +first sight of each ticker. This is a refinement, not part of the core build — the change % it produces is more meaningful, and it is +the only place where the two sources would differ in behaviour rather than in timing. + +--- + +## 11. `factory.py` — the switch + +```python +"""Factory selecting the market data source from the environment.""" + +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: + """Return an UNSTARTED market data source. + + MASSIVE_API_KEY set and non-empty -> MassiveDataSource (real data) + otherwise -> SimulatorDataSource (GBM simulation) + + The caller owns the lifecycle, which keeps this function synchronous and + trivially testable. + """ + 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) + + logger.info("Market data source: GBM simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +`.strip()` carries real weight: `.env` files routinely contain `MASSIVE_API_KEY=` with nothing +after it, and an empty or whitespace-only value must mean "use the simulator", not "authenticate +with an empty key". The log line is not decoration either — it is the first thing to look at when +someone reports that prices are not moving. + +--- + +## 12. `stream.py` — SSE + +```python +"""SSE endpoint for live price updates.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import time +from collections.abc import AsyncGenerator + +from fastapi import APIRouter, Request +from fastapi.responses import StreamingResponse + +from .cache import PriceCache + +logger = logging.getLogger(__name__) + +POLL_INTERVAL = 0.5 # How often the generator looks at the cache +HEARTBEAT_INTERVAL = 15.0 # Comment frame cadence, price activity or not + + +def create_stream_router(price_cache: PriceCache) -> APIRouter: + """Build the /api/stream router bound to a specific cache. + + The router is created inside the factory, not at module level, so calling + this twice (an app plus a test app) does not register the route twice. + """ + router = APIRouter(prefix="/api/stream", tags=["streaming"]) + + @router.get("/prices") + async def stream_prices(request: Request) -> StreamingResponse: + """Live price stream for the browser's EventSource.""" + return StreamingResponse( + _generate_events(price_cache, request), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + # Stop nginx buffering the stream into uselessness if proxied + "X-Accel-Buffering": "no", + }, + ) + + return router + + +async def _generate_events( + price_cache: PriceCache, + request: Request, + interval: float = POLL_INTERVAL, + heartbeat: float = HEARTBEAT_INTERVAL, +) -> AsyncGenerator[str, None]: + """Yield SSE frames until the client disconnects. + + Emits one data event carrying EVERY tracked ticker, keyed by symbol, and + only when the cache version has moved. A heartbeat comment goes out every + `heartbeat` seconds regardless, so silence is legible to the frontend. + """ + client = request.client.host if request.client else "unknown" + logger.info("SSE client connected: %s", client) + + # EventSource reconnects on its own; tell it how long to wait. + yield "retry: 1000\n\n" + + last_version = -1 + last_beat = time.monotonic() + + try: + while True: + if await request.is_disconnected(): + logger.info("SSE client disconnected: %s", client) + break + + version = price_cache.version + if version != last_version: + last_version = version + prices = price_cache.get_all() + if prices: + payload = {ticker: update.to_dict() for ticker, update in prices.items()} + yield f"data: {json.dumps(payload)}\n\n" + + now = time.monotonic() + if now - last_beat >= heartbeat: + yield ": ping\n\n" + last_beat = now + + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("SSE stream cancelled: %s", client) + raise +``` + +### Wire format + +``` +retry: 1000 + +data: {"AAPL": {"ticker":"AAPL","price":190.50,"previous_price":190.40, + "open_price":189.20,"timestamp":1753401234.5, + "change":0.10,"change_percent":0.052, + "change_from_open_percent":0.687,"direction":"up"}, + "GOOGL": {...}} + +: ping + +``` + +**One event for all tickers, not one per ticker.** Ten tickers at 2Hz as separate events would be +20 messages per second per client, each needing its own parse and its own React state update. One +keyed object is one parse and one batched update — and it lets the frontend recompute the +portfolio total exactly once per frame (`PLAN.md` §10). + +**Emit only on change.** Overnight on real data the version never advances and the stream costs +nothing but heartbeats. + +**The heartbeat is what makes silence legible.** Without it the frontend cannot tell "connected, +market quiet" from "backend stalled" — both look like no data. With it, the connection dot can be +honest: green when a price event or heartbeat arrived within 30s, yellow when the stream is open +but silent for longer or `readyState === CONNECTING`, red when `CLOSED`. A comment frame (`:` and +no field name) is ignored by `EventSource` handlers but still resets the frontend's liveness +timer, which is exactly the semantic wanted. + +**Timestamps here are epoch seconds** — the SSE payload is the *only* place in FinAlly that uses +them. Every REST timestamp and every `*_at` column is an ISO 8601 UTC string. The two never mix +inside one payload. + +--- + +## 13. Wiring into FastAPI + +```python +# app/main.py +from __future__ import annotations + +import time +from contextlib import asynccontextmanager + +from fastapi import FastAPI, Request +from fastapi.staticfiles import StaticFiles + +from app.market import PriceCache, create_market_data_source, create_stream_router + +# Genuinely process-global state with a single lifetime. +price_cache = PriceCache() + + +@asynccontextmanager +async def lifespan(app: FastAPI): + init_database() # lazy init + seed + tickers = load_watchlist_tickers() # from SQLite + + source = create_market_data_source(price_cache) + await source.start(tickers) # cache is populated on return + + app.state.price_cache = price_cache + app.state.market_source = source + snapshot_task = start_snapshot_task(app) # portfolio snapshots, PLAN.md section 7 + try: + yield + finally: + snapshot_task.cancel() + await source.stop() + + +app = FastAPI(title="FinAlly", lifespan=lifespan) + +app.include_router(create_stream_router(price_cache)) +app.include_router(portfolio_router) +app.include_router(watchlist_router) +app.include_router(chat_router) +app.include_router(health_router) + +# MUST be last. Mounted before the routers it shadows every /api/* path and +# each endpoint 404s while the UI still appears to work — the most common way +# this architecture breaks (PLAN.md section 11). +app.mount("/", StaticFiles(directory="static", html=True), name="static") +``` + +The source is stashed on `app.state` so the watchlist router can call `add_ticker` / +`remove_ticker`. The cache is a module-level singleton rather than app state because +`create_stream_router` needs it at import time and there is exactly one of it per process; it is +also exposed on `app.state` so handlers can reach it through `Request` without importing +`main`. + +### `/api/health` + +```python +@router.get("/api/health") +async def health(request: Request) -> dict: + cache: PriceCache = request.app.state.price_cache + source: MarketDataSource = request.app.state.market_source + newest = cache.newest_timestamp() + return { + "status": "ok", + "market_source": source.source_name, # "simulator" | "massive" + "tickers_cached": len(cache), + "newest_price_age_seconds": ( + round(time.time() - newest, 3) if newest is not None else None + ), + } +``` + +Four fields chosen to answer "is the stream alive?" in one request. A simulator run shows an age +under a second; a Massive run overnight shows an age of many hours with `market_source: +"massive"`, which explains a flat UI without anyone having to read logs. + +--- + +## 14. Consumer recipes + +Everything below lives *outside* `app/market/`. It is included so the routers built next are +written against the real shapes. + +### `GET /api/watchlist` — price plus sparkline + +```python +@router.get("/api/watchlist") +async def get_watchlist(request: Request) -> dict: + cache: PriceCache = request.app.state.price_cache + tickers = [] + for ticker in list_watchlist_tickers(): # from SQLite, ordered + update = cache.get(ticker) + tickers.append( + { + "ticker": ticker, + "price": update.price if update else None, + "open_price": update.open_price if update else None, + "change_from_open_percent": ( + update.change_from_open_percent if update else 0.0 + ), + "history": cache.get_history(ticker), # ~60 points, oldest first + } + ) + return {"tickers": tickers} +``` + +### `POST /api/watchlist` — add + +```python +ticker = normalize_ticker(body.ticker) # ValueError -> 400 +insert_watchlist_row(ticker) # UNIQUE(user_id, ticker) +await request.app.state.market_source.add_ticker(ticker) +``` + +Order matters: persist first, then tell the source. If the insert fails on the unique +constraint, the source is never asked to track a duplicate. + +### `DELETE /api/watchlist/{ticker}` — remove + +```python +ticker = normalize_ticker(ticker) +if not in_watchlist(ticker): + raise HTTPException(404, f"{ticker} is not on the watchlist") +if has_open_position(ticker): + raise HTTPException(409, f"Cannot remove {ticker} while you hold a position in it") +delete_watchlist_row(ticker) +await request.app.state.market_source.remove_ticker(ticker) +``` + +The 409 plus the "trades auto-add their ticker" rule together hold the invariant that every +position has a live price feed. Without it a held position could lose its price and portfolio +valuation would silently drop a line. + +### `POST /api/portfolio/trade` — fill at the server's price + +```python +ticker = normalize_ticker(body.ticker) +if not in_watchlist(ticker): + insert_watchlist_row(ticker) + await request.app.state.market_source.add_ticker(ticker) + +try: + fill_price = await wait_for_price(cache, ticker) # up to 2s +except ValueError as exc: + raise HTTPException(400, str(exc)) from exc +``` + +The client's displayed price is advisory. The fill is whatever is in the cache when the request +lands, and the response returns it as `fill_price` so the UI shows the fill it got rather than the +price that was clicked. + +### Portfolio valuation + +```python +def total_value(cache: PriceCache, cash: float, positions: list[Position]) -> float: + return cash + sum(p.quantity * (cache.get_price(p.ticker) or p.avg_cost) for p in positions) +``` + +Falling back to `avg_cost` when a price is missing keeps the total finite during the sub-second +window after a restart. The frontend does the same arithmetic on every SSE frame from its own copy +of cash and positions — there is no portfolio SSE channel and no polling loop. + +--- + +## 15. Delta from the current implementation + +What is on disk today versus the design above. Nothing here is a rewrite; it is roughly 200 lines +of change across seven files, plus one new file. + +| # | File | Change | Why | +|---|---|---|---| +| 1 | `models.py` | Add `open_price` field, `change_from_open` / `change_from_open_percent` properties, both new keys in `to_dict()` | Session baseline — the whole frontend change % column depends on it | +| 2 | `cache.py:30` | `ts = time.time() if timestamp is None else timestamp` | `timestamp or time.time()` discards a legitimate `0.0` | +| 3 | `cache.py` | Carry `open_price` forward in `update()` | First write pins the baseline | +| 4 | `cache.py` | Add `_history` / `_history_at` deques, `seed_history()`, `get_history()`, `newest_timestamp()`; clear them in `remove()` | Sparklines on first paint; `/api/health` | +| 5 | `cache.py` | Bump `version` only when the rounded price changes | "Emit only on change" is currently untrue on the Massive path | +| 6 | `seed_prices.py:151` (used by `simulator.py`) | Replace `random.uniform(50.0, 300.0)` with `synthesize_params()` / `params_for()` | A restart currently reprices PYPL at random and makes the user's P&L nonsense | +| 7 | `seed_prices.py` | Drop `TSLA` from the tech group, drop `TSLA_CORR` and `DEFAULT_PARAMS` | The carve-out in `_pairwise_correlation` says the same thing twice; removing it makes positive definiteness provable (§8). Existing tests still pass — TSLA pairs stay at 0.3 | +| 8 | `simulator.py` | Add `backfill_history()`; seed history in `start()` and `add_ticker()` | 60 points at a one-minute cadence, ending at the live price | +| 9 | `simulator.py` | Guard `np.linalg.cholesky` with `LinAlgError` -> independent draws | A correlation constant should never take the feed down | +| 10 | `massive_client.py:103` | `/ 1_000_000_000.0`, not `/ 1000.0` | Nanoseconds. Every real-data price is currently stamped ~50,000 years in the future | +| 11 | `massive_client.py` | `_extract_quote()` fallback chain (last_trade -> min -> day) | `last_trade` can be absent off-hours; a missing attribute should not drop the ticker | +| 12 | `massive_client.py` | Startup + per-ticker history backfill via `get_aggs`, spaced by the poll interval | Sparklines on the real-data path without blowing the 5 req/min budget | +| 13 | `massive_client.py` | Exponential backoff on poll failure; market-status log line | 429s should widen the interval, not repeat every 15s | +| 14 | `interface.py` | Add abstract `source_name` property (`"simulator"` / `"massive"`) | `/api/health` needs it without an `isinstance` check | +| 15 | `stream.py:17` | Move `APIRouter(...)` inside `create_stream_router` | A module-level router double-registers `/prices` when the factory is called twice | +| 16 | `stream.py` | 15-second heartbeat comment frame | The connection dot cannot otherwise tell "quiet" from "stalled" | +| 17 | **new** `tickers.py` | `TICKER_PATTERN`, `normalize_ticker()`, exported from `__init__` | One validation rule for the manual and LLM paths | +| 18 | `__init__.py` | Export `wait_for_price`, `normalize_ticker`, `TICKER_PATTERN` | Trade and watchlist routers need them | +| 19 | `pyproject.toml` | Pin `massive==2.2.0` (currently `>=1.0.0`) | `MASSIVE_API.md` documents the 2.2.0 model shapes; a 1.x resolve would not match | +| 20 | `market_data_demo.py` | Show `change_from_open_percent`; read `cache.get_history()` instead of its own deque | Keeps the demo an honest preview of what the UI shows | + +Ordering note: items 1–5 are one commit (the model and cache change together and the tests move +with them); 6–9 and 10–13 are independent of each other; 14–18 are trivial and can ride along with +either. + +--- + +## 16. Testing + +The whole point of the interface is that none of this needs a network, an API key, a browser, or +real elapsed time. Any test that sleeps for more than a few hundred milliseconds is doing it +wrong. + +### 16.1 `PriceCache` + +```python +def test_first_update_pins_the_session_baseline(): + cache = PriceCache() + update = cache.update("AAPL", 190.00) + assert update.open_price == 190.00 + assert update.previous_price == 190.00 + assert update.direction == "flat" + assert update.change_from_open_percent == 0.0 + + +def test_open_price_survives_later_updates(): + cache = PriceCache() + cache.update("AAPL", 190.00) + cache.update("AAPL", 191.00) + update = cache.update("AAPL", 192.00) + assert update.open_price == 190.00 # still the first price seen + assert update.previous_price == 191.00 # but previous tracks the tick + assert update.change_from_open_percent == pytest.approx(1.0526, abs=1e-3) + + +def test_repeated_price_does_not_bump_version(): + cache = PriceCache() + cache.update("AAPL", 190.00) + version = cache.version + cache.update("AAPL", 190.00) + assert cache.version == version # SSE stays quiet + assert cache.get("AAPL").timestamp > 0 # but the feed still looks alive + + +def test_history_is_bounded_and_ordered(): + cache = PriceCache(history_points=5, history_interval=0.0) + for price in range(100, 110): + cache.update("AAPL", float(price)) + assert cache.get_history("AAPL") == [105.0, 106.0, 107.0, 108.0, 109.0] + + +def test_seed_history_then_remove_clears_everything(): + cache = PriceCache() + cache.seed_history("AAPL", [1.0, 2.0, 3.0]) + cache.update("AAPL", 4.0) + cache.remove("AAPL") + assert cache.get("AAPL") is None + assert cache.get_history("AAPL") == [] +``` + +### 16.2 `GBMSimulator` + +Seed both RNGs in a fixture; keep statistical tolerances generous, because a test that fails once +a week is worse than no test. + +| Target | Assertion | +|---|---| +| `step()` at default `dt` | All prices positive; every move under 1% (absent a shock) | +| Drift | Over 100k seeded steps, mean log-return ≈ `(mu − sigma²/2)·dt` | +| Volatility | Sample std of log-returns ≈ `sigma·√dt` | +| Correlation | Tech/tech log-return correlation exceeds tech/finance over many steps | +| Cholesky | Rebuilt on add and remove; matrix stays factorable at n = 1, 2, 50 | +| Shocks | With `event_probability=1.0`, every tick moves 2–5% | +| Unknown ticker | `synthesize_params("PYPL")` is identical across calls *and* processes | +| Backfill | Exactly 60 points, oldest first, last element == current price | + +```python +def test_synthesize_params_is_deterministic(): + # Not just stable within a process — stable across them. Hard-code the value. + price, params = synthesize_params("PYPL") + assert (price, params) == synthesize_params("PYPL") + assert 20.0 <= price <= 500.0 + assert 0.15 <= params["sigma"] <= 0.50 + + +def test_backfill_ends_at_the_current_price(): + sim = GBMSimulator(["AAPL"]) + history = sim.backfill_history("AAPL") + assert len(history) == 60 + assert history[-1] == sim.get_price("AAPL") + + +def test_step_moves_are_realistic(): + sim = GBMSimulator(["AAPL", "TSLA"], event_probability=0.0) + before = {t: sim.get_price(t) for t in sim.get_tickers()} + after = sim.step() + for ticker, price in after.items(): + assert price > 0 + assert abs(price - before[ticker]) / before[ticker] < 0.01 +``` + +### 16.3 Conformance — one suite, both implementations + +Anything only one source passes is a leak in the abstraction, so run the lifecycle contract +against both. + +```python +@pytest.fixture(params=["simulator", "massive"]) +def source_and_cache(request): + cache = PriceCache() + if request.param == "simulator": + yield SimulatorDataSource(cache, update_interval=0.05), cache + else: + source = MassiveDataSource("test-key", cache, poll_interval=60.0, + backfill_history=False) + with patch.object(source, "_fetch_snapshots", return_value=[ + _snapshot("AAPL", 190.50), _snapshot("GOOGL", 175.25), + ]), patch.object(source, "_log_market_status", new=AsyncMock()), \ + patch("app.market.massive_client.RESTClient"): + yield source, cache + + +async def test_start_populates_cache_before_returning(source_and_cache): + source, cache = source_and_cache + await source.start(["AAPL", "GOOGL"]) + assert cache.get_price("AAPL") is not None # no sleep, no polling + await source.stop() + + +async def test_stop_is_idempotent(source_and_cache): + source, cache = source_and_cache + await source.start(["AAPL"]) + await source.stop() + await source.stop() # must not raise + + +async def test_remove_ticker_clears_the_cache(source_and_cache): + source, cache = source_and_cache + await source.start(["AAPL", "GOOGL"]) + await source.remove_ticker("AAPL") + assert "AAPL" not in cache + assert "AAPL" not in source.get_tickers() + await source.stop() +``` + +### 16.4 `MassiveDataSource` + +Mock the SDK entirely. The two assertions that matter are the nanosecond conversion and that an +exception inside a poll never escapes the loop. + +```python +def _snapshot(ticker: str, price: float, ts_ns: int = 1605195918306274000): + snap = MagicMock() + snap.ticker = ticker + snap.last_trade.price = price + snap.last_trade.timestamp = ts_ns + return snap + + +async def test_nanosecond_timestamps_convert_to_seconds(): + cache = PriceCache() + source = MassiveDataSource("k", cache, poll_interval=60.0) + source._tickers, source._client = ["AAPL"], MagicMock() + + with patch.object(source, "_fetch_snapshots", return_value=[_snapshot("AAPL", 190.5)]): + await source._poll_once() + + # 1605195918306274000 ns == 2020-11-12T15:45:18Z, not the year 52,000 + assert cache.get("AAPL").timestamp == pytest.approx(1605195918.306274) + + +async def test_poll_failure_is_swallowed_and_backs_off(): + cache = PriceCache() + source = MassiveDataSource("k", cache, poll_interval=15.0) + source._tickers, source._client = ["AAPL"], MagicMock() + + with patch.object(source, "_fetch_snapshots", side_effect=RuntimeError("429")): + await source._poll_once() # must not raise + + assert source._backoff == 2.0 + assert len(cache) == 0 +``` + +### 16.5 SSE + +Drive `_generate_events` directly against a hand-fed cache and a stub request — no ASGI server, +no browser. + +```python +class _StubRequest: + client = None + def __init__(self, disconnect_after: int): + self._calls, self._limit = 0, disconnect_after + async def is_disconnected(self) -> bool: + self._calls += 1 + return self._calls > self._limit + + +async def test_no_event_when_the_version_is_unchanged(): + cache = PriceCache() + cache.update("AAPL", 190.00) + frames = [f async for f in _generate_events(cache, _StubRequest(3), interval=0.0)] + data_frames = [f for f in frames if f.startswith("data:")] + assert len(data_frames) == 1 # one payload, then silence + + +async def test_heartbeat_arrives_in_a_quiet_market(): + cache = PriceCache() + frames = [f async for f in _generate_events( + cache, _StubRequest(3), interval=0.0, heartbeat=0.0)] + assert ": ping\n\n" in frames + + +async def test_payload_is_keyed_by_ticker_and_carries_the_baseline(): + cache = PriceCache() + cache.update("AAPL", 189.20) + cache.update("AAPL", 190.50) + frames = [f async for f in _generate_events(cache, _StubRequest(1), interval=0.0)] + payload = json.loads(frames[1].removeprefix("data: ")) + assert payload["AAPL"]["open_price"] == 189.20 + assert payload["AAPL"]["change_from_open_percent"] == pytest.approx(0.687, abs=1e-3) +``` + +### 16.6 Factory + +```python +@pytest.mark.parametrize("value,expected", [ + (None, SimulatorDataSource), + ("", SimulatorDataSource), + (" ", SimulatorDataSource), # a bare `MASSIVE_API_KEY=` in .env + ("real-key", MassiveDataSource), +]) +def test_factory_selection(monkeypatch, value, expected): + if value is None: + monkeypatch.delenv("MASSIVE_API_KEY", raising=False) + else: + monkeypatch.setenv("MASSIVE_API_KEY", value) + assert isinstance(create_market_data_source(PriceCache()), expected) +``` + +--- + +## 17. Failure modes + +| Situation | Behaviour | Rationale | +|---|---|---| +| Empty watchlist at startup | `start([])` succeeds; `step()` returns `{}`; SSE sends heartbeats only | A user can delete every ticker; that is not an error | +| Trade on a just-added ticker | `wait_for_price` polls for up to 2s at 200ms, then 400 with a retryable message | Never happens on the simulator; can happen on a 15s Massive poll | +| Invalid `MASSIVE_API_KEY` | 401 logged each poll, backoff widens to 2 minutes, app serves stale/empty prices | A bad key must not prevent the app from starting | +| 429 rate limit | Backoff doubles to a ceiling of 8× the interval, resets on the first success | Hammering a rate limit makes it worse | +| Market closed (Massive) | Prices flat, version frozen, stream emits heartbeats only, `/api/health` shows a large `newest_price_age_seconds` | Correct behaviour; the health payload is how a user tells it apart from a stall | +| Simulator task raises | Exception logged, loop continues on the next tick | A dead task leaves a UI that looks connected and frozen — the worst outcome | +| Cholesky not factorable | Logged, falls back to independent draws | Losing correlation is a cosmetic regression; losing the feed is not | +| Ticker removed mid-stream | `cache.remove()` drops price and history; the next SSE payload omits it | The frontend keys on the payload, so the row disappears cleanly | +| Two `start()` calls | Undefined; the second overwrites the simulator and orphans the first task | Guarded by the single call site in `lifespan`, not by defensive code | +| Price at exactly `0.0` | `change_percent` and `change_from_open_percent` return `0.0` rather than dividing | Cannot arise from GBM (always positive) but a real feed can send a zero | + +--- + +## Appendix — configuration + +| Variable | Default | Effect on this module | +|---|---|---| +| `MASSIVE_API_KEY` | unset | Non-empty after `.strip()` selects `MassiveDataSource`; otherwise the simulator | + +| Constant | Value | Where | Meaning | +|---|---|---|---| +| `HISTORY_POINTS` | 60 | `cache.py` | Sparkline points retained per ticker | +| `HISTORY_INTERVAL_SECONDS` | 60.0 | `cache.py` | Minimum spacing between recorded history points | +| `TRADING_SECONDS_PER_YEAR` | 5,896,800 | `simulator.py` | 252 days × 6.5h × 3600 | +| `DEFAULT_DT` | ~8.48e-8 | `simulator.py` | 500ms as a fraction of a trading year | +| `HISTORY_STEP_TICKS` | 120 | `simulator.py` | One backfill point per simulated minute | +| `update_interval` | 0.5 | `SimulatorDataSource` | Simulator tick | +| `event_probability` | 0.001 | `SimulatorDataSource` | Shock chance per tick per ticker | +| `poll_interval` | 15.0 | `MassiveDataSource` | REST poll cadence (free tier safe) | +| `MAX_BACKOFF_MULTIPLIER` | 8.0 | `massive_client.py` | Backoff ceiling — 2 minutes at the default interval | +| `POLL_INTERVAL` | 0.5 | `stream.py` | How often the SSE generator reads the cache | +| `HEARTBEAT_INTERVAL` | 15.0 | `stream.py` | Comment frame cadence | diff --git a/planning/MARKET_INTERFACE.md b/planning/MARKET_INTERFACE.md new file mode 100644 index 000000000..a36e2ef5d --- /dev/null +++ b/planning/MARKET_INTERFACE.md @@ -0,0 +1,359 @@ +# Unified Market Data Interface + +The design of `backend/app/market/` — the one Python API the rest of FinAlly uses to get +stock prices, whether they come from the Massive API or the built-in simulator. + +Companion documents: `MASSIVE_API.md` (the real data provider), `MARKET_SIMULATOR.md` (the +default provider). + +## The Problem + +FinAlly needs live prices for portfolio valuation, trade fills, watchlist display, and SSE +streaming. Prices come from one of two very different places: + +- The **simulator**, which produces a new price for every ticker every 500ms, synchronously, + in-process, for free, forever +- The **Massive API**, which is a synchronous HTTP call, rate-limited to 5 requests/minute on + the free tier, 15 minutes delayed, and completely static overnight + +If those differences leak upward, every consumer needs two code paths. The design goal is +that **no code outside `app/market/` knows or cares which source is running.** + +## Shape of the Solution + +``` + MASSIVE_API_KEY? + | + +--------------+--------------+ + | set | unset + v v + MassiveDataSource SimulatorDataSource + (REST poll, 15s) (GBM step, 500ms) + | | + +--------------+--------------+ + | + writes into + v + PriceCache <-- single source of truth + | + +--------------+--------------+-----------------+ + v v v v + SSE stream portfolio trade fill watchlist + /api/stream valuation price lookup response +``` + +The key inversion: **consumers never call the data source.** They read the cache. The source +is a producer that runs on its own schedule and pushes in. This is what makes a 500ms +simulator and a 15-second poller interchangeable — the cache always has a current price, and +the reader does not know how old it is or where it came from. + +## `PriceUpdate` — the unit of data + +An immutable frozen dataclass in `models.py`. One ticker, one moment. + +```python +@dataclass(frozen=True, slots=True) +class PriceUpdate: + ticker: str + price: float + previous_price: float + open_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: ... + @property + def change_percent(self) -> float: ... # vs previous tick + @property + def change_from_open_percent(self) -> float: ... # vs session open + @property + def direction(self) -> str: ... # "up" | "down" | "flat" + + def to_dict(self) -> dict: ... +``` + +### Two different "change" numbers, and why + +`change_percent` compares against the previous tick. At 500ms intervals that number flickers +around zero and is meaningless as a "daily change" column — it says how the last half-second +went. + +`open_price` is the **session baseline**: the first price seen for that ticker after process +start, or after the ticker was added to a running system. `change_from_open_percent` measures +against it, which is the number a user recognises as "how is it doing today". + +The UI rule from `PLAN.md`: the watchlist change column and the price-flash colouring both use +`change_from_open_percent`. `change_percent` and `direction` drive only the momentary flash +animation. + +The baseline survives page reloads and SSE reconnections because it lives in the server's +cache, and resets on container restart. For a simulation that is the right trade — it needs no +persistence and no market-calendar logic. + +Frozen and slotted because a `PriceUpdate` is handed to SSE serialisation and to portfolio +maths concurrently; nothing should be able to mutate one after the cache published it. + +## `PriceCache` — the single point of truth + +Thread-safe in-memory store in `cache.py`. One writer (the active source), many readers. + +```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 get_history(self, ticker: str) -> list[float] + def seed_history(self, ticker: str, prices: list[float]) -> None + def remove(self, ticker: str) -> None + + @property + def version(self) -> int +``` + +Design points: + +**`update()` computes the derived fields.** Callers pass a raw price; the cache looks up the +previous entry, carries `open_price` forward, and constructs the `PriceUpdate`. Sources stay +dumb — they fetch or generate a number and hand it over. All the semantics live in one place. + +**First update establishes the baseline.** When a ticker has no prior entry, +`previous_price == price == open_price`, so direction is `flat` and both change percentages +are zero. No special-casing at the call sites. + +**A `threading.Lock`, not an asyncio lock.** The simulator writes from the event loop, but the +Massive poller writes from a worker thread via `asyncio.to_thread`. A `threading.Lock` is +correct for both; an `asyncio.Lock` would be silently unsafe for the threaded path. + +**A monotonic `version` counter.** Incremented on every write. The SSE generator compares it +against the version it last sent and emits nothing when unchanged. This is what makes a quiet +market cost nothing: overnight on real data the version never advances, so the stream sends +only heartbeats instead of pushing identical payloads twice a second. + +**Bounded per-ticker history.** A `deque(maxlen=60)` of recent prices per ticker, so +`/api/watchlist` can return populated sparklines on first paint rather than making the +frontend accumulate 30 seconds of data before drawing anything. `seed_history()` lets a source +backfill it at startup. Bounded so it cannot grow without limit over a long-running container. + +## `MarketDataSource` — the provider contract + +Abstract base class in `interface.py`. Five methods, all the surface there is: + +```python +class MarketDataSource(ABC): + async def start(self, tickers: list[str]) -> None + async def stop(self) -> None + async def add_ticker(self, ticker: str) -> None + async def remove_ticker(self, ticker: str) -> None + def get_tickers(self) -> list[str] +``` + +Notice what is **not** here: there is no `get_price()`. A source cannot be asked for a price. +It only pushes into the cache. Omitting the read method is what enforces the architecture — +a consumer that tries to bypass the cache finds there is no way to. + +Lifecycle contract: + +- `start()` is called exactly once, at app startup, and must populate the cache before + returning so the first HTTP request already has prices +- `stop()` is idempotent and must not write to the cache afterwards +- `add_ticker()` / `remove_ticker()` are no-ops when the ticker is already present or absent +- `remove_ticker()` also clears the ticker from the cache +- Both are `async` because the Massive implementation may need to await a backfill request; + the simulator's are trivially async + +`get_tickers()` is deliberately synchronous — it reads local state and is called from request +handlers. + +## The two implementations + +### `SimulatorDataSource` + +Wraps a `GBMSimulator` in an asyncio task that steps every 500ms and writes each price to the +cache. No I/O, no failure modes worth handling beyond logging. Full detail in +`MARKET_SIMULATOR.md`. + +### `MassiveDataSource` + +Polls the full-market-snapshot endpoint for the union of watched tickers in one request every +15 seconds, then writes each result to the cache. + +Three things it must get right: + +**Never block the event loop.** The `massive` SDK is synchronous urllib3. Every call goes +through `asyncio.to_thread`. A blocking HTTP call in the loop would freeze every SSE +connection and every API request for the duration. + +**Convert nanoseconds, not milliseconds.** `snap.last_trade.timestamp` is in nanoseconds. + +```python +timestamp = snap.last_trade.timestamp / 1_000_000_000.0 +``` + +The current code divides by `1000.0`, putting every real-data price roughly 50,000 years in +the future. `MASSIVE_API.md` has the proof. This is the first fix required on the Massive +path. + +**Never propagate a failure.** The poll loop catches broadly, logs, and continues. A 429, a +401, or a dropped connection leaves the last good prices in the cache and retries in 15 +seconds. The app stays up with stale data rather than failing requests. + +Poll interval is a constructor argument defaulting to 15.0 seconds — the free tier's 5 +requests/minute allows one per 12 seconds, and 15 leaves headroom. Paid tiers can pass 2-5. + +**Startup backfill.** On `start()`, one `get_aggs(..., timespan="minute", limit=60)` call per +ticker seeds `cache.seed_history()` so sparklines are populated. This is off the polling loop +— once at startup, and once per newly added ticker. + +## `create_market_data_source()` — the switch + +The whole environment-variable decision, in one function in `factory.py`: + +```python +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + 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) + logger.info("Market data source: GBM Simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +`.strip()` matters: `.env` files routinely contain `MASSIVE_API_KEY=` with nothing after it, +and an empty string must mean "use the simulator", not "authenticate with an empty key". + +Returns an **unstarted** source. The caller owns the lifecycle, which keeps the factory +synchronous and trivially testable. + +The log line is not decoration. "Why are prices not moving" is the most likely support +question in this project, and the answer is usually visible in the first line of the log. + +## Ticker Validation — one shared rule + +Manual adds, LLM-driven adds, and trades all funnel through the same check: + +```python +TICKER_PATTERN = re.compile(r"^[A-Z]{1,5}$") + +def normalize_ticker(raw: str) -> str: + """Uppercase and validate. Raises ValueError if malformed.""" + ticker = raw.strip().upper() + if not TICKER_PATTERN.match(ticker): + raise ValueError(f"Invalid ticker: {raw!r}") + return ticker +``` + +Uppercase first, then match. One function, one regex, one error message, so the manual path +and the LLM path cannot drift apart. Callers translate the `ValueError` into a 400. + +Note this validates *shape*, not *existence*. The simulator deliberately accepts any +well-formed symbol — see `MARKET_SIMULATOR.md`. + +## Wiring into FastAPI + +```python +from contextlib import asynccontextmanager +from fastapi import FastAPI +from app.market import PriceCache, create_market_data_source, create_stream_router + +price_cache = PriceCache() + +@asynccontextmanager +async def lifespan(app: FastAPI): + source = create_market_data_source(price_cache) + await source.start(load_watchlist_tickers()) + app.state.market_source = source + yield + await source.stop() + +app = FastAPI(lifespan=lifespan) +app.include_router(create_stream_router(price_cache)) +# ... all other /api routers ... +# StaticFiles mount goes LAST - see PLAN.md section 11 +``` + +The source is stashed on `app.state` so the watchlist router can call `add_ticker` / +`remove_ticker`. The cache is a module-level singleton because it is genuinely process-global +state with a single lifetime. + +## SSE Streaming + +`create_stream_router(price_cache)` returns a router exposing `GET /api/stream/prices`. + +The generator: + +1. Opens with `retry: 1000`, so the browser's `EventSource` reconnects after a second +2. Every 500ms, compares `cache.version` against the last sent version +3. On change, emits **one event carrying every tracked ticker**, keyed by symbol +4. Every 15 seconds regardless of price activity, emits a heartbeat comment `: ping\n\n` +5. Exits when `request.is_disconnected()` + +``` +data: {"AAPL": {"ticker":"AAPL","price":190.50,"previous_price":190.40, + "open_price":189.20,"timestamp":1753401234.5, + "change":0.10,"change_percent":0.052, + "change_from_open_percent":0.687,"direction":"up"}, ...} +``` + +**One event for all tickers, not one per ticker.** Ten tickers at 2Hz would be 20 events per +second per client, each needing its own parse and its own React state update. One keyed +object is one parse and one batched update. + +**Emit only on change** keeps an idle market silent, which is the overnight-real-data case. + +**The heartbeat is what makes silence legible.** Without it the frontend cannot distinguish +"connected, market quiet" from "backend stalled" — both look like no data. With it, the +connection dot can be honest: green when a price event or heartbeat arrived in the last 30 +seconds, yellow when open but silent longer than that or reconnecting, red when closed. + +Response headers set `Cache-Control: no-cache` and `X-Accel-Buffering: no`, the latter to stop +nginx buffering the stream into uselessness if the app is ever proxied. + +## Consuming the Cache + +```python +from app.market import PriceCache + +update = cache.get("AAPL") # PriceUpdate | None +price = cache.get_price("AAPL") # float | None +prices = cache.get_all() # dict[str, PriceUpdate] +history = cache.get_history("AAPL") # list[float], up to 60, oldest first +``` + +Trade execution needs one extra behaviour. A just-added ticker may have no price for a few +hundred milliseconds, and failing the trade would be a poor experience: + +```python +async def wait_for_price(cache: PriceCache, ticker: str, timeout: float = 2.0) -> float: + """Poll the cache for a first tick. Raises ValueError on timeout.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + price = cache.get_price(ticker) + if price is not None: + return price + await asyncio.sleep(0.2) + raise ValueError(f"No price available for {ticker} yet, try again") +``` + +The simulator publishes within 500ms so this effectively never expires. On the Massive path +with a 15-second poll it genuinely can, which is why it returns a retryable message rather +than a hard error. + +## Testing Strategy + +The interface is what makes the subsystem testable without network or waiting. + +- **`PriceCache`** — pure synchronous logic. Assert `open_price` is pinned by the first + update and carried through later ones; assert `version` increments; assert history is + bounded at 60. +- **Conformance** — one parametrised suite runs the lifecycle contract against both + implementations. Anything only one of them passes is a leak in the abstraction. +- **`create_market_data_source`** — `monkeypatch.setenv` over set, unset, empty, and + whitespace-only keys; assert the returned type. +- **`MassiveDataSource`** — mock the SDK entirely. The valuable assertions are the + nanosecond conversion and that a raised exception inside a poll does not escape the loop. +- **`SimulatorDataSource`** — inject a large `dt` to make moves observable without waiting. +- **SSE** — drive `_generate_events` directly with a hand-fed cache; assert no event is + emitted when the version is unchanged, and that a heartbeat still arrives. + +No test should need a real API key, and no test should need to sleep for real time. diff --git a/planning/MARKET_SIMULATOR.md b/planning/MARKET_SIMULATOR.md new file mode 100644 index 000000000..d6ec8ad3a --- /dev/null +++ b/planning/MARKET_SIMULATOR.md @@ -0,0 +1,316 @@ +# Market Simulator Design + +The default market data source for FinAlly: a geometric Brownian motion price simulator that +runs in-process, needs no API key, no network, and no market hours. + +Companion documents: `MARKET_INTERFACE.md` (the contract it implements), `MASSIVE_API.md` (the +real-data alternative). + +## Why the Simulator Is the Default + +It is not a fallback for people without an API key. It is the better demo. + +| | Simulator | Massive (free tier) | +|---|---|---| +| Prices move | Always, every 500ms | Only 09:30-16:00 ET weekdays | +| Latency | None | 15 minutes delayed | +| Cost | Free | Free tier, 5 req/min | +| Setup | None | Signup, key, `.env` | +| Reproducible | Yes, with a seed | No | + +A student running the app at 21:00 on a Sunday against real data sees ten flat prices, no +flash animations, and zero change percentages, and reasonably concludes it is broken. The +simulator always looks alive. Real data is the interesting option, not the default one. + +## The Model: Geometric Brownian Motion + +GBM is the standard model for equity prices — the one underneath Black-Scholes. It produces +prices that are always positive, whose returns are normally distributed, and whose volatility +scales with price level. That is enough realism for a trading UI. + +``` +S(t+dt) = S(t) * exp( (mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z ) +``` + +| Term | Meaning | +|---|---| +| `S(t)` | Current price | +| `mu` | Annualised drift — expected return | +| `sigma` | Annualised volatility | +| `dt` | Time step as a fraction of a trading year | +| `Z` | Standard normal draw, correlated across tickers | + +The `- sigma^2/2` correction is not cosmetic. Without it, `mu` is the drift of log-price and +the *expected price* grows faster than `mu` — prices would inflate visibly over a long +session. With it, `mu` means what it claims. + +### Choosing `dt` + +`mu` and `sigma` are quoted annualised, so `dt` must be a fraction of a trading year measured +the same way. A trading year is 252 days of 6.5 hours: + +```python +TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 +DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8 for a 500ms tick +``` + +That tiny `dt` is what makes the output look right. A stock at $190 with `sigma=0.22` moves on +the order of a cent per tick, wanders a few tenths of a percent over a minute, and can drift a +percent or two over a long session. Sub-cent-per-tick is exactly what a real level-1 feed +looks like. + +Getting `dt` wrong is the classic failure here. Using `dt = 0.5` (half a *year* per tick) +sends prices to five figures within a minute. + +## Correlation Between Tickers + +Independent random walks look wrong. Real markets move together — when tech sells off, it +sells off broadly. Ten independently wandering lines read as noise; correlated ones read as a +market. + +Correlation is imposed by Cholesky decomposition. Given a correlation matrix `C`, factor it as +`C = L * L^T`; then for a vector of independent standard normals `z`, the product `L @ z` has +exactly the correlation structure of `C`. + +```python +corr = np.eye(n) +for i in range(n): + for j in range(i + 1, n): + rho = pairwise_correlation(tickers[i], tickers[j]) + corr[i, j] = corr[j, i] = rho + +cholesky = np.linalg.cholesky(corr) # rebuilt on add/remove, O(n^3) but n < 50 +z = cholesky @ np.random.standard_normal(n) +``` + +The correlation structure is sector-based: + +| Pair | rho | +|---|---| +| Tech / tech | 0.60 | +| Finance / finance | 0.50 | +| Anything involving TSLA | 0.30 | +| Cross-sector, or unknown ticker | 0.30 | + +Groups live in `seed_prices.py`: tech is AAPL, GOOGL, MSFT, AMZN, META, NVDA, NFLX; finance is +JPM and V. TSLA is carved out deliberately — it is nominally tech but famously does its own +thing, and giving it independence makes the watchlist more interesting to watch. + +One constraint to respect: a correlation matrix must be **positive definite** or +`np.linalg.cholesky` raises. Correlations assigned pairwise by ad-hoc rules are not guaranteed +to be. The values above are mild and consistently assigned, so they factor cleanly; anything +more elaborate (high correlations, more groups, exceptions layered on exceptions) needs either +a check or a nearest-positive-definite repair. If `cholesky` ever raises, that is the cause. + +## Random Shock Events + +GBM alone is smooth. Real markets jump — earnings, news, a halt. Each ticker gets a ~0.1% +chance per tick of a 2-5% move in a random direction: + +```python +if random.random() < self._event_prob: + magnitude = random.uniform(0.02, 0.05) + sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + magnitude * sign +``` + +With ten tickers at 2 ticks/second, that is an event roughly every 50 seconds — often enough +that a user watching for a minute sees one, rare enough that it stays an event. This is the +single highest-value line of code in the simulator for demo purposes: it is what makes the +flash animations and the P&L chart do something worth looking at. + +## Unknown Tickers Are Accepted + +`SEED_PRICES` has real parameters for ten symbols. The AI assistant can add any symbol to the +watchlist, and users will type PYPL, AMD, and DIS. Rejecting them would dead-end the headline +demo — "ask the AI to watch a new stock" — on the first plausible thing anyone tries. + +So unknown tickers are accepted, with parameters **synthesized deterministically from the +symbol**: + +```python +def synthesize_params(ticker: str) -> tuple[float, dict[str, float]]: + """Derive a stable seed price and GBM params from the ticker symbol. + + Deterministic so PYPL is the same price on every run - a restart should + not silently reprice the user's position. + """ + digest = hashlib.sha256(ticker.encode()).digest() + price = 20.0 + (int.from_bytes(digest[0:4], "big") % 48_000) / 100.0 # $20-$500 + sigma = 0.15 + (digest[4] / 255.0) * 0.35 # 0.15-0.50 + mu = 0.02 + (digest[5] / 255.0) * 0.06 # 0.02-0.08 + return round(price, 2), {"sigma": sigma, "mu": mu} +``` + +Two properties matter: + +**Deterministic.** SHA-256 of the symbol, not `random.uniform()`. A user holding 10 shares of +PYPL bought at $73 must not restart the container and find PYPL now trades at $412 — their +position's P&L would be nonsense. The current code uses `random.uniform(50.0, 300.0)`, which +has exactly that bug and needs replacing. + +**Plausible ranges.** $20-$500, `sigma` 0.15-0.50, `mu` 0.02-0.08. Every synthesized ticker +looks like an ordinary large-cap and behaves like one. + +Synthesized tickers join the cross-sector correlation group at rho 0.30 — no attempt is made +to guess sectors from a symbol. + +The shape check `^[A-Z]{1,5}$` still applies (see `MARKET_INTERFACE.md`). "ZZZZZ" is accepted +and gets a price; "hello world" and "12345" are rejected with a 400. The simulator validates +form, not existence — and cannot do otherwise, since it has no universe of real symbols. + +## History Backfill + +Sparklines need ~60 points. Without backfill, a fresh page shows sixty seconds of empty +charts, which is the first thing anyone sees on launch. + +So at startup — and again for each newly added ticker — the simulator runs the GBM recurrence +*backwards* from the seed price to manufacture plausible prior history: + +```python +def backfill_history(self, ticker: str, points: int = 60) -> list[float]: + """Generate synthetic prior history ending at the current price.""" + params = self._params[ticker] + sigma, mu = params["sigma"], params["mu"] + dt = self._dt * 120 # one point per minute, not per tick + + price = self._prices[ticker] + history = [price] + for _ in range(points - 1): + drift = (mu - 0.5 * sigma**2) * dt + diffusion = sigma * math.sqrt(dt) * np.random.standard_normal() + price /= math.exp(drift + diffusion) # step backwards + history.append(round(price, 2)) + return list(reversed(history)) # oldest first +``` + +Dividing rather than multiplying walks the process backwards, so the series **ends** at the +current price and joins continuously with the live stream. The coarser `dt` gives history a +per-minute cadence, so the sparkline shows an hour of price action rather than thirty seconds +of it — visible shape instead of a flat line. + +The result goes into `cache.seed_history()`, and `/api/watchlist` serves it. The frontend +extends it from SSE thereafter. + +## Code Structure + +``` +backend/app/market/ +├── models.py PriceUpdate +├── interface.py MarketDataSource ABC +├── cache.py PriceCache +├── seed_prices.py Seed prices, GBM params, correlation groups, synthesize_params() +├── simulator.py GBMSimulator + SimulatorDataSource +├── massive_client.py MassiveDataSource +├── factory.py create_market_data_source() +└── stream.py SSE router +``` + +The split inside `simulator.py` is the important one. + +### `GBMSimulator` — the maths, no I/O + +Pure, synchronous, testable. Holds prices, parameters, and the Cholesky factor. Knows nothing +about the cache, asyncio, or FastAPI. + +```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 one tick + def add_ticker(self, ticker: str) -> None + def remove_ticker(self, ticker: str) -> None + def get_price(self, ticker: str) -> float | None + def get_tickers(self) -> list[str] + def backfill_history(self, ticker: str, points: int = 60) -> list[float] +``` + +`step()` is the hot path — every 500ms, forever. It draws all `n` normals in one numpy call +and applies one matrix multiply, rather than looping per ticker. At `n < 50` this is +irrelevant to performance, but it is also simply the clearer way to express "correlated draws". + +Because it is pure, its tests need no clock and no event loop: construct with a large `dt`, +call `step()`, assert prices moved and stayed positive. Statistical properties (drift and +variance over many steps) are testable by seeding numpy. + +### `SimulatorDataSource` — the plumbing, no maths + +Implements `MarketDataSource`. Owns the asyncio task, writes into the cache, and does nothing +else. + +```python +class SimulatorDataSource(MarketDataSource): + async def start(self, tickers: list[str]) -> None: + self._sim = GBMSimulator(tickers, event_probability=self._event_prob) + for ticker in tickers: + self._cache.seed_history(ticker, self._sim.backfill_history(ticker)) + self._cache.update(ticker, self._sim.get_price(ticker)) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + + async def _run_loop(self) -> None: + while True: + try: + for ticker, price in self._sim.step().items(): + self._cache.update(ticker, price) + except Exception: + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +Two details worth keeping: + +**The cache is seeded before the task starts.** `start()` returns with prices already +available, so the first HTTP request never sees an empty cache. Same for `add_ticker()` — a +newly added ticker gets a price immediately, which is what keeps the 2-second trade wait from +ever mattering on the simulator path. + +**The loop catches and continues.** A background task that raises dies silently and takes the +entire price feed with it, leaving a UI that looks connected and frozen. Logging and +continuing is the right call for a loop that will get another chance in 500ms. + +## Parameters + +`seed_prices.py`, chosen to be recognisable rather than current: + +| Ticker | Seed | sigma | mu | | +|---|---|---|---|---| +| AAPL | 190.00 | 0.22 | 0.05 | | +| GOOGL | 175.00 | 0.25 | 0.05 | | +| MSFT | 420.00 | 0.20 | 0.05 | | +| AMZN | 185.00 | 0.28 | 0.05 | | +| TSLA | 250.00 | 0.50 | 0.03 | high volatility | +| NVDA | 800.00 | 0.40 | 0.08 | high volatility, strong drift | +| META | 500.00 | 0.30 | 0.05 | | +| JPM | 195.00 | 0.18 | 0.04 | low volatility | +| V | 280.00 | 0.17 | 0.04 | low volatility | +| NFLX | 600.00 | 0.35 | 0.05 | | + +The spread in `sigma` is the point. TSLA at 0.50 against V at 0.17 means TSLA visibly jumps +while V barely moves — the watchlist has texture instead of ten lines doing the same thing. +NVDA carries the strongest drift so something in the portfolio tends to trend upward, which +makes the P&L chart more interesting than a random walk around zero. + +Seed prices bear no relation to the actual market on any given day, and should not. This is a +simulation with $10,000 of pretend money. + +## Testing + +| Target | Assertion | +|---|---| +| GBM step | Prices stay positive; moves are sub-percent at the default `dt` | +| Drift | Over many seeded steps, mean log-return approximates `mu * dt` | +| Volatility | Sample std of log-returns approximates `sigma * sqrt(dt)` | +| Correlation | Tech pairs co-move more than tech/finance pairs over many steps | +| Cholesky | Rebuilds on add and remove; matrix stays factorable | +| Unknown tickers | `synthesize_params("PYPL")` returns identical values across calls and processes | +| Backfill | Returns exactly 60 points, oldest first, ending at the current price | +| Events | With `event_probability=1.0`, every tick moves 2-5% | +| Data source | `start()` populates cache and history; `stop()` is idempotent; a raising step does not kill the loop | + +Determinism comes from `np.random.seed()` and `random.seed()` in fixtures. Statistical +assertions need generous tolerances — these are random processes, and a test that fails once a +week is worse than no test. + +Nothing here requires waiting in real time. Inject a large `dt` to make moves observable, and +drive `step()` directly rather than sleeping through the asyncio loop. diff --git a/planning/MASSIVE_API.md b/planning/MASSIVE_API.md new file mode 100644 index 000000000..7dac4d967 --- /dev/null +++ b/planning/MASSIVE_API.md @@ -0,0 +1,313 @@ +# Massive API Reference (formerly Polygon.io) + +Research notes for the FinAlly market data subsystem. Covers only what this project needs: +fetching current and end-of-day prices for a handful of tickers. + +Polygon.io rebranded to Massive on 30 October 2025. Existing API keys, code, and +`api.polygon.io` URLs continue to work; the Python SDK and docs have moved to the new name. + +## Package and Client + +The official SDK is the `massive` PyPI package. This project pins **massive 2.2.0**. + +```bash +uv add massive +``` + +```python +from massive import RESTClient + +client = RESTClient(api_key="your-key") # explicit +client = RESTClient() # reads MASSIVE_API_KEY from the environment +``` + +Verified against the installed package: `RESTClient()` with no argument picks up +`MASSIVE_API_KEY` automatically. FinAlly passes the key explicitly anyway, because the +factory has already read the variable to decide simulator-vs-real (see `MARKET_INTERFACE.md`). + +Default base URL is `https://api.massive.com`. Constructor options worth knowing: + +```python +RESTClient( + api_key=None, + connect_timeout=10.0, + read_timeout=10.0, + retries=3, # SDK retries failed requests itself + base="https://api.massive.com", + pagination=True, +) +``` + +The client is **synchronous** (urllib3 under the hood). In an async FastAPI app every call +must be wrapped in `asyncio.to_thread(...)` or it blocks the event loop. + +## Authentication + +Two equivalent forms: + +```bash +# Query parameter +curl "https://api.massive.com/v2/aggs/ticker/AAPL/prev?apiKey=YOUR_KEY" + +# Authorization header (preferred - keeps the key out of logs and URLs) +curl -H "Authorization: Bearer YOUR_KEY" \ + "https://api.massive.com/v2/aggs/ticker/AAPL/prev" +``` + +The SDK uses the header form. Keys come from https://massive.com/dashboard/keys. + +## Rate Limits and Plan Tiers + +| Tier | Requests | Data recency | History | +|---|---|---|---| +| Basic (free) | 5 / minute | End-of-day | 2 years | +| Starter / Developer | Unlimited | 15-minute delayed | 5-10 years | +| Advanced / Business | Unlimited | Real-time | Back to Sept 2003 | + +The free tier's 5 requests/minute is the binding constraint on FinAlly's design: it permits +one request every 12 seconds. This is why the poller fetches **all watched tickers in a +single request** and defaults to a 15-second interval, rather than one request per ticker. + +## Endpoints + +### Full Market Snapshot — the one FinAlly polls + +Returns a complete snapshot for many tickers in one request. This is the workhorse. + +``` +GET /v2/snapshot/locale/us/markets/stocks/tickers +``` + +| Parameter | Type | Notes | +|---|---|---| +| `tickers` | comma-separated list | Case-sensitive. Omit or pass empty to get all 10,000+ tickers | +| `include_otc` | boolean | Defaults to `false` | + +```python +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +client = RESTClient(api_key=key) +snapshots = client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "GOOGL", "MSFT"], +) + +for snap in snapshots: + print(snap.ticker, snap.last_trade.price, snap.todays_change_percent) +``` + +Signature: + +```python +get_snapshot_all( + market_type: str | SnapshotMarketType, + tickers: str | List[str] | None = None, + params: Dict[str, Any] | None = None, + raw: bool = False, + include_otc: bool | None = False, +) -> List[TickerSnapshot] +``` + +`TickerSnapshot` fields (from the installed package): + +``` +ticker str +last_trade LastTradeSnapshot (.price, .size, .timestamp, .exchange) +last_quote LastQuoteSnapshot (.bid_price, .ask_price, .timestamp) +day Agg (.open, .high, .low, .close, .volume, .vwap) +min MinuteSnapshot (.open, .high, .low, .close, .volume, .timestamp) +prev_day Agg (previous session OHLCV) +todays_change float +todays_change_percent float +updated int (nanoseconds - see below) +fair_market_value float | None +``` + +Raw JSON, if you bypass the SDK, uses short keys — `lastTrade.p` is price, `day.c` is close, +`todaysChangePerc`, and so on. The SDK's long names map onto these. + +### Single Ticker Snapshot + +``` +GET /v2/snapshot/locale/us/markets/stocks/tickers/{stocksTicker} +``` + +```python +snap = client.get_snapshot_ticker(SnapshotMarketType.STOCKS, "AAPL") +``` + +Not used by FinAlly — one request per ticker exhausts the free tier's budget almost +immediately. Documented for completeness. + +### Unified Snapshot (v3) + +A newer, paginated endpoint spanning asset classes. + +``` +GET /v3/snapshot?ticker.any_of=AAPL,GOOGL,MSFT&limit=250 +``` + +| Parameter | Notes | +|---|---| +| `ticker.any_of` | Comma-separated, **max 250 tickers** | +| `limit` | Default 10, max 250 | +| `type` | `stocks`, `options`, `fx`, `crypto`, `indices` | +| `sort`, `order`, `ticker.gt/gte/lt/lte` | Filtering and ordering | + +```python +for snap in client.list_universal_snapshots( + type="stocks", ticker_any_of=["AAPL", "GOOGL"], limit=250 +): + print(snap.ticker, snap.session) +``` + +Note the **default `limit` is 10** — asking for 30 tickers without raising `limit` silently +returns 10. FinAlly stays on the v2 endpoint, which has no such trap and returns exactly the +tickers requested. + +### Custom Bars / Aggregates — for sparkline history + +``` +GET /v2/aggs/ticker/{stocksTicker}/range/{multiplier}/{timespan}/{from}/{to} +``` + +| Parameter | Notes | +|---|---| +| `multiplier` | Integer scaling the timespan (e.g. `1`) | +| `timespan` | `minute`, `hour`, `day`, `week`, `month`, `quarter`, `year` | +| `from` / `to` | `YYYY-MM-DD` or millisecond epoch | +| `adjusted` | Split-adjusted; defaults `true` | +| `sort` | `asc` or `desc` | +| `limit` | Default 5,000, max 50,000 | + +```python +bars = client.get_aggs( + ticker="AAPL", multiplier=1, timespan="minute", + from_="2026-08-03", to="2026-08-03", limit=60, sort="desc", +) +history = [bar.close for bar in reversed(bars)] # oldest first +``` + +`Agg` fields: `open`, `high`, `low`, `close`, `volume`, `vwap`, `timestamp`, +`transactions`, `otc`. + +This is how FinAlly backfills the ~60 points of sparkline history when running against real +data. One request per ticker, made once at startup — acceptable because it is not on the +polling loop. + +### Previous Day Bar + +``` +GET /v2/aggs/ticker/{stocksTicker}/prev +``` + +```python +prev = client.get_previous_close_agg("AAPL") +print(prev.close) +``` + +Returns the prior session's OHLCV. Available on every tier including free, which makes it the +reliable fallback for a session baseline when the market is closed. + +### Daily Ticker Summary — end-of-day open/close + +``` +GET /v1/open-close/{stocksTicker}/{date} +``` + +```python +day = client.get_daily_open_close_agg(ticker="AAPL", date="2026-08-03") +print(day.open, day.close, day.pre_market, day.after_hours) +``` + +`DailyOpenCloseAgg` fields: `open`, `high`, `low`, `close`, `volume`, `pre_market`, +`after_hours`, `from_`, `status`, `symbol`, `otc`. + +This is the true end-of-day endpoint, including extended-hours prices. Useful if FinAlly ever +wants a real "official daily open" rather than the session baseline described in `PLAN.md`. + +### Market Status + +```python +status = client.get_market_status() # GET /v1/marketstatus/now +``` + +Tells you whether the market is open, closed, or in extended hours. Worth surfacing in +`/api/health` so "prices are not moving" can be explained rather than debugged. + +## Timestamp Units — the trap + +Massive uses **different time units in different places**, and mixing them up produces +timestamps tens of thousands of years off without raising an error. + +| Field | Unit | +|---|---| +| `last_trade.timestamp`, `last_quote.timestamp`, `snapshot.updated` | **nanoseconds** | +| `Agg.timestamp` (bars from `/v2/aggs`), `min.t` | **milliseconds** | + +Proof, using the `lastTrade.t` value from Massive's own sample response +(`1605195918306274000`): + +``` +/ 1e3 (as ms) -> out of range +/ 1e6 (as us) -> out of range +/ 1e9 (as ns) -> 2020-11-12 15:45:18 UTC correct +``` + +So the conversion to the Unix-seconds float that `PriceCache` expects is: + +```python +timestamp = snap.last_trade.timestamp / 1_000_000_000.0 # ns -> s +bar_time = bar.timestamp / 1_000.0 # ms -> s +``` + +**This is currently wrong in the codebase.** `backend/app/market/massive_client.py:106` reads +`timestamp = snap.last_trade.timestamp / 1000.0`, treating nanoseconds as milliseconds. Every +real-data price lands in the cache stamped roughly 50,000 years in the future. It has gone +unnoticed because the default path is the simulator and nothing yet reads the timestamp. It +must be fixed before the Massive path is trusted — see `MARKET_INTERFACE.md`. + +## Behaviour Outside Market Hours + +US equities trade 09:30-16:00 ET, weekdays. Outside that window `last_trade` holds the final +trade of the previous session and does not change. Consequences for FinAlly: + +- Every price is flat; no flash animations fire +- `change_from_open_percent` sits at zero +- The SSE cache version never advances, so the stream emits only heartbeats + +This is correct behaviour, not a fault. It is the main reason the simulator remains the +recommended default for demos and for the course. A UI running on real data at 21:00 looks +broken to someone who does not know the market is shut, so `/api/health` reporting +`market_source` and price age matters. + +Free-tier data is additionally 15 minutes delayed, and the free tier's snapshot access is +limited — the single-ticker and full-market snapshot endpoints require Starter or above for +real-time. On the free tier the aggregate endpoints (`/v2/aggs/...`) are the dependable ones. + +## Error Handling + +| Status | Cause | Response | +|---|---|---| +| 401 | Missing or invalid key | Log once, keep polling; do not crash the app | +| 403 | Endpoint not in plan | Log the endpoint; consider falling back to aggregates | +| 429 | Rate limit exceeded | Back off; raise the poll interval | +| 5xx / network | Transient | SDK retries 3 times; the poll loop retries on the next tick | + +The governing rule for FinAlly: **a market data failure must never take down the app**. The +poller catches broadly, logs, and lives to poll again. Stale prices in the cache are strictly +better than a 500 on every portfolio request. + +## Sources + +- [Polygon.io is Now Massive](https://massive.com/blog/polygon-is-now-massive) +- [Stocks REST API Overview](https://massive.com/docs/rest/stocks/overview) +- [Full Market Snapshot](https://massive.com/docs/rest/stocks/snapshots/full-market-snapshot) +- [Single Ticker Snapshot](https://massive.com/docs/rest/stocks/snapshots/single-ticker-snapshot) +- [Unified Snapshot](https://massive.com/docs/rest/stocks/snapshots/unified-snapshot) +- [Custom Bars (OHLC)](https://massive.com/docs/rest/stocks/aggregates/custom-bars) +- [Previous Day Bar](https://massive.com/docs/rest/stocks/aggregates/previous-day-bar) +- [Daily Ticker Summary](https://massive.com/docs/rest/stocks/aggregates/daily-ticker-summary) +- [Official Python client](https://github.com/massive-com/client-python) +- [Request limits](https://polygon.io/knowledge-base/article/what-is-the-request-limit-for-polygons-restful-apis) diff --git a/planning/PLAN.md b/planning/PLAN.md index bc1811b33..148d32ccd 100644 --- a/planning/PLAN.md +++ b/planning/PLAN.md @@ -22,7 +22,7 @@ The user runs a single Docker command (or a provided start script). A browser op ### What the User Can Do - **Watch prices stream** — prices flash green (uptick) or red (downtick) with subtle CSS animations that fade -- **View sparkline mini-charts** — price action beside each ticker in the watchlist, accumulated on the frontend from the SSE stream since page load (sparklines fill in progressively) +- **View sparkline mini-charts** — price action beside each ticker in the watchlist. `GET /api/watchlist` returns ~60 points of recent history per ticker so sparklines are populated on first paint; the frontend then extends them live from the SSE stream - **Click a ticker** to see a larger detailed chart in the main chart area - **Buy and sell shares** — market orders only, instant fill at current price, no fees, no confirmation dialog - **Monitor their portfolio** — a heatmap (treemap) showing positions sized by weight and colored by P&L, plus a P&L chart tracking total portfolio value over time @@ -34,7 +34,7 @@ The user runs a single Docker command (or a provided start script). A browser op - **Dark theme**: backgrounds around `#0d1117` or `#1a1a2e`, muted gray borders, no pure black - **Price flash animations**: brief green/red background highlight on price change, fading over ~500ms via CSS transitions -- **Connection status indicator**: a small colored dot (green = connected, yellow = reconnecting, red = disconnected) visible in the header +- **Connection status indicator**: a small colored dot in the header, driven by observable `EventSource` state — green = open and something (price event or heartbeat) received within the last 30s; yellow = `readyState === CONNECTING` after an error, or open but silent for more than 30s; red = `readyState === CLOSED` - **Professional, data-dense layout**: inspired by Bloomberg/trading terminals — every pixel earns its place - **Responsive but desktop-first**: optimized for wide screens, functional on tablet @@ -88,7 +88,13 @@ The user runs a single Docker command (or a provided start script). A browser op finally/ ├── frontend/ # Next.js TypeScript project (static export) ├── backend/ # FastAPI uv project (Python) -│ └── db/ # Schema definitions, seed data, migration logic +│ ├── app/ +│ │ ├── market/ # Market data subsystem (BUILT) +│ │ ├── db/ # Schema SQL, seed data, lazy init +│ │ └── ... # Portfolio, watchlist, chat routers +│ ├── tests/ # pytest suite +│ ├── market_data_demo.py # Rich terminal demo of the price stream +│ └── pyproject.toml ├── planning/ # Project-wide documentation for agents │ ├── PLAN.md # This document │ └── ... # Additional agent reference docs @@ -97,8 +103,8 @@ finally/ │ ├── stop_mac.sh # Stop Docker container (macOS/Linux) │ ├── start_windows.ps1 # Launch Docker container (Windows PowerShell) │ └── stop_windows.ps1 # Stop Docker container (Windows PowerShell) -├── test/ # Playwright E2E tests + docker-compose.test.yml -├── db/ # Volume mount target (SQLite file lives here at runtime) +├── test/ # Playwright E2E tests (run on the host) +├── db/ # Bind mount target (SQLite file lives here at runtime) │ └── .gitkeep # Directory exists in repo; finally.db is gitignored ├── Dockerfile # Multi-stage build (Node → Python) ├── docker-compose.yml # Optional convenience wrapper @@ -110,18 +116,20 @@ finally/ - **`frontend/`** is a self-contained Next.js project. It knows nothing about Python. It talks to the backend via `/api/*` endpoints and `/api/stream/*` SSE endpoints. Internal structure is up to the Frontend Engineer agent. - **`backend/`** is a self-contained uv project with its own `pyproject.toml`. It owns all server logic including database initialization, schema, seed data, API routes, SSE streaming, market data, and LLM integration. Internal structure is up to the Backend/Market Data agents. -- **`backend/db/`** contains schema SQL definitions and seed logic. The backend lazily initializes the database on first request — creating tables and seeding default data if the SQLite file doesn't exist or is empty. +- **`backend/app/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. - **`planning/`** contains project-wide documentation, including this plan. All agents reference files here as the shared contract. -- **`test/`** contains Playwright E2E tests and supporting infrastructure (e.g., `docker-compose.test.yml`). Unit tests live within `frontend/` and `backend/` respectively, following each framework's conventions. +- **`test/`** contains Playwright E2E tests, which run on the host against the running container. Unit tests live within `frontend/` and `backend/` respectively, following each framework's conventions. - **`scripts/`** contains start/stop scripts that wrap Docker commands. --- ## 5. Environment Variables +Copy `.env.example` to `.env` at the project root and fill in what you have. Docker passes the file with `--env-file .env`. For local development outside Docker the backend runs from `backend/` but loads `../.env`, so there is only ever one env file. + ```bash -# Required: OpenRouter API key for LLM chat functionality +# OpenRouter API key for LLM chat. Everything except /api/chat works without it. OPENROUTER_API_KEY=your-openrouter-api-key-here # Optional: Massive (Polygon.io) API key for real market data @@ -137,6 +145,7 @@ LLM_MOCK=false - If `MASSIVE_API_KEY` is set and non-empty → backend uses Massive REST API for market data - If `MASSIVE_API_KEY` is absent or empty → backend uses the built-in market simulator - If `LLM_MOCK=true` → backend returns deterministic mock LLM responses (for E2E tests) +- If `OPENROUTER_API_KEY` is absent or empty → the app still starts and every other feature works. `/api/chat` returns a normal-shaped response whose `message` explains that no API key is configured, with empty `trades` and `watchlist_changes`. It never raises, and startup never fails on a missing key. - The backend reads `.env` from the project root (mounted into the container or read via docker `--env-file`) --- @@ -155,6 +164,9 @@ Both the simulator and the Massive client implement the same abstract interface. - Occasional random "events" — sudden 2-5% moves on a ticker for drama - Starts from realistic seed prices (e.g., AAPL ~$190, GOOGL ~$175, etc.) - Runs as an in-process background task — no external dependencies +- **Unknown tickers are accepted, not rejected.** `seed_prices.py` only has real parameters for the ten defaults. Any other symbol gets a seed price and volatility synthesized deterministically from a hash of the symbol (so `PYPL` is the same price on every run), and joins the cross-sector correlation group. This keeps the "AI manages your watchlist" demo from dead-ending on a plausible symbol the simulator has never heard of. +- **Ticker validation is one shared rule** used by the manual and LLM paths alike: uppercase the input, then accept `^[A-Z]{1,5}$`. Anything else is rejected with a 400. +- **History backfill.** On startup the simulator generates ~60 points of prior GBM history per ticker (and does the same for each newly added ticker) so sparklines are populated on first paint instead of filling in over the first 30 seconds. ### Massive API (Optional) @@ -163,6 +175,7 @@ Both the simulator and the Massive client implement the same abstract interface. - 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 +- **Outside market hours real quotes do not move.** Overnight and at weekends every price is flat, no flash animations fire, and change percentages sit at zero. This is correct behavior, not a bug. The simulator remains the recommended default for demos. ### Shared Price Cache @@ -171,13 +184,35 @@ Both the simulator and the Massive client implement the same abstract interface. - SSE streams read from this cache and push updates to connected clients - This architecture supports future multi-user scenarios without changes to the data layer +### Session Baseline (the "change %" the user sees) + +`PriceUpdate.change_percent` compares against the *previous tick* — a number that flickers around zero every 500ms and is useless as a "daily change" column. The cache therefore also records an **open price** per ticker: the first price seen for that ticker after process start (for a ticker added later, the first price after it was added). + +- `PriceCache` stores `open_price` alongside latest and previous price +- `PriceUpdate` gains `open_price` and `change_from_open_percent` +- Both appear in the SSE payload; the watchlist "change %" column and the price-flash coloring use `change_from_open_percent` +- The baseline survives page reloads and reconnects, and resets on container restart — acceptable for a simulation + +This is an addition to the already-built `backend/app/market/` module and is the one change the market data subsystem still needs. + ### SSE Streaming - Endpoint: `GET /api/stream/prices` - Long-lived SSE connection; client uses native `EventSource` API -- Server pushes price updates for all tickers known to the system at a regular cadence (~500ms) — in the single-user model this is equivalent to the user's watchlist -- Each SSE event contains ticker, price, previous price, timestamp, and change direction -- Client handles reconnection automatically (EventSource has built-in retry) +- The server polls the price cache every ~500ms and emits **one event containing every tracked ticker**, keyed by symbol — not one event per ticker. It emits only when the cache version has changed, so a quiet market produces no price events. +- A heartbeat comment frame (`: ping\n\n`) is emitted every 15s regardless of price activity. This keeps proxies from dropping an idle connection and lets the frontend distinguish "connected, market quiet" from "backend stalled". +- The stream opens with `retry: 1000`; the client handles reconnection automatically (EventSource has built-in retry) + +Payload shape: + +``` +data: {"AAPL": {"ticker":"AAPL","price":190.50,"previous_price":190.40, + "open_price":189.20,"timestamp":1753401234.5, + "change":0.10,"change_percent":0.052, + "change_from_open_percent":0.687,"direction":"up"}, ...} +``` + +**Timestamp convention:** SSE timestamps are Unix epoch seconds (float). Every REST timestamp — and every `*_at` column in the database — is an ISO 8601 UTC string. The two formats never mix within a single payload. --- @@ -215,6 +250,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)` +- A sell that takes quantity to zero **deletes the row** — there are no zero-quantity positions +- Invariant: every ticker with a position is on the watchlist (see section 8) **trades** — Trade history (append-only log) - `id` TEXT PRIMARY KEY (UUID) @@ -222,14 +259,17 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - `ticker` TEXT - `side` TEXT (`"buy"` or `"sell"`) - `quantity` REAL (fractional shares supported) -- `price` REAL +- `price` REAL (the actual fill price) - `executed_at` TEXT (ISO timestamp) +- This is an **audit log only** — no UI panel and no endpoint reads it. It exists so trade history survives, and so realized P&L can be derived later if a panel is ever added. **portfolio_snapshots** — Portfolio value over time (for P&L chart). Recorded every 30 seconds by a background task, and immediately after each trade execution. - `id` TEXT PRIMARY KEY (UUID) - `user_id` TEXT (default: `"default"`) - `total_value` REAL - `recorded_at` TEXT (ISO timestamp) +- One snapshot is written at seed time so the P&L chart has a data point on first launch +- The background task skips the write when total value is unchanged since the last snapshot, which keeps an idle portfolio from accumulating identical rows **chat_messages** — Conversation history with LLM - `id` TEXT PRIMARY KEY (UUID) @@ -239,6 +279,10 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - `actions` TEXT (JSON — trades executed, watchlist changes made; null for user messages) - `created_at` TEXT (ISO timestamp) +### What Is Not Tracked + +Realized P&L has no column and no display. On a sell, the proceeds land in `cash_balance` and the position's cost basis goes away with it. Total portfolio value stays correct and the positions table shows *unrealized* P&L only. This is a deliberate simplification — the `trades` table holds everything needed to compute realized P&L should it ever be wanted. + ### Default Seed Data - One user profile: `id="default"`, `cash_balance=10000.0` @@ -258,24 +302,76 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod |--------|------|-------------| | GET | `/api/portfolio` | Current positions, cash balance, total value, unrealized P&L | | POST | `/api/portfolio/trade` | Execute a trade: `{ticker, quantity, side}` | -| GET | `/api/portfolio/history` | Portfolio value snapshots over time (for P&L chart) | +| GET | `/api/portfolio/history` | Portfolio value snapshots over time (for P&L chart). Query params: `?limit=` (default 500, newest first) and `?since=` (ISO timestamp) | ### Watchlist | Method | Path | Description | |--------|------|-------------| -| GET | `/api/watchlist` | Current watchlist tickers with latest prices | +| GET | `/api/watchlist` | Watchlist tickers with latest price and ~60 points of recent history for sparklines | | POST | `/api/watchlist` | Add a ticker: `{ticker}` | -| DELETE | `/api/watchlist/{ticker}` | Remove a ticker | +| DELETE | `/api/watchlist/{ticker}` | Remove a ticker. Rejected with 409 if a position is held in it | ### Chat | Method | Path | Description | |--------|------|-------------| +| GET | `/api/chat` | Conversation history, oldest first, so the panel repopulates after a reload. Query param: `?limit=` (default 100) | | POST | `/api/chat` | Send a message, receive complete JSON response (message + executed actions) | ### System | Method | Path | Description | |--------|------|-------------| -| GET | `/api/health` | Health check (for Docker/deployment) | +| GET | `/api/health` | Health check. Returns `{status, market_source, tickers_cached, newest_price_age_seconds}` — enough to answer "is the stream alive?" in one request | + +### Request and Response Shapes + +```jsonc +// GET /api/portfolio +{ + "cash_balance": 8234.50, + "total_value": 10120.75, + "positions": [ + {"ticker": "AAPL", "quantity": 10, "avg_cost": 188.60, + "current_price": 190.50, "market_value": 1905.00, + "unrealized_pnl": 19.00, "unrealized_pnl_percent": 1.007} + ] +} + +// POST /api/portfolio/trade -> {"ticker": "AAPL", "quantity": 10, "side": "buy"} +{ + "ticker": "AAPL", "side": "buy", "quantity": 10, + "fill_price": 190.52, // the server-side price, not what the client saw + "total_cost": 1905.20, + "cash_balance": 8234.50, + "executed_at": "2026-07-25T14:03:11Z" +} + +// GET /api/watchlist +{ + "tickers": [ + {"ticker": "AAPL", "price": 190.50, "open_price": 189.20, + "change_from_open_percent": 0.687, + "history": [189.20, 189.35, ...]} // ~60 points, oldest first + ] +} + +// GET /api/portfolio/history +{"snapshots": [{"total_value": 10000.0, "recorded_at": "2026-07-25T14:00:00Z"}]} + +// GET /api/chat +{"messages": [{"role": "user", "content": "...", "actions": null, + "created_at": "2026-07-25T14:02:00Z"}]} +``` + +Errors use FastAPI's default envelope — `{"detail": "Insufficient cash: need $1905.20, have $800.00"}` — with 400 for validation and business-rule failures, 404 for unknown tickers on DELETE, and 409 for removing a watchlist entry with an open position. Messages are written to be shown to the user verbatim. + +### Trade Rules + +- **Market orders only, no shorting, no margin.** Buys require sufficient cash; sells require sufficient shares. There is no borrowing on either side. +- **Fill price is the server's price.** The client's displayed price is advisory; the server fills at whatever is in the price cache when the request lands and returns that as `fill_price`. The UI shows the fill it got, not the price that was clicked. +- **A just-added ticker may not have a price yet.** Rather than failing, the endpoint polls the cache for up to 2 seconds (every 200ms) waiting for a first tick. The simulator publishes within ~500ms, so this effectively never expires; if it does, return 400 with a "no price available yet, try again" message. +- **Quantity** must be a finite number greater than zero, rounded to at most 4 decimal places. Zero, negative, `NaN`, and `Infinity` are 400s. +- **Trading a ticker that is not on the watchlist adds it to the watchlist** as part of the trade. This holds the invariant that every position has a live price feed, and it is what makes "forbid removing a ticker you hold" sufficient on its own. +- **Every trade writes a `portfolio_snapshots` row immediately**, so the P&L chart shows the step. --- @@ -314,9 +410,11 @@ 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) -- `watchlist_changes` (optional): Array of watchlist modifications +**All three fields are required**, with empty arrays when there is nothing to do. No optional keys — structured outputs are more reliable without them, and the parsing code loses its `None` branches. A response with no actions is `{"message": "...", "trades": [], "watchlist_changes": []}`. + +- `message`: The conversational text shown to the user +- `trades`: Trades to auto-execute. Each goes through exactly the same validation as a manual trade — same fill-price rule, same quantity rules, same auto-add-to-watchlist behavior +- `watchlist_changes`: Watchlist modifications, subject to the same `^[A-Z]{1,5}$` rule and the same "cannot remove a held ticker" rule ### Auto-Execution @@ -327,6 +425,10 @@ Trades specified by the LLM execute automatically — no confirmation dialog. Th If a trade fails validation (e.g., insufficient cash), the error is included in the chat response so the LLM can inform the user. +### Keeping the UI in Sync + +The LLM changes server state the user did not directly cause. Rather than push those changes down a second channel, the frontend refetches `/api/portfolio` and `/api/watchlist` after any chat response whose `trades` or `watchlist_changes` are non-empty — and after any manual trade. That one rule covers every path by which portfolio state changes. + ### System Prompt Guidance The LLM should be prompted as "FinAlly, an AI trading assistant" with instructions to: @@ -344,6 +446,18 @@ When `LLM_MOCK=true`, the backend returns deterministic mock responses instead o - Development without an API key - CI/CD pipelines +The mock is keyword-triggered on the lowercased user message, checked in this order. The E2E suite asserts against this contract, so it is part of the spec, not an implementation detail: + +| Message contains | Mock response | +|---|---| +| `"buy"` | `trades: [{ticker, side: "buy", quantity: 1}]` — ticker is the first `^[A-Z]{1,5}$` token in the message, else `AAPL` | +| `"sell"` | Same, with `side: "sell"` | +| `"watch"` or `"add"` | `watchlist_changes: [{ticker, action: "add"}]`, same ticker extraction, defaulting to `PYPL` | +| `"remove"` | `watchlist_changes: [{ticker, action: "remove"}]` | +| anything else | A fixed analysis string that echoes live portfolio numbers — `"You are holding N positions worth $X with $Y in cash."` — and empty arrays | + +The mock still routes its trades and watchlist changes through the real execution and validation path, so an E2E test that mocks the LLM is still exercising genuine trade logic. + --- ## 10. Frontend Design @@ -352,7 +466,7 @@ When `LLM_MOCK=true`, the backend returns deterministic mock responses instead o The frontend is a single-page application with a dense, terminal-inspired layout. The specific component architecture and layout system is up to the Frontend Engineer, but the UI should include these elements: -- **Watchlist panel** — grid/table of watched tickers with: ticker symbol, current price (flashing green/red on change), daily change %, and a sparkline mini-chart (accumulated from SSE since page load) +- **Watchlist panel** — grid/table of watched tickers with: ticker symbol, current price (flashing green/red on change), change % since the session open (`change_from_open_percent`, not the tick-over-tick number), and a sparkline mini-chart seeded from the watchlist response and extended from SSE - **Main chart area** — larger chart for the currently selected ticker, with at minimum price over time. Clicking a ticker in the watchlist selects it here. - **Portfolio heatmap** — treemap visualization where each rectangle is a position, sized by portfolio weight, colored by P&L (green = profit, red = loss) - **P&L chart** — line chart showing total portfolio value over time, using data from `portfolio_snapshots` @@ -361,10 +475,14 @@ The frontend is a single-page application with a dense, terminal-inspired layout - **AI chat panel** — docked/collapsible sidebar. Message input, scrolling conversation history, loading indicator while waiting for LLM response. Trade executions and watchlist changes shown inline as confirmations. - **Header** — portfolio total value (updating live), connection status indicator, cash balance +### Live Values Are Computed on the Client + +The only live channel is the price stream. There is no portfolio SSE channel and no polling loop. The client holds `cash_balance` and positions from `/api/portfolio` and recomputes `cash + Σ(quantity × live price)` on every SSE frame. The same derivation drives the header total, the positions table's current-price and P&L columns, the heatmap colors, and the live end of the P&L line. Server state is refetched only on the events listed in section 9. + ### Technical Notes - Use `EventSource` for SSE connection to `/api/stream/prices` -- Canvas-based charting library preferred (Lightweight Charts or Recharts) for performance +- **Recharts for every chart** — the line chart, the sparklines, and the treemap. Lightweight Charts has no treemap, so choosing it would mean shipping a second charting library and a second bundle for one panel. One library, one visual language. - 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 @@ -376,30 +494,34 @@ The frontend is a single-page application with a dense, terminal-inspired layout ### Multi-Stage Dockerfile ``` -Stage 1: Node 20 slim +Stage 1: Node 22 slim - Copy frontend/ - - npm install && npm run build (produces static export) + - npm ci && npm run build (produces static export) Stage 2: Python 3.12 slim - Install uv - Copy backend/ - - uv sync (install Python dependencies from lockfile) + - uv sync --frozen --no-dev (install Python dependencies from lockfile) - Copy frontend build output into a static/ directory - Expose port 8000 - CMD: uvicorn serving FastAPI app ``` +`npm ci` and `uv sync --frozen` build from the lockfiles rather than re-resolving, which is the reason for having lockfiles. + FastAPI serves the static frontend files and all API routes on port 8000. +**Mount order matters.** `app.mount("/", StaticFiles(directory="static", html=True))` must come *after* every `/api/*` router is registered. Mounted first, it shadows the API and every endpoint 404s while the UI appears to work — the most common way this architecture breaks. + ### Docker Volume -The SQLite database persists via a named Docker volume: +The SQLite database persists via a bind mount to the `db/` directory in the project root: ```bash -docker run -v finally-data:/app/db -p 8000:8000 --env-file .env finally +docker run -v "$PWD/db:/app/db" -p 8000:8000 --env-file .env finally ``` -The `db/` directory in the project root maps to `/app/db` in the container. The backend writes `finally.db` to this path. +The backend writes `finally.db` to this path. A bind mount rather than a named volume is deliberate for a teaching project: students can see the database file, inspect it, and delete it to reset. The start scripts use the platform-appropriate form of `$PWD`. ### Start/Stop Scripts @@ -442,15 +564,68 @@ The container is designed to deploy to AWS App Runner, Render, or any container ### E2E Tests (in `test/`) -**Infrastructure**: A separate `docker-compose.test.yml` in `test/` that spins up the app container plus a Playwright container. This keeps browser dependencies out of the production image. +**Infrastructure**: Playwright runs **on the host** against the container started by the normal start script — `npx playwright test` pointed at `http://localhost:8000`. No test compose file, no second container, no service graph, no networking hop. Browser dependencies stay out of the production image because they were never in it. This is materially simpler for students, particularly on Windows. -**Environment**: Tests run with `LLM_MOCK=true` by default for speed and determinism. +**Environment**: Tests run with `LLM_MOCK=true` by default for speed and determinism, and assert against the mock contract in section 9. **Key Scenarios**: -- Fresh start: default watchlist appears, $10k balance shown, prices are streaming +- Fresh start: default watchlist appears, $10k balance shown, prices are streaming, sparklines are already populated - Add and remove a ticker from the watchlist -- Buy shares: cash decreases, position appears, portfolio updates -- Sell shares: cash increases, position updates or disappears +- Removing a ticker with an open position is rejected with a visible error +- Buy shares: cash decreases, position appears, portfolio updates, fill price is displayed +- Sell shares: cash increases, position updates or disappears entirely when it hits zero - Portfolio visualization: heatmap renders with correct colors, P&L chart has data points -- AI chat (mocked): send a message, receive a response, trade execution appears inline -- SSE resilience: disconnect and verify reconnection +- AI chat (mocked): send a message, receive a response, trade execution appears inline; reload the page and the conversation is still there +- SSE resilience: block the `/api/stream/prices` route with `page.route()`, assert the status dot leaves green, unblock, assert it returns to green + +--- + +## 13. Build Order + +The market data subsystem (`backend/app/market/`) is built and tested. Everything else is open. This order is what an agent may assume already exists when it starts. + +1. **Session baseline in the market module** — `open_price` and `change_from_open_percent` in `PriceCache` and `PriceUpdate`, history backfill, synthesized params for unknown tickers, SSE heartbeat. The one remaining change to an otherwise finished module. Do it first: the frontend contract depends on it. +2. **Database and portfolio API** — schema, lazy init, seed data, `/api/portfolio`, `/api/portfolio/trade`, `/api/portfolio/history`, the 30-second snapshot task. Reads the price cache for valuation. +3. **Watchlist API** — `/api/watchlist` CRUD, wired to `add_ticker` / `remove_ticker` on the market source. +4. **Frontend shell** — layout, SSE wiring, watchlist panel, header, trade bar. +5. **Charts** — main chart, sparklines, heatmap, P&L chart. Recharts throughout. +6. **Chat** — `/api/chat` GET and POST, mock mode, LLM integration, chat panel. +7. **Docker and start/stop scripts.** +8. **E2E tests.** + +Steps 1-3 are independent of each other and can run in parallel. Step 4 depends on 1 and 3. Everything from step 4 onward builds against the shapes in section 8, not against the backend implementation. + +--- + +## 14. Review Decisions + +A documentation review raised 24 issues against this plan. All are resolved in the sections above; this log records what was decided and where it now lives, so the reasoning survives. + +| # | Decision | Section | +|---|---|---| +| 1 | "Daily change %" is measured against a session open price recorded in `PriceCache`, not the previous tick | 6 | +| 2 | Unknown tickers are accepted; seed price and volatility are synthesized deterministically from the symbol | 6 | +| 3 | A watchlist ticker with an open position cannot be removed (409). Trades auto-add their ticker, so every position always has a feed | 7, 8 | +| 4 | Trades fill at the server-side price and return it. A ticker with no price yet is waited on for up to 2s rather than rejected | 8 | +| 5 | `GET /api/chat` added so the conversation survives a page reload | 8 | +| 6 | `LLM_MOCK` is keyword-triggered and specified as a contract the E2E suite asserts against | 9 | +| 7 | Bind mount `./db:/app/db`; the contradictory named-volume line is gone | 11 | +| 8 | SSE contract corrected to what is actually implemented: one event carrying all tickers, emitted only on change | 6 | +| 9 | Timestamps are epoch seconds on SSE and ISO 8601 UTC everywhere else | 6 | +| 10 | `OPENROUTER_API_KEY` is optional; chat degrades to a friendly message, the app does not fail | 5 | +| 11 | Request and response shapes written into section 8 directly, rather than a separate contract document — one document, one authority | 8 | +| 12 | Build order stated | 13 | +| 13 | Trade validation fully specified: no shorting or margin, quantity > 0 at 4dp, zero-quantity positions deleted | 7, 8 | +| 14 | Realized P&L is deliberately not tracked; derivable from `trades` if ever wanted | 7 | +| 15 | `trades` is an audit log with no reader and no UI | 7 | +| 16 | SSE heartbeat every 15s; connection dot defined in terms of observable `EventSource` state | 2, 6 | +| 17 | Sparklines seeded from `/api/watchlist`; a snapshot is written at seed time so no panel is empty on first paint | 2, 6, 7 | +| 18 | `/api/portfolio/history` takes `?limit=` and `?since=`; unchanged snapshots are skipped | 7, 8 | +| 19 | Massive off-hours flatness documented as expected behavior | 6 | +| 20 | One refetch rule covers every server-side state change the user did not directly cause | 9 | +| 21 | Live totals are computed on the client from cash, positions, and the SSE stream — no portfolio channel, no polling | 10 | +| 22 | Recharts for every chart, including the treemap | 10 | +| 23 | Playwright runs on the host against the container; no test compose file | 4, 12 | +| 24 | All three LLM output fields are required, with empty arrays as the default | 9 | + +Minor notes also applied: static mount ordering (11), `npm ci` and `uv sync --frozen --no-dev` (11), Node 22 (11), `.env.example` instructions (5), a `/api/health` payload worth reading (8), the SSE resilience test rewritten as route blocking (12), and the directory tree refreshed (4).