From cad076736bef62f832d5db785c58e090a5c0b20f Mon Sep 17 00:00:00 2001 From: Hoang Nguyen Date: Sun, 2 Aug 2026 22:13:33 +0800 Subject: [PATCH 1/4] commit --- planning/MARKET_INTERFACE.md | 438 ++++++++++++++++++++++++++++++++ planning/MARKET_SIMULATOR.md | 419 +++++++++++++++++++++++++++++++ planning/MASSIVE_API.md | 474 +++++++++++++++++++++++++++++++++++ 3 files changed, 1331 insertions(+) create mode 100644 planning/MARKET_INTERFACE.md create mode 100644 planning/MARKET_SIMULATOR.md create mode 100644 planning/MASSIVE_API.md diff --git a/planning/MARKET_INTERFACE.md b/planning/MARKET_INTERFACE.md new file mode 100644 index 000000000..f5d9762b6 --- /dev/null +++ b/planning/MARKET_INTERFACE.md @@ -0,0 +1,438 @@ +# Market Data Interface + +The unified Python API for retrieving stock prices in FinAlly. One interface, +two implementations: the **Massive API** when `MASSIVE_API_KEY` is set, the +**GBM simulator** otherwise. + +Companion documents: `MASSIVE_API.md` (the real API), `MARKET_SIMULATOR.md` +(the simulation model). + +--- + +## 1. Design Goal + +Every consumer of price data in FinAlly — the SSE stream, portfolio valuation, +trade execution, the LLM's context builder — must be **completely unaware of +where prices come from**. Swapping a simulated feed for a live one is an +environment-variable change, not a code change. + +This is achieved with two ideas: + +1. **A producer interface** (`MarketDataSource`) that both backends implement. +2. **A shared cache** (`PriceCache`) that decouples producers from consumers. + +``` + ┌──────────────────────────────────────────┐ + │ MarketDataSource (ABC) │ + │ start / stop / add_ticker / │ + │ remove_ticker / get_tickers │ + └──────────────────────────────────────────┘ + ▲ ▲ + │ │ + ┌───────────┴────────┐ ┌─────────┴──────────┐ + │ SimulatorDataSource│ │ MassiveDataSource │ + │ GBM, 500ms ticks │ │ REST poll, 15s │ + └───────────┬────────┘ └─────────┬──────────┘ + │ │ + └──────── writes ──────┘ + │ + ▼ + ┌───────────────────────┐ + │ PriceCache │ + │ thread-safe, in-mem │ + │ + version counter │ + └───────────┬───────────┘ + │ reads + ┌────────────────┼────────────────┐ + ▼ ▼ ▼ + SSE /api/stream Portfolio Trade execution + /prices valuation & LLM context +``` + +**The cache is the only contract consumers depend on.** No consumer ever holds +a reference to a `MarketDataSource`. + +--- + +## 2. Module Layout + +Located at `backend/app/market/`: + +| Module | Responsibility | +|---|---| +| `models.py` | `PriceUpdate` — the immutable unit of price data | +| `interface.py` | `MarketDataSource` — the abstract producer contract | +| `cache.py` | `PriceCache` — thread-safe store, versioned | +| `factory.py` | `create_market_data_source()` — environment-driven selection | +| `simulator.py` | `GBMSimulator` + `SimulatorDataSource` | +| `massive_client.py` | `MassiveDataSource` — REST poller | +| `seed_prices.py` | Seed prices, GBM params, correlation groups | +| `stream.py` | `create_stream_router()` — SSE endpoint factory | + +Public surface, re-exported from `app.market`: + +```python +from app.market import ( + PriceUpdate, + PriceCache, + MarketDataSource, + create_market_data_source, + create_stream_router, +) +``` + +--- + +## 3. `PriceUpdate` — The Data Unit + +An immutable, frozen, slotted dataclass. Immutability matters: instances are +handed to SSE serializers and portfolio math concurrently, and nothing should +be able to mutate a price after the fact. + +```python +@dataclass(frozen=True, slots=True) +class PriceUpdate: + ticker: str + price: float + previous_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: ... # price - previous_price + @property + def change_percent(self) -> float: ... # % change, 0.0 if previous == 0 + @property + def direction(self) -> str: ... # "up" | "down" | "flat" + + def to_dict(self) -> dict: ... # JSON/SSE serialization +``` + +Design decisions: + +- **Derived values are properties, not fields.** `change`, `change_percent`, + and `direction` cannot drift out of sync with the prices they derive from. +- **`previous_price` is the previous *tick*, not the previous *close*.** This + drives the frontend's green/red flash animation. Day-over-day change is a + separate concern computed by the portfolio layer. +- **`timestamp` is Unix seconds (float)**, normalized regardless of source. + Massive's nanosecond timestamps are converted at the boundary. +- **Division-by-zero guard** on `change_percent` returns `0.0` rather than + raising — a price feed glitch must not propagate an exception into the SSE + generator. + +`to_dict()` output — this is the wire format the frontend consumes: + +```json +{ + "ticker": "AAPL", + "price": 190.42, + "previous_price": 190.35, + "timestamp": 1785432000.123, + "change": 0.07, + "change_percent": 0.0368, + "direction": "up" +} +``` + +--- + +## 4. `MarketDataSource` — The Producer Contract + +```python +class MarketDataSource(ABC): + @abstractmethod + async def start(self, tickers: list[str]) -> None: ... + @abstractmethod + async def stop(self) -> None: ... + @abstractmethod + async def add_ticker(self, ticker: str) -> None: ... + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: ... + @abstractmethod + def get_tickers(self) -> list[str]: ... +``` + +### Contract semantics + +| Method | Guarantee | +|---|---| +| `start(tickers)` | Begins a background task writing to the cache. Called exactly once. Populates the cache with initial prices **before returning**, so the first SSE client never sees an empty payload. | +| `stop()` | Cancels the background task and releases resources. Idempotent — safe to call multiple times, including when never started. After `stop()`, the source never writes to the cache again. | +| `add_ticker(t)` | Idempotent. Ticker appears in the next update cycle (or immediately, for the simulator). | +| `remove_ticker(t)` | Idempotent. **Also removes the ticker from the cache**, so a de-watchlisted symbol stops appearing in the SSE payload. | +| `get_tickers()` | Synchronous (no I/O). Returns a copy — callers cannot mutate internal state. | + +### Why this shape + +- **Async lifecycle, sync accessor.** `start`/`stop`/`add`/`remove` are async + because the Massive implementation performs network I/O and both manage + asyncio tasks. `get_tickers()` is a pure in-memory read, so forcing `await` + on it would be noise. +- **No `get_price()` on the interface.** Deliberately absent. If sources + exposed price reads, consumers would couple to the source and the cache + would become an implementation detail rather than the contract. Prices are + read from `PriceCache`, always. +- **Push, not pull.** Sources write to the cache on their own schedule. + Consumers never trigger a fetch, so a slow API can never block a request. + +--- + +## 5. `PriceCache` — The Consumer Contract + +```python +class PriceCache: + def update(self, ticker: str, price: float, + timestamp: float | None = None) -> PriceUpdate: ... + def get(self, ticker: str) -> PriceUpdate | None: ... + def get_price(self, ticker: str) -> float | None: ... + def get_all(self) -> dict[str, PriceUpdate]: ... + def remove(self, ticker: str) -> None: ... + + @property + def version(self) -> int: ... + + def __len__(self) -> int: ... + def __contains__(self, ticker: str) -> bool: ... +``` + +### Thread safety + +Guarded by a `threading.Lock`, not an `asyncio.Lock`. This is deliberate: the +Massive client is synchronous and runs inside `asyncio.to_thread(...)`, so +writes genuinely originate from a worker thread while reads happen on the event +loop. An asyncio lock would not protect that boundary. + +Every method holds the lock for the shortest possible span. `get_all()` returns +a **shallow copy** of the dict — safe to iterate outside the lock because +`PriceUpdate` is frozen. + +### The version counter + +`version` is a monotonic integer incremented on every `update()`. It exists so +the SSE generator can answer "has anything changed since I last sent?" in O(1) +without diffing dicts or comparing timestamps. + +```python +last_version = -1 +while True: + if price_cache.version != last_version: + last_version = price_cache.version + yield f"data: {json.dumps(...)}\n\n" + await asyncio.sleep(0.5) +``` + +This keeps idle connections silent — when the simulator is stopped or the +Massive poll fails, no redundant frames are pushed. + +### First-update semantics + +On the first `update()` for a ticker, `previous_price` is set equal to `price`, +so `direction` is `"flat"` and `change` is `0.0`. The frontend therefore never +flashes a spurious green/red on page load. + +Prices are rounded to 2 decimal places on write. Rounding at the cache boundary +means every consumer sees identical values — no display/execution mismatch +where the UI shows $190.42 but a trade fills at $190.41999999. + +--- + +## 6. `create_market_data_source()` — Selection + +```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) +``` + +Selection rules: + +| `MASSIVE_API_KEY` | Result | +|---|---| +| Unset | Simulator | +| Empty string | Simulator | +| Whitespace only | Simulator (`.strip()` handles this) | +| Non-empty | Massive | + +The `.strip()` matters in practice: a `.env` file containing `MASSIVE_API_KEY=` +followed by a trailing space would otherwise select the real API with a garbage +key, and the app would silently show no prices. + +The factory returns an **unstarted** source — the caller owns the lifecycle. +This keeps the factory synchronous and makes it trivial to unit-test selection +logic without spawning background tasks. + +--- + +## 7. Application Wiring + +```python +from contextlib import asynccontextmanager +from fastapi import FastAPI +from app.market import PriceCache, create_market_data_source, create_stream_router + + +@asynccontextmanager +async def lifespan(app: FastAPI): + cache = PriceCache() + source = create_market_data_source(cache) + + tickers = load_watchlist_tickers() # from SQLite + await source.start(tickers) + + app.state.price_cache = cache + app.state.market_source = source + try: + yield + finally: + await source.stop() + + +app = FastAPI(lifespan=lifespan) +app.include_router(create_stream_router(app.state.price_cache)) +``` + +Both the cache and the source live on `app.state` rather than in module-level +globals, which keeps tests isolated — each test app gets its own cache. + +### Keeping the source in sync with the watchlist + +Watchlist mutations must update the database *and* the data source: + +```python +@router.post("/api/watchlist") +async def add_to_watchlist(body: TickerBody, request: Request): + ticker = body.ticker.upper().strip() + db_add_ticker(ticker) + await request.app.state.market_source.add_ticker(ticker) + return {"ok": True} + + +@router.delete("/api/watchlist/{ticker}") +async def remove_from_watchlist(ticker: str, request: Request): + ticker = ticker.upper().strip() + db_remove_ticker(ticker) + await request.app.state.market_source.remove_ticker(ticker) + return {"ok": True} +``` + +Ticker normalization (`.upper().strip()`) happens at the API boundary *and* +defensively inside both sources, since the cache is keyed by exact string. + +--- + +## 8. Reading Prices — Consumer Patterns + +```python +cache: PriceCache = request.app.state.price_cache + +# Single price for trade execution +price = cache.get_price("AAPL") +if price is None: + raise HTTPException(400, "No price available for AAPL") + +# Full update with direction, for display +update = cache.get("AAPL") + +# Portfolio valuation across all positions +prices = cache.get_all() +total = cash + sum( + pos.quantity * prices[pos.ticker].price + for pos in positions + if pos.ticker in prices +) +``` + +**Always handle `None`.** A ticker can be absent from the cache when it was +just added and the first poll has not landed, or when a Massive poll failed +before any successful fetch. Trade execution must reject rather than assume a +price. + +--- + +## 9. SSE Streaming Layer + +`create_stream_router(cache)` returns a FastAPI `APIRouter` exposing +`GET /api/stream/prices`. + +``` +retry: 1000 + +data: {"AAPL": {"ticker":"AAPL","price":190.42,...}, "GOOGL": {...}} + +data: {...} +``` + +Behaviour: + +- Emits a `retry: 1000` directive first, so `EventSource` reconnects after 1s. +- Sends **all** tracked tickers in each frame, keyed by ticker. Simpler for the + client than per-ticker events, and at ~10 tickers the payload is trivial. +- Polls the cache every 500ms and only emits when `version` changed. +- Detects disconnect via `await request.is_disconnected()` and exits the loop. +- Sets `X-Accel-Buffering: no` and `Cache-Control: no-cache` so the stream is + not buffered by an intermediate proxy. + +The 500ms stream cadence is independent of the source cadence. With the +simulator (500ms ticks) the client sees a frame per tick. With Massive (15s +polls) the version is unchanged between polls, so frames are emitted only when +new data actually arrives — no wasted bandwidth. + +--- + +## 10. Implementation Comparison + +| Aspect | `SimulatorDataSource` | `MassiveDataSource` | +|---|---|---| +| Update cadence | 500ms | 15s (configurable) | +| Mechanism | In-process GBM step | REST poll, batched | +| Blocking I/O | None | Yes — wrapped in `asyncio.to_thread` | +| New ticker latency | Immediate (seeded on add) | Next poll cycle | +| Failure mode | Logs and continues | Logs and continues; cache goes stale | +| External dependency | None | Network + API key + paid plan | +| Cost | Free | Snapshot endpoints need Starter+ | + +Both share the same failure philosophy: **the loop never dies**. Exceptions are +caught, logged, and the next cycle proceeds. A market data problem degrades to +stale prices, never to a 500 on the SSE endpoint. + +--- + +## 11. Testing Strategy + +The interface makes each layer independently testable: + +- **`PriceCache`** — pure unit tests: update/get/remove, version increments, + first-update flat semantics, rounding, concurrent writes from threads. +- **`PriceUpdate`** — property math, zero-division guard, `to_dict()` shape. +- **Factory** — `monkeypatch.setenv` across unset/empty/whitespace/valid, + asserting the returned type. No network, no tasks. +- **`SimulatorDataSource`** — start, let a few ticks elapse, assert the cache + populated and prices moved; assert `stop()` halts writes. +- **`MassiveDataSource`** — mock `RESTClient.get_snapshot_all` with realistic + `TickerSnapshot` objects built via `TickerSnapshot.from_dict(raw_json)`. + Building from raw JSON rather than hand-constructing the dataclass is what + catches field-name errors like `sip_timestamp` vs `timestamp`. +- **Consumers** — inject a `PriceCache` pre-populated via `update()`. No data + source needed at all to test portfolio math or trade execution. + +A conformance test parametrized over both implementations asserts they satisfy +the contract identically (idempotent `stop()`, `remove_ticker` clears the +cache, `get_tickers()` reflects mutations). + +--- + +## 12. Extending to a New Provider + +To add, say, an Alpaca or Finnhub feed: + +1. Implement `MarketDataSource` in a new module. +2. Convert that provider's price and timestamp into Unix-seconds floats and + call `cache.update(...)`. +3. Add a branch to `create_market_data_source()`. +4. Add it to the conformance test parametrization. + +No consumer changes. That property is the entire point of the design. diff --git a/planning/MARKET_SIMULATOR.md b/planning/MARKET_SIMULATOR.md new file mode 100644 index 000000000..69d8cad33 --- /dev/null +++ b/planning/MARKET_SIMULATOR.md @@ -0,0 +1,419 @@ +# Market Simulator + +The default market data source for FinAlly. Generates realistic, correlated, +continuously-moving stock prices with no external dependency, no API key, and +no cost. + +Companion documents: `MARKET_INTERFACE.md` (the contract it implements), +`MASSIVE_API.md` (the real-data alternative). + +--- + +## 1. Why Simulate + +The simulator is the **default**, not a fallback. Most people running FinAlly +will never set `MASSIVE_API_KEY`, and the simulator is a better experience for +them: + +| Property | Benefit | +|---|---| +| Always moving | Real markets are closed nights, weekends, and holidays — a demo showing frozen prices looks broken | +| 500ms updates | Matches the UI's flash-animation design; real free-tier data is 15-minute delayed | +| No API key | Zero-setup `docker run` | +| No rate limits | Add 50 tickers, no throttling | +| Deterministic under seed | Reproducible tests | +| Free | Massive's snapshot endpoints require a paid plan | + +The design goal is **plausibility, not prediction**. Prices must look like a +trading terminal: drifting, occasionally jumping, with tech names moving +together. Nobody should mistake this for a forecast. + +--- + +## 2. The Model — Geometric Brownian Motion + +GBM is the standard model for equity price paths (the basis of Black-Scholes). +It has the two properties that matter here: prices stay **strictly positive**, +and **returns** rather than absolute prices are normally distributed, so a $800 +stock and a $15 stock both move in plausible percentage terms. + +The discrete-time update: + +``` +S(t+dt) = S(t) · exp( (μ − σ²/2)·dt + σ·√dt·Z ) + └──────┘ └────────────┘ └───────┘ + current drift diffusion +``` + +| Symbol | Meaning | +|---|---| +| `S(t)` | Current price | +| `μ` | Annualized drift (expected return) | +| `σ` | Annualized volatility | +| `dt` | Time step, as a fraction of a trading year | +| `Z` | Correlated standard normal draw | + +The `−σ²/2` term is the Itô correction. Without it the *median* path drifts +below the intended `μ`, because `E[exp(X)] ≠ exp(E[X])` for a normal `X`. Its +presence is what makes `μ` mean "expected annual return" rather than an +arbitrary tuning knob. + +### Choosing `dt` + +`dt` must be expressed in the same time units as `μ` and `σ`, which are +annualized. A trading year is: + +``` +252 trading days × 6.5 hours/day × 3600 s/hour = 5,896,800 seconds +``` + +So a 500ms tick is: + +```python +TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 +DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ≈ 8.479e-8 +``` + +Using wall-clock seconds per year (31.5M) instead would understate volatility +by roughly 2.3×, making the terminal look sleepy. Anchoring to *trading* time +means one hour of watching FinAlly produces about as much price action as one +hour of watching a real market. + +### Verified magnitudes + +Per-tick standard deviation is `σ·√dt`. Measured against the shipped seed +values: + +| Ticker | σ | Per-tick move | Per-tick $ | Over 1 min | Over 1 hr | +|---|---|---|---|---|---| +| AAPL | 0.22 | 0.0064% | $0.012 | 0.070% | 0.544% | +| TSLA | 0.50 | 0.0146% | $0.036 | 0.159% | 1.235% | +| NVDA | 0.40 | 0.0116% | $0.093 | 0.128% | 0.988% | +| JPM | 0.18 | 0.0052% | $0.010 | 0.057% | 0.445% | +| V | 0.17 | 0.0050% | $0.014 | 0.054% | 0.420% | + +An hour of TSLA moving ~1.2% and JPM ~0.45% is squarely in realistic +territory, and the sub-cent-to-few-cent per-tick moves produce the continuous +flicker the UI wants without prices visibly running away. + +--- + +## 3. Correlation via Cholesky Decomposition + +Independent random walks look wrong. In a real market, tech names move +together — when the sector sells off, the whole watchlist reddens at once. That +collective motion is most of what makes a terminal feel alive. + +### The technique + +To generate correlated normals with a target correlation matrix `C`: + +1. Compute the Cholesky decomposition `C = L·Lᵀ`, where `L` is lower-triangular. +2. Draw a vector `Z` of independent standard normals. +3. `Z_correlated = L · Z` now has correlation matrix exactly `C`. + +```python +z_independent = np.random.standard_normal(n) +z_correlated = self._cholesky @ z_independent +``` + +Each ticker then applies its *own* `σ` to its correlated draw, so tickers share +directional tendency while keeping individual volatility. + +### The correlation structure + +```python +CORRELATION_GROUPS = { + "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 # banks/payments move together +CROSS_GROUP_CORR = 0.3 # broad market beta between sectors +TSLA_CORR = 0.3 # TSLA does its own thing +``` + +Pairwise resolution order: + +```python +if t1 == "TSLA" or t2 == "TSLA": return TSLA_CORR # 0.3 +if t1 in tech and t2 in tech: return INTRA_TECH_CORR # 0.6 +if t1 in finance and t2 in finance: return INTRA_FINANCE_CORR # 0.5 +return CROSS_GROUP_CORR # 0.3 +``` + +TSLA is checked **first**, before the tech-group test. It is a member of the +tech set but is deliberately decorrelated — it is the ticker most likely to be +doing something idiosyncratic, and giving it independence adds visual variety. + +Unknown tickers (anything a user adds) fall through to `CROSS_GROUP_CORR`, +which means a newly added symbol still participates in broad market moves. + +### Verified behaviour + +Building the 10-ticker default matrix and reconstructing `L·Lᵀ`: + +``` +AAPL/MSFT → 0.600 JPM/V → 0.500 AAPL/JPM → 0.300 TSLA/AAPL → 0.300 +minimum eigenvalue → 0.400 (positive definite ✓) +``` + +Empirical correlation of log returns over 20,000 simulated steps: + +| Pair | Target | Measured | +|---|---|---| +| AAPL / MSFT | 0.60 | 0.593 | +| JPM / V | 0.50 | 0.497 | +| AAPL / JPM | 0.30 | 0.298 | +| TSLA / AAPL | 0.30 | 0.295 | + +The model reproduces its target structure to within sampling error. + +> **Positive-definiteness constraint.** `np.linalg.cholesky` raises +> `LinAlgError` on a non-positive-definite matrix. The current values are safe +> (min eigenvalue 0.4), but arbitrary hand-tuned correlations can easily be +> invalid — e.g. A/B = 0.9, A/C = 0.9, B/C = 0.0 is not a realizable +> correlation matrix. Any change to these constants must be checked with +> `np.linalg.eigvalsh(corr).min() > 0`, and there is a unit test asserting the +> matrix builds for the default watchlist. + +--- + +## 4. Random Shock Events + +Pure GBM is smooth. Real markets gap on news. Each tick, each ticker has a +small chance of a discrete jump: + +```python +if random.random() < self._event_prob: # default 0.001 + shock_magnitude = random.uniform(0.02, 0.05) # 2-5% + shock_sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock_magnitude * shock_sign +``` + +Expected frequency with the default watchlist: + +``` +10 tickers × 2 ticks/sec × 0.001 = 0.02 events/sec ≈ one event every 50 seconds +``` + +Roughly one dramatic move per minute across the board — frequent enough that a +user watching for a minute sees something happen, rare enough that it reads as +an event rather than noise. The shock is symmetric (equal up/down probability), +so it adds variance without biasing long-run drift. + +--- + +## 5. Code Structure + +Two classes with a clean split of concerns: + +``` +GBMSimulator ← pure math, fully synchronous, no I/O, no asyncio + ├── step() → advance all tickers one tick, return {ticker: price} + ├── add_ticker() → seed price/params, rebuild Cholesky + ├── remove_ticker() → drop state, rebuild Cholesky + ├── get_price() → current price for one ticker + └── get_tickers() → tracked tickers + +SimulatorDataSource ← implements MarketDataSource, owns the asyncio task + ├── start() → build simulator, seed cache, launch _run_loop + ├── stop() → cancel task + ├── add_ticker() → delegate + seed cache immediately + ├── remove_ticker() → delegate + evict from cache + ├── get_tickers() → delegate + └── _run_loop() → step → write cache → sleep, forever +``` + +**`GBMSimulator` is deliberately free of asyncio and of `PriceCache`.** It is a +deterministic function of its state plus the RNG, which makes the math directly +unit-testable: seed numpy, call `step()` a thousand times, assert on the +distribution of returns. No event loop, no mocking. + +`SimulatorDataSource` handles everything stateful and asynchronous. This split +mirrors `MassiveDataSource`, where the REST client is likewise isolated behind +the async lifecycle. + +### The core loop + +```python +async def _run_loop(self) -> None: + while True: + try: + if self._sim: + prices = self._sim.step() + for ticker, price in prices.items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +The `try`/`except` sits **inside** the loop, so a transient failure (e.g. a +Cholesky error after a bad ticker add) logs and retries on the next tick rather +than silently killing the background task and freezing every price in the app. + +### Startup seeding + +`start()` writes initial prices to the cache **before** returning: + +```python +async def start(self, tickers: list[str]) -> None: + self._sim = GBMSimulator(tickers=tickers, event_probability=self._event_prob) + for ticker in tickers: + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") +``` + +Without this, the first SSE frame after startup would be empty and the +watchlist would render blank for up to 500ms. `add_ticker()` seeds the same way, +so a newly added ticker shows a price instantly rather than after a tick. + +--- + +## 6. Seed Data + +`seed_prices.py` holds all tunable parameters, separate from the logic: + +```python +SEED_PRICES = { + "AAPL": 190.00, "GOOGL": 175.00, "MSFT": 420.00, "AMZN": 185.00, + "TSLA": 250.00, "NVDA": 800.00, "META": 500.00, "JPM": 195.00, + "V": 280.00, "NFLX": 600.00, +} + +TICKER_PARAMS = { + "AAPL": {"sigma": 0.22, "mu": 0.05}, + "GOOGL": {"sigma": 0.25, "mu": 0.05}, + "MSFT": {"sigma": 0.20, "mu": 0.05}, + "AMZN": {"sigma": 0.28, "mu": 0.05}, + "TSLA": {"sigma": 0.50, "mu": 0.03}, # high volatility + "NVDA": {"sigma": 0.40, "mu": 0.08}, # high volatility, strong drift + "META": {"sigma": 0.30, "mu": 0.05}, + "JPM": {"sigma": 0.18, "mu": 0.04}, # low volatility (bank) + "V": {"sigma": 0.17, "mu": 0.04}, # low volatility (payments) + "NFLX": {"sigma": 0.35, "mu": 0.05}, +} + +DEFAULT_PARAMS = {"sigma": 0.25, "mu": 0.05} +``` + +Volatilities are chosen to match each name's real-world character: TSLA the +most volatile at 0.50, NVDA next at 0.40 with the strongest drift, the payment +and banking names calmest at 0.17-0.18. The ordering is what a user notices — +TSLA visibly jumping around while V barely moves is the detail that sells the +simulation. + +### Unknown tickers + +A ticker not in `SEED_PRICES` (anything the user or the LLM adds) gets: + +```python +self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) +self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) +``` + +A random starting price in $50-300 and mid-range parameters. Note +`dict(DEFAULT_PARAMS)` — a **copy**. Sharing the dict would mean per-ticker +parameter tuning silently mutating the defaults for every other unknown ticker. + +--- + +## 7. Known Limitations + +Documented deliberately, since the simulator is the default experience. + +### Low-priced tickers barely move visibly + +Prices are rounded to 2 decimals at the cache boundary. When `σ·√dt·S` is well +under a cent, most ticks round to the same displayed price. Measured fraction +of ticks with **no visible change**: + +| Price level | Flat ticks | Visible move | +|---|---|---| +| $800 (NVDA) | 4.4% | 95.6% | +| $190 (AAPL) | 31.3% | 68.7% | +| $195 (JPM) | 35.3% | 64.7% | +| **$15** | **92.4%** | **7.6%** | + +A user adding a sub-$20 ticker sees a nearly frozen price and almost no flash +animation. The default watchlist is unaffected (cheapest is GOOGL at $175), but +this is a real edge for user-added penny-ish names. + +Mitigation if it matters: scale volatility upward for low-priced tickers when +assigning `DEFAULT_PARAMS`, or set a floor on per-tick movement. Not currently +implemented — the default watchlist doesn't hit it. + +### Other simplifications + +- **No mean reversion.** Prices random-walk; over a long session a ticker can + wander far from its seed. Acceptable for a demo, unrealistic over days. +- **No volume, bid/ask, or order book.** The plan specifies market orders with + instant fill, so none is needed. +- **No market hours.** Prices move 24/7 by design — a frozen weekend terminal + would look broken. +- **No day-open reference.** `PriceUpdate.previous_price` is the previous + *tick*. Day-change percentage (which Massive provides directly as + `todays_change_percent`) has no simulator equivalent; the portfolio layer + computes change against the session's first observed price instead. +- **Correlation is static.** Real correlations spike in a crash. Here they are + fixed constants. + +--- + +## 8. Testing + +The math/async split makes both halves straightforwardly testable. + +**`GBMSimulator` — pure, synchronous:** + +- Prices stay strictly positive across many thousands of steps. +- `step()` returns an entry for every tracked ticker, rounded to 2dp. +- With `np.random.seed(...)` fixed, output is reproducible. +- Empirical volatility of log returns ≈ `σ·√dt` for each ticker. +- Empirical correlation matches the target structure (verified above). +- Cholesky builds and is positive definite for the default watchlist. +- `add_ticker`/`remove_ticker` rebuild the matrix to the right dimension; + adding a duplicate is a no-op; removing an unknown ticker is a no-op. +- With `event_probability=1.0`, every tick applies a 2-5% shock; with `0.0`, + none ever does. +- Unknown tickers get a price in $50-300 and a *copy* of `DEFAULT_PARAMS`. +- Single-ticker and empty-ticker cases don't crash (Cholesky is `None` when + `n <= 1`; `step()` returns `{}` when empty). + +**`SimulatorDataSource` — async integration:** + +- `start()` populates the cache before returning. +- Prices change after letting several ticks elapse. +- `stop()` halts writes — cache version is stable afterward. +- `stop()` is idempotent, including when never started. +- `add_ticker()` seeds the cache immediately; `remove_ticker()` evicts it. +- An exception raised inside `step()` is logged without killing the loop. + +Tests use a short `update_interval` (e.g. 0.01s) so integration cases finish in +milliseconds rather than seconds. + +--- + +## 9. Tuning Guide + +| Goal | Change | +|---|---| +| More dramatic price action | Raise `sigma` in `TICKER_PARAMS` | +| Faster/slower updates | `SimulatorDataSource(update_interval=...)` | +| More/fewer shock events | `event_probability` (default `0.001`) | +| Stronger sector coupling | Raise `INTRA_TECH_CORR` — **re-verify positive definiteness** | +| Different starting prices | `SEED_PRICES` | +| Persistent upward market | Raise `mu` across `TICKER_PARAMS` | + +`update_interval` and `dt` are independent knobs. Changing the interval without +changing `dt` alters how much simulated time passes per real second — halving +the interval to 250ms while leaving `dt` at the 500ms value makes the market +run at half speed. To keep them consistent, derive `dt` from the interval: + +```python +dt = update_interval / TRADING_SECONDS_PER_YEAR +``` diff --git a/planning/MASSIVE_API.md b/planning/MASSIVE_API.md new file mode 100644 index 000000000..86a15b0bd --- /dev/null +++ b/planning/MASSIVE_API.md @@ -0,0 +1,474 @@ +# Massive API Reference (formerly Polygon.io) + +Reference for the Massive REST API as used by FinAlly's market data layer. + +> **Verification note.** Every field name, method signature, and default in this +> document was verified against the installed `massive` Python SDK by +> introspecting the dataclasses and `RESTClient.__init__`, and cross-checked +> against the published REST schemas at . Where the +> raw JSON field names differ from the SDK attribute names, both are given — +> this is the single most common source of bugs when working with this API. + +--- + +## 1. Background + +Polygon.io rebranded to **Massive** on 30 October 2025. + +| Item | Value | +|---|---| +| Base URL | `https://api.massive.com` | +| Legacy base URL | `https://api.polygon.io` (still supported for an extended period) | +| Python package | `massive` (renamed from `polygon-api-client`) | +| Install | `uv add massive` / `pip install -U massive` | +| Minimum Python | 3.9+ | +| Auth | `Authorization: Bearer ` — handled by the SDK | +| API key env var | `MASSIVE_API_KEY` | + +Existing Polygon.io API keys continue to work unchanged. + +--- + +## 2. Rate Limits & Plan Tiers + +| Tier | Price | Rate limit | Data recency | +|---|---|---|---| +| Basic (free) | $0 | **5 requests/minute** | End-of-day / delayed | +| Starter | ~$29/mo | Unlimited | 15-minute delayed | +| Developer | ~$79/mo | Unlimited | Real-time | +| Advanced | ~$199/mo | Unlimited + WebSocket | Real-time | + +"Unlimited" is not literally unbounded — Massive monitors usage and recommends +staying **under 100 requests/second**. Exceeding the free tier returns +**HTTP 429**. + +### What this means for FinAlly + +The free tier's 5 req/min is the binding constraint on our polling design. This +is why we use the **full market snapshot** endpoint: it returns *every requested +ticker in a single HTTP call*, so watchlist size does not affect our request +count. + +| Tier | Recommended poll interval | Requests/min | +|---|---|---| +| Free | 15s | 4 | +| Starter / Developer | 5s | 12 | +| Advanced | 2s | 30 | + +A 15-second poll on the free tier leaves one spare request per minute of +headroom for retries. + +> **Important caveat:** the single-ticker and full-market snapshot endpoints are +> **not included in the free Basic plan** (they require Starter or above). On a +> free key, `get_snapshot_all` will return HTTP 403. FinAlly treats any Massive +> failure as non-fatal and keeps serving the last cached prices, but a free-tier +> key will effectively yield no data. Users without a paid plan should leave +> `MASSIVE_API_KEY` unset and use the simulator. + +--- + +## 3. Client Initialization + +```python +from massive import RESTClient + +# Reads MASSIVE_API_KEY from the environment +client = RESTClient() + +# Or pass the key explicitly (what FinAlly does) +client = RESTClient(api_key="your_key_here") +``` + +Verified constructor defaults: + +```python +RESTClient( + api_key: str | None = None, # falls back to MASSIVE_API_KEY env var + connect_timeout: float = 10.0, + read_timeout: float = 10.0, + num_pools: int = 10, + retries: int = 3, # automatic retry on 5xx + base: str = "https://api.massive.com", + pagination: bool = True, + verbose: bool = False, + trace: bool = False, # trace=True prints request/response +) +``` + +**The client is synchronous.** It uses `urllib3` under the hood and will block +the event loop if called directly from async code. Always wrap calls in +`asyncio.to_thread(...)` — see §7. + +--- + +## 4. Primary Endpoint — Full Market Snapshot + +This is the workhorse endpoint for FinAlly: current prices for many tickers in +one call. + +**REST:** `GET /v2/snapshot/locale/us/markets/stocks/tickers` + +| Query param | Type | Notes | +|---|---|---| +| `tickers` | comma-separated list | Optional. Omit to get the entire market. | +| `include_otc` | boolean | Optional, defaults to `false`. | + +**Python:** + +```python +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +client = RESTClient(api_key=api_key) + +snapshots = client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"], +) + +for snap in snapshots: + print(f"{snap.ticker}: ${snap.last_trade.price}") + print(f" Day change: {snap.todays_change_percent:.2f}%") + print(f" Prev close: ${snap.prev_day.close}") + print(f" Day OHLC: O={snap.day.open} H={snap.day.high} " + f"L={snap.day.low} C={snap.day.close}") +``` + +Verified signature: + +```python +get_snapshot_all( + market_type: str | SnapshotMarketType, + tickers: str | list[str] | None = None, + params: dict | None = None, + raw: bool = False, + include_otc: bool | None = False, + options: RequestOptionBuilder | None = None, +) -> list[TickerSnapshot] | HTTPResponse +``` + +`SnapshotMarketType` members: `STOCKS`, `FOREX`, `CRYPTO`, `INDICES`. + +### 4.1 Raw JSON vs. SDK attribute names + +The API returns terse single-letter JSON keys; the SDK maps them to readable +attribute names. **Use the right-hand column in Python code.** + +Top level (`TickerSnapshot`): + +| Raw JSON | SDK attribute | Meaning | +|---|---|---| +| `ticker` | `.ticker` | Symbol | +| `todaysChange` | `.todays_change` | Absolute change vs. previous close | +| `todaysChangePerc` | `.todays_change_percent` | Percent change vs. previous close | +| `updated` | `.updated` | Last update, Unix **nanoseconds** | +| `day` | `.day` | Today's aggregate bar (`Agg`) | +| `prevDay` | `.prev_day` | Previous day's bar (`Agg`) | +| `min` | `.min` | Most recent minute bar (`MinuteSnapshot`) | +| `lastTrade` | `.last_trade` | Most recent trade (`LastTrade`) | +| `lastQuote` | `.last_quote` | Most recent NBBO quote (`LastQuote`) | +| `fmv` | `.fair_market_value` | Fair market value (Business plans) | + +`Agg` (used by both `.day` and `.prev_day`): + +| Raw | SDK attribute | +|---|---| +| `o` | `.open` | +| `h` | `.high` | +| `l` | `.low` | +| `c` | `.close` | +| `v` | `.volume` | +| `vw` | `.vwap` | +| `t` | `.timestamp` | +| `n` | `.transactions` | + +`LastTrade` (snapshot variant): + +| Raw | SDK attribute | +|---|---| +| `p` | `.price` | +| `s` | `.size` | +| `x` | `.exchange` | +| `t` | **`.sip_timestamp`** | +| `i` | `.id` | +| `c` | `.conditions` | + +`LastQuote`: + +| Raw | SDK attribute | +|---|---| +| `p` | `.bid_price` | +| `s` | `.bid_size` | +| `P` | `.ask_price` | +| `S` | `.ask_size` | +| `t` | `.sip_timestamp` | + +> ### ⚠️ Two field-name traps +> +> **1. `last_trade` has no `.timestamp` attribute.** The raw key `t` maps to +> **`.sip_timestamp`**. Writing `snap.last_trade.timestamp` raises +> `AttributeError`. Full verified field list for the snapshot `LastTrade`: +> `ticker`, `trf_timestamp`, `sequence_number`, `sip_timestamp`, +> `participant_timestamp`, `conditions`, `correction`, `id`, `price`, `trf_id`, +> `size`, `exchange`, `tape`. +> +> **2. There is no `day.previous_close` or `day.change_percent`.** Previous +> close lives at **`snap.prev_day.close`**; percent change at +> **`snap.todays_change_percent`**. + +### 4.2 Timestamp units + +Massive mixes units across fields — this is a frequent source of "prices from +1970" bugs. + +| Field | Unit | Convert to Unix seconds | +|---|---|---| +| `last_trade.sip_timestamp` | **nanoseconds** | `/ 1_000_000_000` | +| `last_quote.sip_timestamp` | **nanoseconds** | `/ 1_000_000_000` | +| `updated` | **nanoseconds** | `/ 1_000_000_000` | +| `day.timestamp`, `prev_day.timestamp` | **milliseconds** | `/ 1_000` | +| Aggregate bar `t` (`list_aggs`, `prev`) | **milliseconds** | `/ 1_000` | + +Rather than hardcode a divisor, prefer a defensive normalizer (see §7). + +--- + +## 5. Other Relevant Endpoints + +### 5.1 Single Ticker Snapshot + +`GET /v2/snapshot/locale/us/markets/stocks/tickers/{stocksTicker}` + +```python +snap = client.get_snapshot_ticker( + market_type=SnapshotMarketType.STOCKS, + ticker="AAPL", +) +print(snap.last_trade.price, snap.todays_change_percent) +``` + +Returns the same `TickerSnapshot` shape as §4. FinAlly does not use this in the +poll loop — batching via `get_snapshot_all` is strictly better for rate limits. + +### 5.2 Previous Day Bar + +`GET /v2/aggs/ticker/{stocksTicker}/prev` + +Useful for establishing seed/reference prices without a snapshot subscription. + +```python +for agg in client.get_previous_close_agg(ticker="AAPL"): + print(f"Prev close: ${agg.close} O={agg.open} H={agg.high} L={agg.low}") +``` + +Query param: `adjusted` (boolean, default `true` — split-adjusted). + +### 5.3 Custom Bars (Aggregates) + +`GET /v2/aggs/ticker/{ticker}/range/{multiplier}/{timespan}/{from}/{to}` + +Historical OHLCV. Not used for live polling, but the natural source if FinAlly +later adds real historical charts (currently sparklines accumulate client-side +from the SSE stream). + +```python +bars = list(client.list_aggs( + ticker="AAPL", + multiplier=1, + timespan="day", # minute | hour | day | week | month | quarter | year + from_="2026-01-01", + to="2026-01-31", + limit=50000, +)) +for b in bars: + print(b.timestamp, b.open, b.high, b.low, b.close, b.volume) +``` + +Pagination is on by default; `limit` controls **page size**, not total results. +Pass `pagination=False` to `RESTClient(...)` for a fixed result count. + +### 5.4 Unified Snapshot (v3) + +`GET /v3/snapshot` — cross-asset-class snapshot with a cleaner schema +(`last_trade.price`, `last_quote.bid`/`.ask`, `session`, `market_status`). + +```python +snaps = list(client.list_universal_snapshots( + market_type="stocks", + ticker_any_of=["AAPL", "GOOGL", "MSFT"], +)) +``` + +Constraint: `ticker.any_of` accepts **at most 250 tickers**; `limit` defaults to +10 and maxes at 250 — remember to raise it or you will silently get 10 results. + +FinAlly uses v2 `get_snapshot_all` because it is the better-documented, more +widely available endpoint for a stocks-only use case, and it has no per-request +ticker cap for our scale. + +### 5.5 Last Trade / Last Quote + +```python +trade = client.get_last_trade(ticker="AAPL") +print(trade.price, trade.size) + +quote = client.get_last_quote(ticker="AAPL") +print(quote.bid_price, quote.ask_price) +``` + +One HTTP call per ticker — avoid in the poll loop. + +--- + +## 6. Error Handling + +| Status | Meaning | FinAlly response | +|---|---|---| +| 401 | Invalid / missing API key | Log error, keep serving cached prices | +| 403 | Plan does not include endpoint | Log error, keep serving cached prices | +| 429 | Rate limit exceeded (free tier) | Back off, retry next interval | +| 5xx | Server error | SDK auto-retries 3× before raising | + +The SDK raises `urllib3`/`requests`-style exceptions and a `BadResponse` on +non-2xx. Because a market data outage must never take down the app, the poller +catches **all** exceptions, logs, and continues — the cache simply holds stale +prices until the next successful poll. + +Individual snapshots may also be partially populated (e.g. `last_trade` is +`None` outside market hours on some plans), so per-ticker parsing is wrapped in +its own try/except and skips rather than aborting the batch. + +--- + +## 7. Reference Implementation — Poll Loop + +This is the pattern FinAlly's `MassiveDataSource` follows. Note the two details +that matter: threading the blocking client off the event loop, and defensive +timestamp normalization. + +```python +import asyncio +import logging +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +logger = logging.getLogger(__name__) + +# Anything larger than this is not plausibly Unix seconds, so scale it down. +_YEAR_2100_SECONDS = 4_102_444_800 + + +def normalize_timestamp(raw: float | None) -> float | None: + """Coerce a Massive timestamp (s / ms / us / ns) to Unix seconds. + + Massive returns nanoseconds for trade/quote SIP timestamps but + milliseconds for aggregate bars. Rather than depend on the field, scale + by orders of magnitude until the value is a plausible epoch-seconds value. + """ + if raw is None: + return None + ts = float(raw) + while ts > _YEAR_2100_SECONDS: + ts /= 1000.0 + return ts + + +def extract_price(snap) -> float | None: + """Best available current price, in priority order.""" + if snap.last_trade and snap.last_trade.price: + return snap.last_trade.price + if snap.min and snap.min.close: # most recent minute bar + return snap.min.close + if snap.day and snap.day.close: # today's close so far + return snap.day.close + if snap.prev_day and snap.prev_day.close: # market closed / pre-open + return snap.prev_day.close + return None + + +async def poll_massive(api_key, get_tickers, price_cache, interval=15.0): + client = RESTClient(api_key=api_key) + + while True: + tickers = get_tickers() + if tickers: + try: + # RESTClient is synchronous — never call it on the event loop. + snapshots = await asyncio.to_thread( + client.get_snapshot_all, + market_type=SnapshotMarketType.STOCKS, + tickers=tickers, + ) + for snap in snapshots: + try: + price = extract_price(snap) + if price is None: + continue + ts = normalize_timestamp( + snap.last_trade.sip_timestamp if snap.last_trade + else snap.updated + ) + price_cache.update( + ticker=snap.ticker, price=price, timestamp=ts + ) + except (AttributeError, TypeError) as e: + logger.warning("Skipping %s: %s", + getattr(snap, "ticker", "???"), e) + except Exception as e: + # Never let a data outage kill the loop. + logger.error("Massive poll failed: %s", e) + + await asyncio.sleep(interval) +``` + +### Why the price fallback chain matters + +`last_trade` is only populated when the plan includes trade data and there has +been a trade. Outside US market hours, or on a delayed plan, relying solely on +`last_trade.price` yields no data at all. The chain +`last_trade → min → day → prev_day` guarantees a usable price whenever the API +returns anything at all — important because FinAlly's portfolio valuation and +trade execution both read from the cache. + +--- + +## 8. Market Hours Behaviour + +- The `day` aggregate resets at market open; during pre-market its values may + still reflect the previous session. +- `last_trade.price` includes extended-hours trades on plans that carry them. +- On weekends and holidays, snapshots return the last session's values and + `todays_change` is typically `0`. +- FinAlly's UI makes no distinction — the price cache is the only contract, and + a stale-but-valid price renders identically to a live one. + +--- + +## 9. Applicability to FinAlly + +| Requirement | Endpoint | Notes | +|---|---|---| +| Live prices for watchlist | `get_snapshot_all` | One call for all tickers | +| Day change % for watchlist | `.todays_change_percent` | Already computed by API | +| Reference / previous close | `.prev_day.close` | No extra call needed | +| Seed prices (simulator parity) | `get_previous_close_agg` | Optional bootstrap | +| Historical charts (future) | `list_aggs` | Not in current scope | + +Everything FinAlly needs in the live path comes from **one endpoint, one call +per poll cycle**. See `MARKET_INTERFACE.md` for how this is wrapped behind the +provider-agnostic interface. + +--- + +## Sources + +- [Massive API docs](https://massive.com/docs) +- [Stocks REST API overview](https://massive.com/docs/rest/stocks/overview) +- [Full market snapshot](https://massive.com/docs/rest/stocks/snapshots/full-market-snapshot.md) +- [Single ticker snapshot](https://massive.com/docs/rest/stocks/snapshots/single-ticker-snapshot.md) +- [Unified snapshot](https://massive.com/docs/rest/stocks/snapshots/unified-snapshot.md) +- [Previous day bar](https://massive.com/docs/rest/stocks/aggregates/previous-day-bar.md) +- [massive-com/client-python](https://github.com/massive-com/client-python) +- [Polygon.io is now Massive](https://massive.com/blog/polygon-is-now-massive) +- [REST request limits](https://massive.com/knowledge-base/article/what-is-the-request-limit-for-massives-restful-apis) +- [Pricing](https://massive.com/pricing) From f72c4f5bbe423c6b068a59b56b843c20ff60ea6a Mon Sep 17 00:00:00 2001 From: hoangnguyentran <49850190+hoangnguyentran@users.noreply.github.com> Date: Sun, 2 Aug 2026 22:27:30 +0800 Subject: [PATCH 2/4] "Update Claude PR Assistant workflow" --- .github/workflows/claude.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/claude.yml b/.github/workflows/claude.yml index d300267f1..6b15fac7a 100644 --- a/.github/workflows/claude.yml +++ b/.github/workflows/claude.yml @@ -46,5 +46,5 @@ jobs: # Optional: Add claude_args to customize behavior and configuration # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md # or https://code.claude.com/docs/en/cli-reference for available options - # claude_args: '--allowed-tools Bash(gh pr:*)' + # claude_args: '--allowed-tools Bash(gh pr *)' From 9ec2212ed6951721b55ab089f0c7bf51c67789f7 Mon Sep 17 00:00:00 2001 From: hoangnguyentran <49850190+hoangnguyentran@users.noreply.github.com> Date: Sun, 2 Aug 2026 22:27:32 +0800 Subject: [PATCH 3/4] "Update Claude Code Review workflow" From 7ba774be1db7cf5560a494967497dd5e52cd6628 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 8 Aug 2026 03:01:34 +0000 Subject: [PATCH 4/4] Add detailed market data backend design document Consolidates the market data design into planning/MARKET_DATA_DESIGN.md: unified MarketDataSource interface, PriceCache, GBM simulator, Massive REST poller, factory selection, SSE endpoint, FastAPI wiring, consumer patterns, frontend integration, testing strategy, and configuration. Every module is given with complete, runnable code snippets plus the rationale behind each design decision. Incorporates the verified Massive SDK field names and timestamp units from MASSIVE_API.md, and the fixes from the earlier code review (top-level imports, public get_tickers, AsyncGenerator annotation, single correlation-fallback constant). Section 20 records six places where the shipped code differs from this design, three of them in the Massive path: last_trade has no .timestamp attribute (verified against the installed SDK), there is no price fallback chain for when last_trade is absent, and snapshot SIP timestamps are nanoseconds rather than milliseconds. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Qmvr6X765aSabbMAvD8vxi --- planning/MARKET_DATA_DESIGN.md | 2136 ++++++++++++++++++++++++++++++++ 1 file changed, 2136 insertions(+) create mode 100644 planning/MARKET_DATA_DESIGN.md diff --git a/planning/MARKET_DATA_DESIGN.md b/planning/MARKET_DATA_DESIGN.md new file mode 100644 index 000000000..99820094b --- /dev/null +++ b/planning/MARKET_DATA_DESIGN.md @@ -0,0 +1,2136 @@ +# Market Data Backend — Detailed Design + +Implementation-ready design for FinAlly's market data subsystem: the unified +provider interface, the shared price cache, the GBM simulator, the Massive +(Polygon.io) REST client, the SSE streaming endpoint, and FastAPI wiring. + +Everything described here lives under `backend/app/market/`. + +Companion documents: +- `MARKET_INTERFACE.md` — the contract, in prose +- `MARKET_SIMULATOR.md` — the simulation model, verified magnitudes +- `MASSIVE_API.md` — the real API, verified field names and units +- `MARKET_DATA_SUMMARY.md` — what shipped, test counts, coverage + +--- + +## Table of Contents + +1. [Design Goals](#1-design-goals) +2. [Architecture](#2-architecture) +3. [File Layout](#3-file-layout) +4. [`models.py` — PriceUpdate](#4-modelspy--priceupdate) +5. [`cache.py` — PriceCache](#5-cachepy--pricecache) +6. [`interface.py` — MarketDataSource](#6-interfacepy--marketdatasource) +7. [`seed_prices.py` — Tunable Constants](#7-seed_pricespy--tunable-constants) +8. [`simulator.py` — GBM Simulator](#8-simulatorpy--gbm-simulator) +9. [`massive_client.py` — Massive REST Poller](#9-massive_clientpy--massive-rest-poller) +10. [`factory.py` — Source Selection](#10-factorypy--source-selection) +11. [`stream.py` — SSE Endpoint](#11-streampy--sse-endpoint) +12. [`__init__.py` — Public Surface](#12-__init__py--public-surface) +13. [FastAPI Wiring](#13-fastapi-wiring) +14. [Consumer Patterns](#14-consumer-patterns) +15. [Frontend Integration](#15-frontend-integration) +16. [Testing Strategy](#16-testing-strategy) +17. [Error Handling & Edge Cases](#17-error-handling--edge-cases) +18. [Configuration Reference](#18-configuration-reference) +19. [Extending to a New Provider](#19-extending-to-a-new-provider) +20. [Deltas From the Current Implementation](#20-deltas-from-the-current-implementation) + +--- + +## 1. Design Goals + +| Goal | How it is achieved | +|---|---| +| Consumers never know where prices come from | Both sources implement one ABC; consumers read only from `PriceCache` | +| Swapping simulator ↔ real data is config, not code | `create_market_data_source()` reads `MASSIVE_API_KEY` | +| A data outage never takes down the app | Every loop catches all exceptions and continues; cache serves stale prices | +| The UI never shows a blank watchlist | `start()` and `add_ticker()` seed the cache before returning | +| The math is unit-testable without an event loop | `GBMSimulator` is pure, synchronous, cache-free | +| No wasted SSE bandwidth | Monotonic `version` counter drives change detection | + +The single most important structural rule: + +> **No consumer ever holds a reference to a `MarketDataSource`.** Sources are +> write-only producers. `PriceCache` is the only read contract. + +The one exception is the watchlist route, which must call `add_ticker` / +`remove_ticker` — that is lifecycle management, not price reading. + +--- + +## 2. Architecture + +``` + ┌──────────────────────────────────────┐ + │ MarketDataSource (ABC) │ + │ start / stop / add_ticker / │ + │ remove_ticker / get_tickers │ + └──────────────────────────────────────┘ + ▲ ▲ + │ │ + ┌───────────────┴─────┐ ┌──────┴──────────────┐ + │ SimulatorDataSource │ │ MassiveDataSource │ + │ GBM step, 500 ms │ │ REST poll, 15 s │ + │ in-process, no I/O │ │ asyncio.to_thread │ + └───────────────┬─────┘ └──────┬──────────────┘ + │ │ + └─── writes ─────┘ + │ + ▼ + ┌───────────────────────────┐ + │ PriceCache │ + │ dict + threading.Lock │ + │ + monotonic version │ + └─────────────┬─────────────┘ + │ reads + ┌───────────────┬───────┴────────┬────────────────┐ + ▼ ▼ ▼ ▼ + GET /api/stream Portfolio Trade LLM context + /prices (SSE) valuation execution builder +``` + +Timing is fully decoupled. The simulator writes every 500 ms; Massive writes +every 15 s; the SSE generator reads every 500 ms and emits only on change. +None of the three knows the others' cadence. + +--- + +## 3. File Layout + +``` +backend/app/market/ +├── __init__.py # Public re-exports +├── models.py # PriceUpdate — the unit of price data +├── cache.py # PriceCache — thread-safe store, versioned +├── interface.py # MarketDataSource — the producer ABC +├── seed_prices.py # SEED_PRICES, TICKER_PARAMS, correlation constants +├── simulator.py # GBMSimulator (pure math) + SimulatorDataSource (async) +├── massive_client.py # MassiveDataSource — REST poller +├── factory.py # create_market_data_source() +└── stream.py # create_stream_router() — SSE endpoint +``` + +Dependency direction is strictly downward — `models` depends on nothing, +`cache` depends on `models`, the sources depend on `cache` + `interface` + +`seed_prices`, `factory` depends on the sources, and nothing depends on +`factory` except application startup. There are no cycles. + +--- + +## 4. `models.py` — PriceUpdate + +The only data structure that leaves the market data layer. + +```python +"""Data models for market data.""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass(frozen=True, slots=True) +class PriceUpdate: + """Immutable snapshot of a single ticker's price at a point in time.""" + + ticker: str + price: float + previous_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: + """Absolute price change from the previous update.""" + return round(self.price - self.previous_price, 4) + + @property + def change_percent(self) -> float: + """Percentage change from the previous update.""" + if self.previous_price == 0: + return 0.0 + return round((self.price - self.previous_price) / self.previous_price * 100, 4) + + @property + def direction(self) -> str: + """'up', 'down', or 'flat'.""" + if self.price > self.previous_price: + return "up" + elif self.price < self.previous_price: + return "down" + return "flat" + + def to_dict(self) -> dict: + """Serialize for JSON / SSE transmission.""" + return { + "ticker": self.ticker, + "price": self.price, + "previous_price": self.previous_price, + "timestamp": self.timestamp, + "change": self.change, + "change_percent": self.change_percent, + "direction": self.direction, + } +``` + +### Design decisions + +- **`frozen=True`** — instances are handed simultaneously to the SSE + serializer and to portfolio math. Nothing may mutate a price after the fact. +- **`slots=True`** — with 10 tickers at 2 Hz we allocate ~20 of these per + second, plus one per cache write. Slots remove the per-instance `__dict__`. +- **Derived values are properties, not fields.** `change`, `change_percent`, + and `direction` cannot drift out of sync with the prices they derive from. + There is no code path that can produce `direction="up"` with a falling price. +- **`previous_price` is the previous *tick*, not the previous *close*.** It + exists to drive the frontend's green/red flash. Day-over-day change is a + separate concern owned by the portfolio layer. +- **`timestamp` is always Unix seconds (float).** Massive's nanosecond + timestamps are normalized at the source boundary (§9), never downstream. +- **Zero-division guard** returns `0.0` rather than raising. A feed glitch must + not propagate an exception into the SSE generator and kill a client stream. + +### Wire format + +`to_dict()` is the single serialization point, used by SSE and by any REST +route that returns a price: + +```json +{ + "ticker": "AAPL", + "price": 190.42, + "previous_price": 190.35, + "timestamp": 1785432000.123, + "change": 0.07, + "change_percent": 0.0368, + "direction": "up" +} +``` + +--- + +## 5. `cache.py` — PriceCache + +The central hub. Producers write, everyone else reads. + +```python +"""Thread-safe in-memory price cache.""" + +from __future__ import annotations + +import time +from threading import Lock + +from .models import PriceUpdate + + +class PriceCache: + """Thread-safe in-memory cache of the latest price for each ticker. + + Writers: SimulatorDataSource or MassiveDataSource (exactly one at a time). + Readers: SSE streaming endpoint, portfolio valuation, trade execution, + LLM context builder. + """ + + def __init__(self) -> None: + self._prices: dict[str, PriceUpdate] = {} + self._lock = Lock() + self._version: int = 0 # Monotonic; bumped on every update() + + def update( + self, + ticker: str, + price: float, + timestamp: float | None = None, + ) -> PriceUpdate: + """Record a new price for a ticker. Returns the created PriceUpdate. + + Computes previous_price from the prior cached value. On the first + update for a ticker, previous_price == price, so direction is 'flat' + and the frontend does not flash on page load. + """ + with self._lock: + ts = time.time() if timestamp is None else timestamp + prev = self._prices.get(ticker) + previous_price = prev.price if prev else price + + update = PriceUpdate( + ticker=ticker, + price=round(price, 2), + previous_price=round(previous_price, 2), + timestamp=ts, + ) + self._prices[ticker] = update + self._version += 1 + return update + + def get(self, ticker: str) -> PriceUpdate | None: + """Latest update for a single ticker, or None if unknown.""" + with self._lock: + return self._prices.get(ticker) + + def get_price(self, ticker: str) -> float | None: + """Convenience: just the price float, or None.""" + update = self.get(ticker) + return update.price if update else None + + def get_all(self) -> dict[str, PriceUpdate]: + """Snapshot of all current prices. Returns a shallow copy.""" + with self._lock: + return dict(self._prices) + + def remove(self, ticker: str) -> None: + """Drop a ticker (e.g. removed from the watchlist).""" + with self._lock: + self._prices.pop(ticker, None) + + @property + def version(self) -> int: + """Monotonic counter, incremented on every update().""" + with self._lock: + return self._version + + def __len__(self) -> int: + with self._lock: + return len(self._prices) + + def __contains__(self, ticker: str) -> bool: + with self._lock: + return ticker in self._prices +``` + +### Why `threading.Lock` and not `asyncio.Lock` + +The Massive client is synchronous and runs inside `asyncio.to_thread(...)`, so +writes genuinely originate on a worker thread while SSE reads happen on the +event loop. An `asyncio.Lock` does not protect that boundary — it only +serializes coroutines on one loop. `threading.Lock` is correct from both sides. + +Contention is negligible: the critical section is a dict lookup plus an +assignment, executed ~20 times/second. + +### `ts = time.time() if timestamp is None else timestamp` + +Written this way, not `timestamp or time.time()`. The `or` form silently +replaces a legitimate `0.0` timestamp with the current time — an epoch-zero +timestamp is exactly what a broken feed produces, and it should be visible in +the data rather than laundered into something plausible. + +### Rounding at the boundary + +Prices are rounded to 2 dp on write, in one place. Every consumer therefore +sees identical values: the UI cannot display $190.42 while a trade fills at +$190.41999999. + +### The version counter + +`version` answers "has anything changed since I last sent?" in O(1), with no +dict diffing and no timestamp comparison: + +```python +last_version = -1 +while True: + if price_cache.version != last_version: + last_version = price_cache.version + yield format_sse(price_cache.get_all()) + await asyncio.sleep(0.5) +``` + +Note it is incremented on *every* `update()`, even when the price is unchanged. +That is intentional — "the source produced a fresh observation" is the event +the stream cares about, not "the number differs". When the source stops +producing (loop cancelled, Massive polls failing), the version freezes and the +stream goes quiet rather than re-sending identical frames. + +The property takes the lock. On CPython an `int` read is atomic under the GIL, +so this is not strictly required today, but it costs nothing and keeps the +class correct on a free-threaded build (PEP 703). + +--- + +## 6. `interface.py` — MarketDataSource + +```python +"""Abstract interface for market data providers.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class MarketDataSource(ABC): + """Contract for market data providers. + + Implementations push price updates into a shared PriceCache on their own + schedule. Downstream code never asks a data source for a price — it reads + from the cache. + + Lifecycle: + source = create_market_data_source(cache) + await source.start(["AAPL", "GOOGL", ...]) + ... + await source.add_ticker("TSLA") + await source.remove_ticker("GOOGL") + ... + await source.stop() + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: + """Begin producing price updates for the given tickers. + + Must populate the cache with initial prices BEFORE returning, then + start a background task. Called exactly once per instance. + """ + + @abstractmethod + async def stop(self) -> None: + """Cancel the background task and release resources. + + Idempotent — safe to call multiple times, including when never + started. After stop(), the source never writes to the cache again. + """ + + @abstractmethod + async def add_ticker(self, ticker: str) -> None: + """Add a ticker to the active set. Idempotent.""" + + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: + """Remove a ticker from the active set. Idempotent. + + Must also remove the ticker from the PriceCache so a de-watchlisted + symbol stops appearing in the SSE payload. + """ + + @abstractmethod + def get_tickers(self) -> list[str]: + """Currently tracked tickers. Synchronous; returns a copy.""" +``` + +### Contract semantics + +| Method | Guarantee | Simulator | Massive | +|---|---|---|---| +| `start(tickers)` | Cache populated before return | Seeds from `SEED_PRICES` | Performs one blocking poll | +| `stop()` | Idempotent, no further writes | Cancels task | Cancels task, drops client | +| `add_ticker(t)` | Idempotent | Price available immediately | Appears next poll cycle | +| `remove_ticker(t)` | Idempotent, evicts from cache | Rebuilds Cholesky | Filters ticker list | +| `get_tickers()` | Sync, returns a copy | Delegates to `GBMSimulator` | Copies internal list | + +### Why this shape + +- **Async lifecycle, sync accessor.** `start`/`stop`/`add`/`remove` are async + because Massive performs network I/O and both manage asyncio tasks. + `get_tickers()` is a pure in-memory read; forcing `await` on it would be + noise at every call site. +- **No `get_price()` on the interface — deliberately.** If sources exposed + price reads, consumers would couple to the source and the cache would become + an implementation detail rather than the contract. Prices come from + `PriceCache`, always. +- **Push, not pull.** Sources write on their own schedule. A consumer can never + trigger a fetch, so a slow API can never block an HTTP request. + +--- + +## 7. `seed_prices.py` — Tunable Constants + +Constants only. No logic, no imports. Everything a person would want to tune +lives here rather than scattered through the simulator. + +```python +"""Seed prices and per-ticker parameters for the market simulator.""" + +# Realistic starting prices for the default watchlist +SEED_PRICES: dict[str, float] = { + "AAPL": 190.00, + "GOOGL": 175.00, + "MSFT": 420.00, + "AMZN": 185.00, + "TSLA": 250.00, + "NVDA": 800.00, + "META": 500.00, + "JPM": 195.00, + "V": 280.00, + "NFLX": 600.00, +} + +# Per-ticker GBM parameters +# sigma: annualized volatility (higher = more movement) +# mu: annualized drift / expected return +TICKER_PARAMS: dict[str, dict[str, float]] = { + "AAPL": {"sigma": 0.22, "mu": 0.05}, + "GOOGL": {"sigma": 0.25, "mu": 0.05}, + "MSFT": {"sigma": 0.20, "mu": 0.05}, + "AMZN": {"sigma": 0.28, "mu": 0.05}, + "TSLA": {"sigma": 0.50, "mu": 0.03}, # High volatility + "NVDA": {"sigma": 0.40, "mu": 0.08}, # High volatility, strong drift + "META": {"sigma": 0.30, "mu": 0.05}, + "JPM": {"sigma": 0.18, "mu": 0.04}, # Low volatility (bank) + "V": {"sigma": 0.17, "mu": 0.04}, # Low volatility (payments) + "NFLX": {"sigma": 0.35, "mu": 0.05}, +} + +# Applied to any ticker the user or the LLM adds +DEFAULT_PARAMS: dict[str, float] = {"sigma": 0.25, "mu": 0.05} + +# Correlation groups for the Cholesky decomposition +CORRELATION_GROUPS: dict[str, set[str]] = { + "tech": {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"}, + "finance": {"JPM", "V"}, +} + +# Correlation coefficients +INTRA_TECH_CORR = 0.6 # Tech names move together +INTRA_FINANCE_CORR = 0.5 # Bank / payments move together +CROSS_GROUP_CORR = 0.3 # Broad market beta; also the unknown-ticker default +TSLA_CORR = 0.3 # TSLA does its own thing +``` + +Volatilities are chosen to match each name's real-world character. The +*ordering* is what a user actually notices — TSLA visibly jumping while V +barely moves is the detail that sells the simulation. + +There is deliberately **no separate `DEFAULT_CORR`**. An earlier revision +defined one alongside `CROSS_GROUP_CORR` with the same value 0.3; only +`CROSS_GROUP_CORR` was ever referenced. One constant, one meaning: "any pair +we have no specific rule for." + +--- + +## 8. `simulator.py` — GBM Simulator + +Two classes with a hard split: + +``` +GBMSimulator ← pure math, synchronous, no asyncio, no PriceCache + ├── step() → advance all tickers one tick, return {ticker: price} + ├── add_ticker() → seed price/params, rebuild Cholesky + ├── remove_ticker() → drop state, rebuild Cholesky + ├── get_price() → current price for one ticker + └── get_tickers() → tracked tickers + +SimulatorDataSource ← implements MarketDataSource, owns the asyncio task + ├── start() → build simulator, seed cache, launch _run_loop + ├── stop() → cancel task + ├── add_ticker() → delegate + seed cache immediately + ├── remove_ticker() → delegate + evict from cache + ├── get_tickers() → delegate + └── _run_loop() → step → write cache → sleep, forever +``` + +`GBMSimulator` being free of asyncio and of `PriceCache` is what makes the math +directly testable: seed numpy, call `step()` ten thousand times, assert on the +distribution of log returns. No event loop, no mocks. + +### 8.1 The model + +``` +S(t+dt) = S(t) · exp( (μ − σ²/2)·dt + σ·√dt·Z ) + └──────┘ └────────────┘ └───────┘ + current drift diffusion +``` + +GBM is the standard equity price model (the basis of Black–Scholes) and has the +two properties that matter here: prices stay **strictly positive**, and +*returns* rather than absolute prices are normally distributed — so an $800 +stock and a $15 stock both move in plausible percentage terms. + +The `−σ²/2` term is the Itô correction. Without it the median path drifts below +the intended `μ`, because `E[exp(X)] ≠ exp(E[X])` for normal `X`. Its presence +is what makes `μ` mean "expected annual return" rather than an arbitrary knob. + +**Choosing `dt`.** `dt` must be in the same units as `μ` and `σ`, which are +annualized. A *trading* year is: + +``` +252 trading days × 6.5 hours/day × 3600 s/hour = 5,896,800 seconds +``` + +so a 500 ms tick is `0.5 / 5_896_800 ≈ 8.479e-8`. Using wall-clock seconds per +year (31.5 M) instead would understate volatility by ~2.3×, making the terminal +look sleepy. Anchoring to trading time means an hour of watching FinAlly +produces about as much price action as an hour of watching a real market. + +Resulting per-tick standard deviation (`σ·√dt`), measured against the shipped +seeds: + +| Ticker | σ | Per-tick move | Per-tick $ | Over 1 min | Over 1 hr | +|---|---|---|---|---|---| +| AAPL | 0.22 | 0.0064% | $0.012 | 0.070% | 0.544% | +| TSLA | 0.50 | 0.0146% | $0.036 | 0.159% | 1.235% | +| NVDA | 0.40 | 0.0116% | $0.093 | 0.128% | 0.988% | +| JPM | 0.18 | 0.0052% | $0.010 | 0.057% | 0.445% | +| V | 0.17 | 0.0050% | $0.014 | 0.054% | 0.420% | + +### 8.2 Correlation via Cholesky decomposition + +Independent random walks look wrong. In a real market tech names move together +— when the sector sells off the whole watchlist reddens at once, and that +collective motion is most of what makes a terminal feel alive. + +To generate correlated normals with target correlation matrix `C`: + +1. Compute `C = L·Lᵀ` (Cholesky), `L` lower-triangular. +2. Draw independent standard normals `Z`. +3. `L·Z` has correlation matrix exactly `C`. + +Each ticker then applies its own `σ` to its correlated draw, so tickers share +directional tendency while keeping individual volatility. + +Pairwise resolution order: + +```python +if t1 == "TSLA" or t2 == "TSLA": return TSLA_CORR # 0.3 +if t1 in tech and t2 in tech: return INTRA_TECH_CORR # 0.6 +if t1 in finance and t2 in finance: return INTRA_FINANCE_CORR # 0.5 +return CROSS_GROUP_CORR # 0.3 +``` + +TSLA is checked **first**, before the tech test. It is a member of the tech set +but deliberately decorrelated — it is the ticker most likely to be doing +something idiosyncratic, and giving it independence adds visual variety. + +Verified on the default 10-ticker matrix: `AAPL/MSFT → 0.600`, `JPM/V → 0.500`, +`AAPL/JPM → 0.300`, `TSLA/AAPL → 0.300`, minimum eigenvalue `0.400` +(positive definite ✓). Empirical correlation of log returns over 20,000 +simulated steps reproduces the targets to within sampling error (0.593, 0.497, +0.298, 0.295). + +> **Positive-definiteness constraint.** `np.linalg.cholesky` raises +> `LinAlgError` on a non-positive-definite matrix, and hand-tuned correlations +> are easy to get wrong — A/B = 0.9, A/C = 0.9, B/C = 0.0 is not a realizable +> correlation matrix. Any change to these constants must be checked with +> `np.linalg.eigvalsh(corr).min() > 0`, and there is a unit test asserting the +> matrix builds for the default watchlist. + +### 8.3 `GBMSimulator` + +```python +"""GBM-based market simulator.""" + +from __future__ import annotations + +import asyncio +import logging +import math +import random + +import numpy as np + +from .cache import PriceCache +from .interface import MarketDataSource +from .seed_prices import ( + CORRELATION_GROUPS, + CROSS_GROUP_CORR, + DEFAULT_PARAMS, + INTRA_FINANCE_CORR, + INTRA_TECH_CORR, + SEED_PRICES, + TICKER_PARAMS, + TSLA_CORR, +) + +logger = logging.getLogger(__name__) + + +class GBMSimulator: + """Geometric Brownian Motion simulator for correlated stock prices. + + S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) + + where Z is a Cholesky-correlated standard normal draw. The tiny dt + (~8.5e-8 for 500ms ticks) produces sub-cent moves per tick that + accumulate naturally over a session. + """ + + # 252 trading days * 6.5 hours/day * 3600 seconds/hour + TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 + DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.479e-8 + + def __init__( + self, + tickers: list[str], + dt: float = DEFAULT_DT, + event_probability: float = 0.001, + ) -> None: + self._dt = dt + self._event_prob = event_probability + + self._tickers: list[str] = [] + self._prices: dict[str, float] = {} + self._params: dict[str, dict[str, float]] = {} + self._cholesky: np.ndarray | None = None + + for ticker in tickers: + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + # --- Public API --- + + def step(self) -> dict[str, float]: + """Advance all tickers one time step. Returns {ticker: new_price}. + + Hot path — called every 500ms. Keep it fast. + """ + n = len(self._tickers) + if n == 0: + return {} + + z_independent = np.random.standard_normal(n) + if self._cholesky is not None: + z_correlated = self._cholesky @ z_independent + else: + z_correlated = z_independent # n == 1: nothing to correlate with + + result: dict[str, float] = {} + for i, ticker in enumerate(self._tickers): + params = self._params[ticker] + mu = params["mu"] + sigma = params["sigma"] + + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z_correlated[i] + self._prices[ticker] *= math.exp(drift + diffusion) + + # Random shock: ~0.1% chance per tick per ticker. + # 10 tickers * 2 ticks/sec * 0.001 -> one event every ~50 seconds. + if random.random() < self._event_prob: + shock_magnitude = random.uniform(0.02, 0.05) + shock_sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock_magnitude * shock_sign + logger.debug( + "Random event on %s: %.1f%% %s", + ticker, + shock_magnitude * 100, + "up" if shock_sign > 0 else "down", + ) + + result[ticker] = round(self._prices[ticker], 2) + + return result + + def add_ticker(self, ticker: str) -> None: + """Add a ticker. Rebuilds the correlation matrix. Idempotent.""" + if ticker in self._prices: + return + self._add_ticker_internal(ticker) + self._rebuild_cholesky() + + def remove_ticker(self, ticker: str) -> None: + """Remove a ticker. Rebuilds the correlation matrix. Idempotent.""" + if ticker not in self._prices: + return + self._tickers.remove(ticker) + del self._prices[ticker] + del self._params[ticker] + self._rebuild_cholesky() + + def get_price(self, ticker: str) -> float | None: + """Current price for a ticker, or None if not tracked.""" + return self._prices.get(ticker) + + def get_tickers(self) -> list[str]: + """Currently tracked tickers (a copy).""" + return list(self._tickers) + + # --- Internals --- + + def _add_ticker_internal(self, ticker: str) -> None: + """Add without rebuilding Cholesky (used for batch initialization).""" + if ticker in self._prices: + return + self._tickers.append(ticker) + self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) + # dict(...) is a COPY — sharing DEFAULT_PARAMS would let per-ticker + # tuning silently mutate the defaults for every other unknown ticker. + self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) + + def _rebuild_cholesky(self) -> None: + """Rebuild the Cholesky factor of the correlation matrix. + + Called on every add/remove. O(n^2) to build, O(n^3) to factor, but + n < 50 and it happens only on watchlist mutations. + """ + 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] = rho + corr[j, i] = rho + + self._cholesky = np.linalg.cholesky(corr) + + @staticmethod + def _pairwise_correlation(t1: str, t2: str) -> float: + """Correlation between two tickers, by sector grouping. + + - TSLA with anything: 0.3 (checked first; it does its own thing) + - Both tech: 0.6 + - Both finance: 0.5 + - Anything else: 0.3 + """ + tech = CORRELATION_GROUPS["tech"] + finance = CORRELATION_GROUPS["finance"] + + if t1 == "TSLA" or t2 == "TSLA": + return TSLA_CORR + if t1 in tech and t2 in tech: + return INTRA_TECH_CORR + if t1 in finance and t2 in finance: + return INTRA_FINANCE_CORR + return CROSS_GROUP_CORR +``` + +### 8.4 `SimulatorDataSource` + +```python +class SimulatorDataSource(MarketDataSource): + """MarketDataSource backed by the GBM simulator. + + Runs a background asyncio task calling GBMSimulator.step() every + `update_interval` seconds and writing the results to the PriceCache. + """ + + def __init__( + self, + price_cache: PriceCache, + update_interval: float = 0.5, + event_probability: float = 0.001, + ) -> None: + self._cache = price_cache + self._interval = update_interval + self._event_prob = event_probability + self._sim: GBMSimulator | None = None + self._task: asyncio.Task | None = None + + async def start(self, tickers: list[str]) -> None: + self._sim = GBMSimulator( + tickers=tickers, + event_probability=self._event_prob, + ) + # Seed the cache BEFORE returning so the first SSE frame has data. + for ticker in tickers: + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + logger.info("Simulator started with %d tickers", len(tickers)) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("Simulator stopped") + + async def add_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.add_ticker(ticker) + # Seed immediately so the ticker shows a price without waiting a tick. + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + logger.info("Simulator: added ticker %s", ticker) + + async def remove_ticker(self, ticker: str) -> None: + if self._sim: + self._sim.remove_ticker(ticker) + self._cache.remove(ticker) + logger.info("Simulator: removed ticker %s", ticker) + + def get_tickers(self) -> list[str]: + return self._sim.get_tickers() if self._sim else [] + + async def _run_loop(self) -> None: + """Core loop: step the simulation, write to cache, sleep.""" + while True: + try: + if self._sim: + prices = self._sim.step() + for ticker, price in prices.items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") + await asyncio.sleep(self._interval) +``` + +Three details carry weight: + +- **`try`/`except` is *inside* the `while`.** A transient failure (say a + Cholesky error after a pathological ticker add) logs and retries next tick + rather than silently killing the task and freezing every price in the app. + An exception escaping `_run_loop` would be swallowed by the Task and produce + a UI that looks connected but never updates — the worst failure mode + available. +- **`stop()` is idempotent**, including when never started: `self._task` is + `None`, both guards short-circuit. FastAPI's lifespan teardown can call it + unconditionally. +- **`get_tickers()` delegates** to `GBMSimulator.get_tickers()` rather than + reaching into `_tickers`. The private-attribute access was flagged in review; + the public accessor keeps the boundary clean and returns a copy either way. + +### 8.5 Known limitations + +Documented deliberately, since the simulator is the default experience. + +**Low-priced tickers barely move visibly.** Prices round to 2 dp at the cache +boundary, so when `σ·√dt·S` is well under a cent most ticks round to the same +displayed price: + +| Price level | Flat ticks | Visible move | +|---|---|---| +| $800 (NVDA) | 4.4% | 95.6% | +| $190 (AAPL) | 31.3% | 68.7% | +| $195 (JPM) | 35.3% | 64.7% | +| **$15** | **92.4%** | **7.6%** | + +The default watchlist is unaffected (cheapest is GOOGL at $175), but a +user-added sub-$20 name looks nearly frozen. Mitigation if it ever matters: +scale `DEFAULT_PARAMS["sigma"]` up for low-priced tickers, or floor the +per-tick move. Not implemented — the default watchlist does not hit it. + +Other simplifications: no mean reversion (prices random-walk and can wander far +over a long session); no volume, bid/ask, or order book (market orders with +instant fill need none); no market hours (a frozen weekend terminal would look +broken); no day-open reference (day-change % is computed by the portfolio layer +against the session's first observed price); static correlations. + +--- + +## 9. `massive_client.py` — Massive REST Poller + +Polls the full-market snapshot endpoint for the union of watched tickers in a +**single HTTP call per cycle**, so watchlist size never affects request count — +which is what makes the free tier's 5 req/min survivable. + +| Tier | Poll interval | Requests/min | +|---|---|---| +| Free (Basic) | 15 s | 4 | +| Starter / Developer | 5 s | 12 | +| Advanced | 2 s | 30 | + +> The snapshot endpoints are **not included in the free Basic plan** — a free +> key returns HTTP 403. FinAlly treats that as non-fatal and keeps serving +> cached prices, but users without a paid plan should leave `MASSIVE_API_KEY` +> unset and use the simulator. + +### 9.1 Two boundary helpers + +Both exist because of verified traps in the Massive SDK (see `MASSIVE_API.md` +§4.1–4.2). They are module-level functions, not methods, so they can be unit +tested without constructing a source. + +```python +"""Massive (Polygon.io) API client for real market data.""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any + +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +from .cache import PriceCache +from .interface import MarketDataSource + +logger = logging.getLogger(__name__) + +# Anything larger than this is not plausibly Unix seconds, so scale it down. +_YEAR_2100_SECONDS = 4_102_444_800 + + +def normalize_timestamp(raw: float | None) -> float | None: + """Coerce a Massive timestamp (s / ms / us / ns) to Unix seconds. + + Massive returns NANOSECONDS for trade/quote SIP timestamps and the + top-level `updated` field, but MILLISECONDS for aggregate bars. Rather + than hardcode a divisor per field, scale by orders of magnitude until the + value is a plausible epoch-seconds value. This is the fix for the classic + "prices from 1970 / prices from year 55000" bug. + """ + if raw is None: + return None + ts = float(raw) + while ts > _YEAR_2100_SECONDS: + ts /= 1000.0 + return ts + + +def extract_price(snap: Any) -> float | None: + """Best available current price, in priority order. + + last_trade is only populated when the plan carries trade data AND a trade + has occurred. Outside US market hours, or on a delayed plan, relying on it + alone yields no data at all. This chain guarantees a usable price whenever + the API returns anything. + """ + if snap.last_trade and snap.last_trade.price: + return snap.last_trade.price + if snap.min and snap.min.close: # most recent minute bar + return snap.min.close + if snap.day and snap.day.close: # today's close so far + return snap.day.close + if snap.prev_day and snap.prev_day.close: # market closed / pre-open + return snap.prev_day.close + return None + + +def extract_timestamp(snap: Any) -> float | None: + """Unix-seconds timestamp for a snapshot. + + NOTE: the snapshot LastTrade has NO `.timestamp` attribute. The raw JSON + key `t` maps to `.sip_timestamp`; writing `snap.last_trade.timestamp` + raises AttributeError and silently drops every ticker in the batch. + """ + if snap.last_trade and snap.last_trade.sip_timestamp: + return normalize_timestamp(snap.last_trade.sip_timestamp) + return normalize_timestamp(getattr(snap, "updated", None)) +``` + +### 9.2 `MassiveDataSource` + +```python +class MassiveDataSource(MarketDataSource): + """MarketDataSource backed by the Massive (Polygon.io) REST API. + + Polls GET /v2/snapshot/locale/us/markets/stocks/tickers for all watched + tickers in one call, then writes the results to the PriceCache. + + Rate limits: + - Free tier: 5 req/min -> poll every 15s (default) + - Paid tiers: higher limits -> poll every 2-5s + """ + + def __init__( + self, + api_key: str, + price_cache: PriceCache, + poll_interval: float = 15.0, + ) -> None: + self._api_key = api_key + self._cache = price_cache + self._interval = poll_interval + self._tickers: list[str] = [] + self._task: asyncio.Task | None = None + self._client: RESTClient | None = None + + async def start(self, tickers: list[str]) -> None: + self._client = RESTClient(api_key=self._api_key) + self._tickers = [t.upper().strip() for t in tickers] + + # Immediate first poll so the cache has data before we return. + await self._poll_once() + + self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") + logger.info( + "Massive poller started: %d tickers, %.1fs interval", + len(self._tickers), + self._interval, + ) + + async def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + self._client = None + logger.info("Massive poller stopped") + + async def add_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + if ticker not in self._tickers: + self._tickers.append(ticker) + logger.info("Massive: added ticker %s (appears on next poll)", ticker) + + async def remove_ticker(self, ticker: str) -> None: + ticker = ticker.upper().strip() + self._tickers = [t for t in self._tickers if t != ticker] + self._cache.remove(ticker) + logger.info("Massive: removed ticker %s", ticker) + + def get_tickers(self) -> list[str]: + return list(self._tickers) + + # --- Internal --- + + async def _poll_loop(self) -> None: + """Poll on interval. The first poll already happened in start().""" + while True: + await asyncio.sleep(self._interval) + await self._poll_once() + + async def _poll_once(self) -> None: + """One poll cycle: fetch snapshots, write to cache.""" + if not self._tickers or not self._client: + return + + try: + # RESTClient is SYNCHRONOUS (urllib3). Calling it directly on the + # event loop would stall every SSE stream for the request duration. + snapshots = await asyncio.to_thread(self._fetch_snapshots) + + processed = 0 + for snap in snapshots: + try: + price = extract_price(snap) + if price is None: + continue + self._cache.update( + ticker=snap.ticker, + price=price, + timestamp=extract_timestamp(snap), + ) + processed += 1 + except (AttributeError, TypeError) as e: + # Per-ticker guard: one malformed snapshot must not abort + # the whole batch. + logger.warning( + "Skipping snapshot for %s: %s", + getattr(snap, "ticker", "???"), + e, + ) + logger.debug( + "Massive poll: updated %d/%d tickers", processed, len(self._tickers) + ) + + except Exception as e: + # Never let a data outage kill the loop. Common: 401 (bad key), + # 403 (plan lacks snapshots), 429 (rate limit), network timeout. + logger.error("Massive poll failed: %s", e) + + def _fetch_snapshots(self) -> list: + """Blocking call to the Massive REST API. Runs in a worker thread.""" + return self._client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=self._tickers, + ) +``` + +### 9.3 Error handling philosophy + +| Failure | Behaviour | +|---|---| +| 401 invalid key | Logged as error; poller keeps running (fix `.env`, restart) | +| 403 plan lacks snapshots | Logged as error; cache stays empty; user should unset the key | +| 429 rate limited | Logged; next cycle retries after `poll_interval` | +| Network timeout / 5xx | SDK auto-retries 3×, then logged; retried next cycle | +| Malformed single snapshot | That ticker skipped with a warning; batch continues | +| Every ticker fails | Cache retains last-known prices; SSE keeps streaming stale data | + +The rule shared with the simulator: **the loop never dies.** A market data +problem degrades to stale prices, never to a 500 on the SSE endpoint. + +### 9.4 Why imports are at module level + +`from massive import RESTClient` sits at the top of the file, not inside +`start()`. `massive` is a declared core dependency in `pyproject.toml`, so it +is always installed; a lazy import bought nothing and actively broke tests — +`patch("app.market.massive_client.RESTClient")` cannot patch a name that does +not exist at module level until `start()` runs. Top-level imports make the +mock target real and the failure mode (missing dependency) loud at import time +instead of at first poll. + +--- + +## 10. `factory.py` — Source Selection + +```python +"""Factory for creating market data sources.""" + +from __future__ import annotations + +import logging +import os + +from .cache import PriceCache +from .interface import MarketDataSource +from .massive_client import MassiveDataSource +from .simulator import SimulatorDataSource + +logger = logging.getLogger(__name__) + + +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """Select a market data source from the environment. + + - MASSIVE_API_KEY set and non-empty -> MassiveDataSource (real data) + - Otherwise -> SimulatorDataSource (GBM) + + Returns an UNSTARTED source. The caller owns the lifecycle and must + await source.start(tickers). + """ + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + + if api_key: + logger.info("Market data source: Massive API (real data)") + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + + logger.info("Market data source: GBM Simulator") + return SimulatorDataSource(price_cache=price_cache) +``` + +| `MASSIVE_API_KEY` | Result | +|---|---| +| Unset | Simulator | +| Empty string | Simulator | +| Whitespace only | Simulator (`.strip()`) | +| Non-empty | Massive | + +The `.strip()` matters in practice: a `.env` line reading `MASSIVE_API_KEY= ` +with a trailing space would otherwise select the real API with a garbage key, +and the app would silently show no prices at all. + +Returning an **unstarted** source keeps the factory synchronous and makes +selection logic trivially unit-testable — `monkeypatch.setenv`, call, assert on +the type, no event loop and no background tasks. + +--- + +## 11. `stream.py` — SSE Endpoint + +```python +"""SSE streaming endpoint for live price updates.""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import AsyncGenerator + +from fastapi import APIRouter, Request +from fastapi.responses import StreamingResponse + +from .cache import PriceCache + +logger = logging.getLogger(__name__) + + +def create_stream_router( + price_cache: PriceCache, + interval: float = 0.5, +) -> APIRouter: + """Build the SSE router bound to a specific PriceCache. + + The router is created INSIDE the factory, not at module level: a + module-level router would accumulate a duplicate /prices route on every + call, which bites as soon as two tests each build their own app. + """ + router = APIRouter(prefix="/api/stream", tags=["streaming"]) + + @router.get("/prices") + async def stream_prices(request: Request) -> StreamingResponse: + """SSE stream of live prices. + + Clients connect with the native EventSource API and receive frames of + the form: + + data: {"AAPL": {"ticker": "AAPL", "price": 190.50, ...}, ...} + """ + return StreamingResponse( + _generate_events(price_cache, request, interval), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", # Defeat nginx response buffering + }, + ) + + return router + + +async def _generate_events( + price_cache: PriceCache, + request: Request, + interval: float = 0.5, +) -> AsyncGenerator[str, None]: + """Yield SSE frames, one per cache version change. + + Runs until the client disconnects. Note the return annotation: this is an + async generator, so `-> None` would be a lie to type checkers. + """ + # Browser EventSource reconnects after this many ms if the stream drops. + yield "retry: 1000\n\n" + + last_version = -1 + client_ip = request.client.host if request.client else "unknown" + logger.info("SSE client connected: %s", client_ip) + + try: + while True: + if await request.is_disconnected(): + logger.info("SSE client disconnected: %s", client_ip) + break + + current_version = price_cache.version + if current_version != last_version: + last_version = current_version + prices = price_cache.get_all() + if prices: + data = {t: u.to_dict() for t, u in prices.items()} + yield f"data: {json.dumps(data)}\n\n" + + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("SSE stream cancelled for: %s", client_ip) +``` + +### Wire format + +``` +retry: 1000 + +data: {"AAPL":{"ticker":"AAPL","price":190.50,"previous_price":190.42,"timestamp":1785432000.5,"change":0.08,"change_percent":0.042,"direction":"up"},"GOOGL":{...}} + +data: {...} +``` + +### Design notes + +- **All tickers in every frame, keyed by ticker.** Simpler for the client than + per-ticker events, and at ~10 tickers the payload is ~1 KB — trivial. The + frontend does a single `Object.entries()` pass per frame. +- **Poll-and-push rather than event-driven.** The generator reads the cache on + a fixed cadence instead of being woken by the producer. This gives evenly + spaced frames, which matters because the frontend accumulates them into + sparklines — irregular spacing would produce visibly lumpy charts. +- **Stream cadence is independent of source cadence.** With the simulator + (500 ms) the client sees roughly one frame per tick. With Massive (15 s) the + version is unchanged between polls, so frames are emitted only when new data + actually lands. No wasted bandwidth, no configuration coupling. +- **Disconnect detection** via `await request.is_disconnected()` at the top of + each iteration; without it, a closed tab leaves the generator running until + the next write fails. + +### Optional: idle keepalive + +If the source stops producing (Massive polls failing, simulator stopped) the +stream goes completely silent, and an intermediate proxy may reap the +connection as idle. A comment frame is a cheap insurance policy — SSE comment +lines start with `:` and are ignored by `EventSource`: + +```python +last_emit = time.monotonic() +... + if current_version != last_version: + ... + last_emit = time.monotonic() + elif time.monotonic() - last_emit > 15.0: + yield ": keepalive\n\n" + last_emit = time.monotonic() +``` + +Not required for the single-container localhost deployment, which has no proxy +in the path. Add it if FinAlly is ever put behind nginx, App Runner, or a CDN. + +--- + +## 12. `__init__.py` — Public Surface + +```python +"""Market data subsystem for FinAlly. + +Public API: + PriceUpdate - Immutable price snapshot dataclass + PriceCache - Thread-safe in-memory price store + MarketDataSource - Abstract interface for data providers + create_market_data_source - Factory selecting simulator or Massive + create_stream_router - FastAPI router factory for the SSE endpoint +""" + +from .cache import PriceCache +from .factory import create_market_data_source +from .interface import MarketDataSource +from .models import PriceUpdate +from .stream import create_stream_router + +__all__ = [ + "PriceUpdate", + "PriceCache", + "MarketDataSource", + "create_market_data_source", + "create_stream_router", +] +``` + +The rest of the backend imports from `app.market`, never from +`app.market.simulator` or `app.market.massive_client`. Concrete source classes +are intentionally *not* exported — importing one by name is exactly the +coupling this design exists to prevent. Tests may import them directly; that is +the only legitimate use. + +--- + +## 13. FastAPI Wiring + +```python +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from app.db import load_watchlist_tickers +from app.market import ( + MarketDataSource, + PriceCache, + create_market_data_source, + create_stream_router, +) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + # --- startup --- + cache = PriceCache() + source = create_market_data_source(cache) + + tickers = load_watchlist_tickers() # from SQLite; seeds on first run + await source.start(tickers) # cache is populated before this returns + + app.state.price_cache = cache + app.state.market_source = source + + try: + yield + finally: + # --- shutdown --- (finally: runs even if startup of a later + # component raised, so we never leak the background task) + await source.stop() + + +app = FastAPI(title="FinAlly", lifespan=lifespan) +``` + +### Router registration + +The SSE router needs the cache, which is created inside `lifespan`. Two +workable orders: + +**A. Create the cache at module scope** (simplest, and what the routes below +assume): + +```python +price_cache = PriceCache() +app = FastAPI(title="FinAlly", lifespan=lifespan) +app.include_router(create_stream_router(price_cache)) +``` + +with `lifespan` reusing that instance rather than constructing its own. + +**B. Include the router inside `lifespan`** before `yield`. FastAPI permits +this, but routes added after startup do not appear in the OpenAPI schema +generated at first request in some versions, so prefer A. + +Either way, **do not use module-level globals for the cache in tests** — each +test app should get its own `PriceCache` so state cannot leak between tests. + +### Dependency accessors + +```python +from fastapi import Depends, HTTPException, Request + + +def get_price_cache(request: Request) -> PriceCache: + return request.app.state.price_cache + + +def get_market_source(request: Request) -> MarketDataSource: + return request.app.state.market_source +``` + +Taking `Request` rather than closing over a module global is what keeps these +usable from a test app built with a different cache. + +### Watchlist coordination + +Watchlist mutations must update the database **and** the data source. Ticker +normalization happens at the API boundary and defensively inside both sources, +because the cache is keyed by exact string. + +```python +@router.post("/api/watchlist") +async def add_to_watchlist( + body: TickerBody, + source: MarketDataSource = Depends(get_market_source), + cache: PriceCache = Depends(get_price_cache), +): + ticker = body.ticker.upper().strip() + db_add_ticker(ticker) + await source.add_ticker(ticker) + # Simulator: price is available immediately. Massive: None until next poll. + update = cache.get(ticker) + return {"ticker": ticker, "price": update.price if update else None} + + +@router.delete("/api/watchlist/{ticker}") +async def remove_from_watchlist( + ticker: str, + source: MarketDataSource = Depends(get_market_source), +): + ticker = ticker.upper().strip() + db_remove_ticker(ticker) + + # Keep tracking if the user still holds shares — portfolio valuation + # reads the same cache, and dropping the ticker would zero the position. + position = db_get_position(ticker) + if position is None or position.quantity == 0: + await source.remove_ticker(ticker) + + return {"ok": True} +``` + +That position check is the one piece of business logic the market layer does +not own but depends on. It belongs in the watchlist route, not in the data +source — the source has no concept of a portfolio. + +--- + +## 14. Consumer Patterns + +```python +cache: PriceCache = request.app.state.price_cache + +# Single price, for trade execution +price = cache.get_price("AAPL") +if price is None: + raise HTTPException(400, "No price available for AAPL yet") + +# Full update with direction, for display +update = cache.get("AAPL") + +# Portfolio valuation across all positions +prices = cache.get_all() +total_value = cash + sum( + pos.quantity * prices[pos.ticker].price + for pos in positions + if pos.ticker in prices +) +``` + +**Always handle `None`.** A ticker is absent from the cache when it was just +added and the first Massive poll has not landed, or when every poll has failed +since startup. Trade execution must reject rather than assume a price: + +```python +@router.post("/api/portfolio/trade") +async def execute_trade( + body: TradeRequest, + cache: PriceCache = Depends(get_price_cache), +): + ticker = body.ticker.upper().strip() + price = cache.get_price(ticker) + if price is None: + raise HTTPException( + 400, f"Price not yet available for {ticker}. Try again in a moment." + ) + # ... validate cash / shares, insert trade, update position, snapshot ... +``` + +The LLM context builder reads the same way — cash, positions marked to +`cache.get_all()`, watchlist with live prices — so the assistant sees exactly +the numbers the user sees on screen. + +--- + +## 15. Frontend Integration + +```typescript +type PriceUpdate = { + ticker: string; + price: number; + previous_price: number; + timestamp: number; // Unix seconds + change: number; + change_percent: number; + direction: "up" | "down" | "flat"; +}; + +const es = new EventSource("/api/stream/prices"); + +es.onmessage = (event) => { + const prices: Record = JSON.parse(event.data); + for (const [ticker, update] of Object.entries(prices)) { + applyPrice(ticker, update); // flash green/red on direction + appendSparklinePoint(ticker, update.price, update.timestamp); + } + setConnectionStatus("connected"); +}; + +es.onerror = () => setConnectionStatus("reconnecting"); +es.onopen = () => setConnectionStatus("connected"); +``` + +`EventSource` reconnects on its own using the `retry: 1000` directive; there is +no client reconnect logic to write. `onerror` fires on the drop, `onopen` on +recovery — which is exactly the signal the header's connection dot needs +(green / yellow / red). + +Sparklines accumulate client-side from the stream since page load, per the +plan; there is no historical price endpoint, and none is needed. + +--- + +## 16. Testing Strategy + +The layering makes each piece independently testable. Backend tests live in +`backend/tests/market/`, run with `uv run --extra dev pytest`. + +| Module | Test file | What it covers | +|---|---|---| +| `models.py` | `test_models.py` | Property math, zero-division guard, `to_dict()` shape | +| `cache.py` | `test_cache.py` | update/get/remove, version, flat-first, rounding, threads | +| `simulator.py` | `test_simulator.py` | GBM math, correlation, shocks, add/remove | +| `simulator.py` | `test_simulator_source.py` | Async lifecycle against a real cache | +| `massive_client.py` | `test_massive.py` | Parsing, timestamp units, error resilience | +| `factory.py` | `test_factory.py` | Env-var selection across all four cases | + +### 16.1 Pure math — no event loop, no mocks + +```python +import numpy as np +import pytest + +from app.market.seed_prices import SEED_PRICES +from app.market.simulator import GBMSimulator + + +def test_prices_stay_positive(): + sim = GBMSimulator(tickers=["AAPL"]) + for _ in range(10_000): + assert sim.step()["AAPL"] > 0 + + +def test_initial_price_is_the_seed(): + sim = GBMSimulator(tickers=["AAPL"]) + assert sim.get_price("AAPL") == SEED_PRICES["AAPL"] + + +def test_unknown_ticker_gets_random_seed_in_range(): + sim = GBMSimulator(tickers=["ZZZZ"]) + assert 50.0 <= sim.get_price("ZZZZ") <= 300.0 + + +def test_default_params_are_copied_not_shared(): + sim = GBMSimulator(tickers=["ZZZZ", "YYYY"]) + sim._params["ZZZZ"]["sigma"] = 99.0 + assert sim._params["YYYY"]["sigma"] == 0.25 + + +def test_cholesky_builds_for_full_default_watchlist(): + sim = GBMSimulator(tickers=list(SEED_PRICES)) + assert sim._cholesky is not None + corr = sim._cholesky @ sim._cholesky.T + assert np.linalg.eigvalsh(corr).min() > 0 # positive definite + i, j = list(SEED_PRICES).index("AAPL"), list(SEED_PRICES).index("MSFT") + assert corr[i, j] == pytest.approx(0.6, abs=1e-9) + + +def test_empirical_volatility_matches_sigma(): + np.random.seed(42) + sim = GBMSimulator(tickers=["AAPL"], event_probability=0.0) + prices = [sim.get_price("AAPL")] + for _ in range(20_000): + prices.append(sim.step()["AAPL"]) + returns = np.diff(np.log(prices)) + expected = 0.22 * np.sqrt(GBMSimulator.DEFAULT_DT) + assert returns.std() == pytest.approx(expected, rel=0.1) + + +def test_shock_events_are_deterministic_at_the_extremes(): + always = GBMSimulator(tickers=["AAPL"], event_probability=1.0) + before = always.get_price("AAPL") + after = always.step()["AAPL"] + assert abs(after / before - 1) >= 0.019 # a 2-5% shock landed + + never = GBMSimulator(tickers=["AAPL"], event_probability=0.0) + b = never.get_price("AAPL") + a = never.step()["AAPL"] + assert abs(a / b - 1) < 0.001 # diffusion only + + +def test_single_and_empty_ticker_do_not_crash(): + assert GBMSimulator(tickers=[]).step() == {} + assert GBMSimulator(tickers=["AAPL"])._cholesky is None +``` + +Setting `event_probability=0.0` is what makes the volatility test stable — a +stray 2-5% shock in a 20,000-step sample would swamp a `σ·√dt` measurement. + +### 16.2 Cache — including real thread contention + +```python +import threading + +from app.market.cache import PriceCache + + +def test_first_update_is_flat(): + cache = PriceCache() + update = cache.update("AAPL", 190.50) + assert update.direction == "flat" + assert update.previous_price == 190.50 + assert update.change == 0.0 + + +def test_direction_and_change(): + cache = PriceCache() + cache.update("AAPL", 190.00) + up = cache.update("AAPL", 191.00) + assert (up.direction, up.change) == ("up", 1.0) + down = cache.update("AAPL", 189.00) + assert (down.direction, down.change) == ("down", -2.0) + + +def test_price_is_rounded_at_the_boundary(): + cache = PriceCache() + assert cache.update("AAPL", 190.41999999).price == 190.42 + + +def test_explicit_zero_timestamp_is_preserved(): + cache = PriceCache() + assert cache.update("AAPL", 190.0, timestamp=0.0).timestamp == 0.0 + + +def test_version_increments_on_every_update(): + cache = PriceCache() + v0 = cache.version + cache.update("AAPL", 190.00) + cache.update("AAPL", 190.00) # same price still counts as an observation + assert cache.version == v0 + 2 + + +def test_concurrent_writes_do_not_lose_updates(): + cache = PriceCache() + + def writer(ticker: str): + for i in range(1000): + cache.update(ticker, 100.0 + i) + + threads = [threading.Thread(target=writer, args=(f"T{i}",)) for i in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert cache.version == 8000 + assert len(cache) == 8 +``` + +### 16.3 Simulator source — async integration + +```python +import asyncio + +import pytest + +from app.market.cache import PriceCache +from app.market.simulator import SimulatorDataSource + + +async def test_start_populates_cache_before_returning(): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.01) + await source.start(["AAPL", "GOOGL"]) + # No sleep — start() must have seeded synchronously. + assert cache.get("AAPL") is not None + assert cache.get("GOOGL") is not None + await source.stop() + + +async def test_stop_halts_writes(): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.01) + await source.start(["AAPL"]) + await asyncio.sleep(0.05) + await source.stop() + + frozen = cache.version + await asyncio.sleep(0.05) + assert cache.version == frozen + + +async def test_stop_is_idempotent_even_if_never_started(): + source = SimulatorDataSource(price_cache=PriceCache()) + await source.stop() + await source.stop() + + +async def test_add_seeds_and_remove_evicts(): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.01) + await source.start(["AAPL"]) + + await source.add_ticker("TSLA") + assert "TSLA" in source.get_tickers() + assert cache.get("TSLA") is not None # seeded immediately, no tick needed + + await source.remove_ticker("TSLA") + assert "TSLA" not in source.get_tickers() + assert cache.get("TSLA") is None # evicted from the cache too + + await source.stop() + + +async def test_step_failure_does_not_kill_the_loop(monkeypatch): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.01) + await source.start(["AAPL"]) + + calls = {"n": 0} + real_step = source._sim.step + + def flaky(): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("boom") + return real_step() + + monkeypatch.setattr(source._sim, "step", flaky) + await asyncio.sleep(0.05) + assert calls["n"] > 1 # loop survived and kept ticking + await source.stop() +``` + +Tests use `update_interval=0.01` so integration cases finish in milliseconds. +`asyncio_mode = "auto"` in `pyproject.toml` means no `@pytest.mark.asyncio` +decorator is needed. + +### 16.4 Massive — build snapshots from raw JSON + +Hand-constructing a `MagicMock` snapshot passes even when the attribute names +are wrong, which is exactly the bug class that hurts here. Build from raw JSON +via the SDK's own `from_dict` so a `sip_timestamp`/`timestamp` mix-up fails the +test: + +```python +from unittest.mock import patch + +from massive.rest.models import TickerSnapshot + +from app.market.cache import PriceCache +from app.market.massive_client import ( + MassiveDataSource, + extract_price, + normalize_timestamp, +) + +RAW = { + "ticker": "AAPL", + "todaysChange": 1.25, + "todaysChangePerc": 0.66, + "updated": 1785432000_000_000_000, # nanoseconds + "day": {"o": 189.0, "h": 191.5, "l": 188.7, "c": 190.5, "v": 41_000_000}, + "prevDay": {"o": 187.0, "h": 189.9, "l": 186.5, "c": 189.25, "v": 38_000_000}, + "min": {"o": 190.3, "h": 190.6, "l": 190.2, "c": 190.42, "v": 120_000}, + "lastTrade": {"p": 190.42, "s": 100, "t": 1785432000_123_000_000}, +} + + +def _snapshot(*drop: str) -> TickerSnapshot: + """Build a real TickerSnapshot from raw JSON, omitting the named keys. + + Keys are dropped rather than set to None: `from_dict` calls `.get()` on + each nested object, so an explicit None raises inside the SDK. + """ + return TickerSnapshot.from_dict({k: v for k, v in RAW.items() if k not in drop}) + + +def test_normalize_timestamp_handles_every_unit(): + seconds = 1785432000.0 + assert normalize_timestamp(1785432000) == seconds + assert normalize_timestamp(1785432000_000) == seconds # ms + assert normalize_timestamp(1785432000_000_000) == seconds # us + assert normalize_timestamp(1785432000_000_000_000) == seconds # ns + assert normalize_timestamp(None) is None + + +def test_price_fallback_chain(): + assert extract_price(_snapshot()) == 190.42 # last_trade + assert extract_price(_snapshot("lastTrade")) == 190.42 # min bar + assert extract_price(_snapshot("lastTrade", "min")) == 190.5 # day + assert extract_price(_snapshot("lastTrade", "min", "day")) == 189.25 # prev_day + assert extract_price(_snapshot("lastTrade", "min", "day", "prevDay")) is None + + +def test_snapshot_last_trade_has_no_timestamp_attribute(): + """Guards against regressing to `snap.last_trade.timestamp` (see §20).""" + with pytest.raises(AttributeError): + _snapshot().last_trade.timestamp + + +async def test_poll_writes_price_and_seconds_timestamp(): + cache = PriceCache() + source = MassiveDataSource("k", cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = object() # non-None so _poll_once proceeds + + with patch.object(source, "_fetch_snapshots", return_value=[_snapshot()]): + await source._poll_once() + + update = cache.get("AAPL") + assert update.price == 190.42 + assert 1_700_000_000 < update.timestamp < 2_000_000_000 # plausible seconds + + +async def test_one_bad_snapshot_does_not_abort_the_batch(): + cache = PriceCache() + source = MassiveDataSource("k", cache, poll_interval=60.0) + source._tickers = ["AAPL", "BAD"] + source._client = object() + + class Broken: + ticker = "BAD" + last_trade = None + min = None + day = None + prev_day = None + + with patch.object( + source, "_fetch_snapshots", return_value=[_snapshot(), Broken()] + ): + await source._poll_once() + + assert cache.get_price("AAPL") == 190.42 + assert cache.get_price("BAD") is None + + +async def test_api_failure_is_swallowed(): + cache = PriceCache() + source = MassiveDataSource("k", cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = object() + + with patch.object(source, "_fetch_snapshots", side_effect=Exception("429")): + await source._poll_once() # must not raise + + assert cache.get_price("AAPL") is None +``` + +Setting `source._client` directly (rather than calling `start()`) is what keeps +these tests network-free while still exercising the real `_poll_once` path. + +### 16.5 Factory — all four env-var cases + +```python +import pytest + +from app.market.cache import PriceCache +from app.market.factory import create_market_data_source +from app.market.massive_client import MassiveDataSource +from app.market.simulator import SimulatorDataSource + + +@pytest.mark.parametrize( + "value,expected", + [ + (None, SimulatorDataSource), + ("", SimulatorDataSource), + (" ", SimulatorDataSource), + ("real-key", MassiveDataSource), + ], +) +def test_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) +``` + +### 16.6 Conformance — both sources, one test + +The contract in §6 is only real if both implementations satisfy it identically: + +```python +@pytest.fixture(params=["simulator", "massive"]) +def source_and_cache(request): + cache = PriceCache() + if request.param == "simulator": + yield SimulatorDataSource(price_cache=cache, update_interval=0.01), cache + else: + src = MassiveDataSource("k", cache, poll_interval=60.0) + with patch.object(src, "_fetch_snapshots", return_value=[_snapshot()]): + yield src, cache + + +async def test_contract(source_and_cache): + source, cache = source_and_cache + await source.start(["AAPL"]) + assert cache.get("AAPL") is not None # seeded before return + assert source.get_tickers() == ["AAPL"] + + await source.remove_ticker("AAPL") + assert cache.get("AAPL") is None # evicted + assert source.get_tickers() == [] + + await source.stop() + await source.stop() # idempotent +``` + +### 16.7 SSE — needs an ASGI client + +`stream.py` is the least covered module because the generator needs a running +ASGI app. One integration test is worth the setup: + +```python +import httpx +from fastapi import FastAPI + +from app.market import PriceCache, create_stream_router + + +async def test_sse_emits_a_frame_per_version_change(): + cache = PriceCache() + app = FastAPI() + app.include_router(create_stream_router(cache, interval=0.01)) + cache.update("AAPL", 190.42) + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as client: + async with client.stream("GET", "/api/stream/prices") as response: + assert response.headers["content-type"].startswith("text/event-stream") + frames = [] + async for line in response.aiter_lines(): + if line.startswith("data: "): + frames.append(json.loads(line[6:])) + if len(frames) == 1: + break + + assert frames[0]["AAPL"]["price"] == 190.42 + assert frames[0]["AAPL"]["direction"] == "flat" +``` + +### 16.8 Testing consumers without any data source + +The payoff of the cache-as-contract design: portfolio math, trade execution, +and LLM context are tested by pre-populating a cache. No simulator, no mocks, +no async. + +```python +def test_portfolio_valuation(): + cache = PriceCache() + cache.update("AAPL", 200.00) + cache.update("MSFT", 400.00) + + positions = [Position("AAPL", qty=10, avg_cost=190.0), + Position("MSFT", qty=5, avg_cost=420.0)] + + assert compute_total_value(cash=1000.0, positions=positions, cache=cache) == 5000.0 +``` + +--- + +## 17. Error Handling & Edge Cases + +| Situation | Behaviour | +|---|---| +| **Empty watchlist at startup** | `start([])` is valid. Simulator produces `{}` per tick; Massive skips the HTTP call. SSE emits nothing (empty dict is falsy). Adding a ticker starts tracking immediately. | +| **Trade on an uncached ticker** | `get_price()` returns `None` → HTTP 400 with a clear retry message. Never assume or fabricate a price. | +| **Invalid Massive key** | First poll 401s, logged, poller keeps retrying. SSE connects successfully but carries no data — the connection dot is green while the grid is empty. Documented so it is not mistaken for a stream bug. | +| **Free-tier Massive key** | Snapshot endpoint 403s every poll. Same visible outcome as above. Fix: unset `MASSIVE_API_KEY`. | +| **Massive returns a partial batch** | Missing tickers keep their last cached value. No entry is deleted on a failed poll. | +| **Ticker removed while held** | Watchlist route keeps it in the source when a position exists (§13), so valuation stays correct. | +| **Duplicate `add_ticker`** | Idempotent in both sources; no duplicate Cholesky row, no duplicated poll entry. | +| **Cholesky failure on a pathological correlation set** | Raised inside `step()` → caught by `_run_loop` → logged → retried. Prices freeze for one tick rather than forever. | +| **Client disconnects mid-frame** | `is_disconnected()` breaks the loop; `CancelledError` is caught and logged. No traceback spam in the logs on every closed tab. | +| **Float precision** | Not a concern: `exp()` is numerically stable, prices are strictly positive by construction, and everything rounds to 2 dp at the cache boundary. | +| **Memory** | The cache stores exactly one `PriceUpdate` per ticker — O(tickers), not O(time). No history accumulates server-side. | + +--- + +## 18. Configuration Reference + +| Parameter | Location | Default | Effect | +|---|---|---|---| +| `MASSIVE_API_KEY` | environment / `.env` | `""` | Non-empty selects the real API; otherwise simulator | +| `update_interval` | `SimulatorDataSource.__init__` | `0.5` s | Simulator tick cadence | +| `poll_interval` | `MassiveDataSource.__init__` | `15.0` s | Massive poll cadence (match your tier) | +| `event_probability` | `GBMSimulator.__init__` | `0.001` | Shock chance per ticker per tick | +| `dt` | `GBMSimulator.__init__` | `~8.479e-8` | GBM step as a fraction of a trading year | +| `interval` | `create_stream_router` / `_generate_events` | `0.5` s | SSE cache-poll cadence | +| retry directive | `_generate_events` | `1000` ms | Browser `EventSource` reconnect delay | +| `SEED_PRICES` | `seed_prices.py` | see §7 | Starting prices | +| `TICKER_PARAMS` | `seed_prices.py` | see §7 | Per-ticker σ and μ | +| correlation constants | `seed_prices.py` | 0.6 / 0.5 / 0.3 | Sector coupling | + +### Tuning guide + +| Goal | Change | +|---|---| +| More dramatic price action | Raise `sigma` in `TICKER_PARAMS` | +| Faster/slower updates | `SimulatorDataSource(update_interval=...)` | +| More/fewer shock events | `event_probability` | +| Stronger sector coupling | Raise `INTRA_TECH_CORR` — **re-verify positive definiteness** | +| Persistent bull market | Raise `mu` across `TICKER_PARAMS` | + +`update_interval` and `dt` are independent knobs. Halving the interval to +250 ms while leaving `dt` at its 500 ms value makes the market run at half +speed. To keep them consistent, derive one from the other: + +```python +dt = update_interval / GBMSimulator.TRADING_SECONDS_PER_YEAR +``` + +--- + +## 19. Extending to a New Provider + +To add an Alpaca, Finnhub, or IEX feed: + +1. Implement `MarketDataSource` in a new module under `app/market/`. +2. Convert that provider's price and timestamp to floats in Unix **seconds**, + and call `cache.update(ticker=..., price=..., timestamp=...)`. Normalize at + this boundary — never let provider units leak downstream. +3. Wrap any synchronous client in `asyncio.to_thread(...)`. +4. Catch all exceptions in the poll loop; log and continue. +5. Seed the cache before `start()` returns. +6. Add a branch to `create_market_data_source()`. +7. Add the class to the conformance-test parametrization (§16.6). + +No consumer changes, no SSE changes, no frontend changes. That property is the +entire point of this design. + +--- + +## 20. Deltas From the Current Implementation + +The code in `backend/app/market/` predates this document and is broadly +identical to it. Four points in §5, §9, and §11 above describe the intended +design where the shipped code currently differs. Listed here so the difference +is a decision rather than a surprise. + +| # | Location | Current code | This design | Severity | +|---|---|---|---|---| +| 1 | `massive_client.py` `_poll_once` | `snap.last_trade.timestamp` | `extract_timestamp()` → `sip_timestamp`, normalized | **High** — the snapshot `LastTrade` has no `.timestamp`; per `MASSIVE_API.md` §4.1 this raises `AttributeError`, and the per-ticker guard catches it, so *every* ticker is skipped and the cache never fills | +| 2 | `massive_client.py` `_poll_once` | `snap.last_trade.price` only | `extract_price()` fallback chain `last_trade → min → day → prev_day` | **High** — outside US market hours, or on a delayed plan, `last_trade` is `None` and no price is ever produced | +| 3 | `massive_client.py` | `/ 1000.0` (assumes milliseconds) | `normalize_timestamp()` scales any unit to seconds | **Medium** — snapshot SIP timestamps are nanoseconds; dividing by 1000 yields a year-56000 timestamp | +| 4 | `stream.py` | Module-level `router`, mutated by `create_stream_router()` | Router constructed inside the factory | **Low** — a second call registers a duplicate `/prices` route; only bites when two tests each build an app | +| 5 | `cache.py` | `ts = timestamp or time.time()` | `ts = time.time() if timestamp is None else timestamp` | **Low** — an explicit `0.0` is silently replaced by "now" | +| 6 | `cache.py` | `version` property reads without the lock | Reads under the lock | **Trivial** — atomic under the GIL today | + +Items 1–3 are all in the Massive path, which is unreachable without a paid API +key and is therefore untested against the live API. They share one root cause: +the original client was written against an assumed field layout rather than the +SDK's verified one. `MASSIVE_API.md` is the authority — its field tables were +verified by introspecting the installed SDK. + +Fixing 1–3 is a contained change to `_poll_once` plus the two helper functions, +and §16.4 gives the tests that would have caught them: building snapshots with +`TickerSnapshot.from_dict(raw_json)` rather than `MagicMock` makes a wrong +attribute name fail loudly instead of silently returning another mock.