From 9f674c08e9040c8bca7c1210147360708788de99 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Thu, 3 Sep 2026 20:48:50 +0000 Subject: [PATCH] Complete the market data backend per planning/MARKET_INTERFACE.md, MARKET_SIMULATOR.md, MASSIVE_API.md Implements the reviewed design for the market data subsystem, fixing bugs identified in the project's own planning docs rather than leaving them latent: - Massive client: fixed the nanosecond `sip_timestamp` field (was reading a nonexistent `.timestamp` in ms, silently swallowed by the exception handler), set retries=0 (SDK's default backoff is inside the same rate-limit window), classified rate-limit errors distinctly from bad responses, captured `prev_day.close` as the daily-change baseline, and surfaced tickers missing from a poll instead of dropping them silently. - New capabilities.py: probes Massive entitlement once at startup, since a free/Basic-tier key authenticates but can't return any live price. - New anchored.py: AnchoredSimulatorDataSource anchors the GBM simulator to real closing prices (one free-tier API call) for keys that are valid but not live-entitled, so a free key still produces a moving, realistic terminal instead of an empty watchlist. - factory.py: now async and capability-probe-driven instead of routing on key presence alone; every branch returns a working source. - Simulator: unknown-ticker seeds are now deterministic (hashed) instead of random per restart; random events are a decaying overlay instead of a permanent level shift (the old calibration inflated realized volatility ~20x); added a LinAlgError fallback and a 5-minute prefilled history ring buffer per ticker. - models.py: PriceUpdate now carries a fixed session `open_price` baseline (change_today/change_percent_today) alongside the existing tick-to-tick change (renamed direction -> tick_direction); wire format is ISO 8601 UTC. - stream.py: SSE heartbeat so idle connections (closed market, slow Massive poll) don't silently die at a proxy. - MarketDataSource interface: added describe() (for /api/health) and get_history() (so charts aren't empty on first paint). - Refreshed seed_prices.py to real closing prices (NVDA/NFLX were 3-7x off after 2024 splits). Rewrote and substantially expanded the unit test suite across all of the above (models, cache, interface, capabilities, simulator, simulator source, massive client, anchored source, factory, SSE stream), including statistical regression tests for the GBM calibration and the shock-decay fix. All Massive API calls are mocked; no test touches the network. Closes #2 Co-authored-by: cloudnote18-lab <290535093+cloudnote18-lab@users.noreply.github.com> --- backend/CLAUDE.md | 10 +- backend/README.md | 10 +- backend/app/market/__init__.py | 9 +- backend/app/market/anchored.py | 173 ++++++++++ backend/app/market/cache.py | 28 +- backend/app/market/capabilities.py | 60 ++++ backend/app/market/factory.py | 46 ++- backend/app/market/interface.py | 16 +- backend/app/market/massive_client.py | 139 ++++++-- backend/app/market/models.py | 81 ++++- backend/app/market/seed_prices.py | 31 +- backend/app/market/simulator.py | 193 ++++++++++-- backend/app/market/stream.py | 21 +- backend/market_data_demo.py | 8 +- backend/tests/market/test_anchored.py | 235 ++++++++++++++ backend/tests/market/test_cache.py | 44 ++- backend/tests/market/test_capabilities.py | 111 +++++++ backend/tests/market/test_factory.py | 104 ++++-- backend/tests/market/test_interface.py | 60 ++++ backend/tests/market/test_massive.py | 180 +++++++++-- backend/tests/market/test_models.py | 199 ++++++++++-- backend/tests/market/test_simulator.py | 298 +++++++++++++++++- backend/tests/market/test_simulator_source.py | 147 +++++++++ backend/tests/market/test_stream.py | 121 +++++++ planning/MARKET_DATA_SUMMARY.md | 53 ++++ 25 files changed, 2177 insertions(+), 200 deletions(-) create mode 100644 backend/app/market/anchored.py create mode 100644 backend/app/market/capabilities.py create mode 100644 backend/tests/market/test_anchored.py create mode 100644 backend/tests/market/test_capabilities.py create mode 100644 backend/tests/market/test_interface.py create mode 100644 backend/tests/market/test_stream.py create mode 100644 planning/MARKET_DATA_SUMMARY.md diff --git a/backend/CLAUDE.md b/backend/CLAUDE.md index 612ff18f5..8df8b871e 100644 --- a/backend/CLAUDE.md +++ b/backend/CLAUDE.md @@ -12,24 +12,24 @@ uv sync --extra dev # Install all dependencies including test/lint tools The market data subsystem lives in `app/market/`. Use these imports: ```python -from app.market import PriceCache, PriceUpdate, MarketDataSource, create_market_data_source +from app.market import PriceCache, PriceUpdate, PricePoint, SourceStatus, MarketDataSource, create_market_data_source ``` ### Core Types -- **`PriceUpdate`** — Immutable dataclass: `ticker`, `price`, `previous_price`, `timestamp`, plus properties `change`, `change_percent`, `direction` ("up"/"down"/"flat"), and `to_dict()` for JSON serialization. +- **`PriceUpdate`** — Immutable dataclass: `ticker`, `price`, `previous_price` (previous tick), `open_price` (session baseline), `timestamp`, plus properties `change`/`change_percent` (tick-to-tick), `tick_direction` ("up"/"down"/"flat", drives the flash animation), `change_today`/`change_percent_today` (vs. `open_price`, drives the daily % column), and `to_dict()` for JSON serialization (timestamp is ISO 8601 UTC on the wire). - **`PriceCache`** — Thread-safe in-memory store. Key methods: - - `update(ticker, price, timestamp=None) -> PriceUpdate` + - `update(ticker, price, timestamp=None, open_price=None) -> PriceUpdate` — `open_price` sticks for the session once set; omit it on subsequent calls to keep the existing baseline. - `get(ticker) -> PriceUpdate | None` - `get_price(ticker) -> float | None` - `get_all() -> dict[str, PriceUpdate]` - `remove(ticker)` - `version` property — monotonic counter, increments on every update (for SSE change detection) -- **`MarketDataSource`** — Abstract interface implemented by `SimulatorDataSource` and `MassiveDataSource`. Lifecycle: `start(tickers)` -> `add_ticker()` / `remove_ticker()` -> `stop()`. +- **`MarketDataSource`** — Abstract interface implemented by `SimulatorDataSource`, `AnchoredSimulatorDataSource`, and `MassiveDataSource`. Lifecycle: `start(tickers)` -> `add_ticker()` / `remove_ticker()` -> `stop()`. Also: `describe() -> SourceStatus` (for `GET /api/health`) and `get_history(ticker, points=120) -> list[PricePoint]` (for chart first-paint; default `[]`). -- **`create_market_data_source(cache)`** — Factory. Returns `MassiveDataSource` if `MASSIVE_API_KEY` is set, otherwise `SimulatorDataSource`. +- **`create_market_data_source(cache)`** — **Async** factory; must be awaited from the FastAPI lifespan handler before `start()`. Selection is driven by a one-time Massive entitlement probe (`capabilities.py`), not just key presence — a key that authenticates but isn't entitled to live prices (free/Basic tier) routes to `AnchoredSimulatorDataSource` (real closing prices, synthetic motion) rather than a `MassiveDataSource` that would poll forever and never populate the cache. See `planning/MARKET_INTERFACE.md` for the full decision table. ### SSE Streaming diff --git a/backend/README.md b/backend/README.md index 7cdd84757..5fe0d4cb9 100644 --- a/backend/README.md +++ b/backend/README.md @@ -6,13 +6,15 @@ FastAPI backend for the FinAlly AI Trading Workstation. - `app/` - Application code - `market/` - Market data subsystem - - `models.py` - PriceUpdate dataclass + - `models.py` - PriceUpdate/PricePoint/SourceStatus dataclasses - `cache.py` - Thread-safe price cache - `interface.py` - MarketDataSource abstract interface + - `capabilities.py` - Massive API entitlement probe - `simulator.py` - GBM-based market simulator - - `massive_client.py` - Massive/Polygon.io API client - - `factory.py` - Data source factory - - `stream.py` - SSE streaming endpoint + - `anchored.py` - Simulator seeded from real Massive closing prices (free-tier keys) + - `massive_client.py` - Massive/Polygon.io API client (real-time tiers) + - `factory.py` - Async, capability-driven data source factory + - `stream.py` - SSE streaming endpoint (with heartbeat) - `seed_prices.py` - Default ticker prices and parameters - `tests/` - Unit and integration tests diff --git a/backend/app/market/__init__.py b/backend/app/market/__init__.py index 57ad0a121..c6d9c93e8 100644 --- a/backend/app/market/__init__.py +++ b/backend/app/market/__init__.py @@ -2,20 +2,25 @@ Public API: PriceUpdate - Immutable price snapshot dataclass + PricePoint - Single point in a historical price series + SourceStatus - Introspection payload for GET /api/health PriceCache - Thread-safe in-memory price store MarketDataSource - Abstract interface for data providers - create_market_data_source - Factory that selects simulator or Massive + create_market_data_source - Async factory; probes Massive entitlement and + selects simulator / anchored-simulator / massive create_stream_router - FastAPI router factory for SSE endpoint """ from .cache import PriceCache from .factory import create_market_data_source from .interface import MarketDataSource -from .models import PriceUpdate +from .models import PricePoint, PriceUpdate, SourceStatus from .stream import create_stream_router __all__ = [ "PriceUpdate", + "PricePoint", + "SourceStatus", "PriceCache", "MarketDataSource", "create_market_data_source", diff --git a/backend/app/market/anchored.py b/backend/app/market/anchored.py new file mode 100644 index 000000000..1c76f745d --- /dev/null +++ b/backend/app/market/anchored.py @@ -0,0 +1,173 @@ +"""GBM simulation seeded from real Massive closing prices. + +Bridges the gap for Basic-tier (free) Massive keys, which authenticate fine but +are not entitled to any live price. One free-tier API call +(`get_grouped_daily_aggs`) prices the whole market at once, so the simulator +starts from real closing levels — e.g. AAPL at its genuine close instead of a +hard-coded, rapidly stale seed — while GBM supplies the tick-to-tick motion so +the terminal still looks alive. See planning/MARKET_INTERFACE.md §6. +""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import date, timedelta + +from massive import RESTClient + +from .cache import PriceCache +from .interface import MarketDataSource +from .models import PricePoint, SourceStatus, epoch_to_iso +from .simulator import SimulatorDataSource + +logger = logging.getLogger(__name__) + +MAX_ANCHOR_LOOKBACK_DAYS = 7 # walk back at most a week to skip weekends/holidays +DEFAULT_REANCHOR_INTERVAL_SECONDS = 3600.0 # re-anchor hourly so a long-running container +# tracks the next session's close instead of drifting from reality + + +class AnchoredSimulatorDataSource(MarketDataSource): + """GBM simulation seeded from real Massive closing prices. + + Bridges the gap for Basic-tier keys: real price *levels* from one + free-tier API call, plus synthetic price *motion* so the terminal is alive. + Displayed prices are simulated, not live — describe().live is always False. + """ + + def __init__( + self, + api_key: str, + price_cache: PriceCache, + update_interval: float = 0.5, + reanchor_interval: float = DEFAULT_REANCHOR_INTERVAL_SECONDS, + ) -> None: + self._client = RESTClient(api_key=api_key, retries=0) + self._cache = price_cache + self._update_interval = update_interval + self._reanchor_interval = reanchor_interval + self._sim: SimulatorDataSource | None = None + self._anchors: dict[str, float] = {} + self._anchor_date: str | None = None + self._reanchor_task: asyncio.Task | None = None + + async def start(self, tickers: list[str]) -> None: + self._anchors, self._anchor_date = await asyncio.to_thread( + self._fetch_anchors, tickers + ) + # Real closes where we have them; the static seed table covers the rest. + self._sim = SimulatorDataSource( + self._cache, + update_interval=self._update_interval, + seed_overrides=self._anchors, + status_detail=self._status_detail(), + ) + await self._sim.start(tickers) + + if self._reanchor_interval > 0: + self._reanchor_task = asyncio.create_task( + self._reanchor_loop(), name="anchored-reanchor" + ) + + async def stop(self) -> None: + if self._reanchor_task and not self._reanchor_task.done(): + self._reanchor_task.cancel() + try: + await self._reanchor_task + except asyncio.CancelledError: + pass + self._reanchor_task = None + if self._sim: + await self._sim.stop() + + async def add_ticker(self, ticker: str) -> None: + # Costs nothing: the underlying simulator synthesizes a seed for any + # symbol not in the anchor set, so the demo never dead-ends on an + # unknown ticker. + if self._sim: + await self._sim.add_ticker(ticker) + + async def remove_ticker(self, ticker: str) -> None: + if self._sim: + await self._sim.remove_ticker(ticker) + + def get_tickers(self) -> list[str]: + return self._sim.get_tickers() if self._sim else [] + + def describe(self) -> SourceStatus: + return SourceStatus( + name="anchored-simulator", + live=False, + detail=self._status_detail(), + tickers=len(self.get_tickers()), + cache_populated=len(self._cache) > 0, + ) + + async def get_history(self, ticker: str, points: int = 120) -> list[PricePoint]: + """Real minute bars from the last completed session (free tier allows this).""" + if not self._anchor_date: + return [] + try: + bars = await asyncio.to_thread( + self._client.get_aggs, + ticker, + 1, + "minute", + self._anchor_date, + self._anchor_date, + limit=50_000, + ) + except Exception as e: + logger.warning("Anchored get_history failed for %s: %s", ticker, e) + return [] + return [PricePoint(epoch_to_iso(bar.timestamp / 1000), bar.close) for bar in bars[-points:]] + + # --- Internal --- + + def _fetch_anchors(self, tickers: list[str]) -> tuple[dict[str, float], str | None]: + """ONE API call prices every ticker. Walks back over weekends/holidays.""" + wanted = {t.upper() for t in tickers} + day = date.today() + for _ in range(MAX_ANCHOR_LOOKBACK_DAYS): + day -= timedelta(days=1) + iso = day.isoformat() + try: + bars = self._client.get_grouped_daily_aggs(iso, adjusted=True) + except Exception as e: + logger.warning("Anchor fetch failed for %s: %s", iso, e) + continue + if not bars: + continue # weekend or holiday + found = {bar.ticker: bar.close for bar in bars if bar.ticker in wanted} + logger.info( + "Anchored %d/%d tickers to %s closes", len(found), len(wanted), iso + ) + return found, iso + logger.warning("No anchors available — falling back to the static seed table") + return {}, None + + def _status_detail(self) -> str: + if self._anchor_date: + return ( + f"simulated from real {self._anchor_date} closes " + f"({len(self._anchors)} anchored)" + ) + return "simulated from static seed prices (anchor fetch failed)" + + async def _reanchor_loop(self) -> None: + """Periodically re-fetch anchors so a long-running container tracks the + next session's close instead of drifting from reality.""" + while True: + await asyncio.sleep(self._reanchor_interval) + try: + tickers = self.get_tickers() + anchors, anchor_date = await asyncio.to_thread( + self._fetch_anchors, tickers + ) + if anchors: + self._anchors, self._anchor_date = anchors, anchor_date + if self._sim: + self._sim.set_status_detail(self._status_detail()) + except Exception: + logger.exception("Re-anchor failed") diff --git a/backend/app/market/cache.py b/backend/app/market/cache.py index 4d0215778..d60354184 100644 --- a/backend/app/market/cache.py +++ b/backend/app/market/cache.py @@ -11,8 +11,8 @@ class PriceCache: """Thread-safe in-memory cache of the latest price for each ticker. - Writers: SimulatorDataSource or MassiveDataSource (one at a time). - Readers: SSE streaming endpoint, portfolio valuation, trade execution. + Writers: SimulatorDataSource, AnchoredSimulatorDataSource, or MassiveDataSource + (one at a time). Readers: SSE streaming endpoint, portfolio valuation, trade execution. """ def __init__(self) -> None: @@ -20,21 +20,39 @@ def __init__(self) -> None: self._lock = Lock() self._version: int = 0 # Monotonically increasing; bumped on every update - def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: + def update( + self, + ticker: str, + price: float, + timestamp: float | None = None, + open_price: float | None = None, + ) -> PriceUpdate: """Record a new price for a ticker. Returns the created PriceUpdate. - Automatically computes direction and change from the previous price. - If this is the first update for the ticker, previous_price == price (direction='flat'). + Automatically computes direction and change from the previous tick. + If this is the first update for the ticker, previous_price == price (tick_direction='flat'). + + `open_price` is the session baseline used for the daily change column. When + omitted, the ticker keeps whatever open_price it already had; on the very + first write it defaults to `price`. """ with self._lock: ts = timestamp or time.time() prev = self._prices.get(ticker) previous_price = prev.price if prev else price + if open_price is not None: + resolved_open = open_price + elif prev is not None: + resolved_open = prev.open_price + else: + resolved_open = price + update = PriceUpdate( ticker=ticker, price=round(price, 2), previous_price=round(previous_price, 2), + open_price=round(resolved_open, 2), timestamp=ts, ) self._prices[ticker] = update diff --git a/backend/app/market/capabilities.py b/backend/app/market/capabilities.py new file mode 100644 index 000000000..b54f143cb --- /dev/null +++ b/backend/app/market/capabilities.py @@ -0,0 +1,60 @@ +"""Entitlement probing for the Massive API. + +A Massive API key can be valid but still unable to return live prices — a free +Basic-tier key authenticates successfully and then returns NOT_AUTHORIZED for +every snapshot/last-trade endpoint, while end-of-day aggregate endpoints work +fine. This module answers "what can this key actually do?" once at startup so +the factory can route to a source that will actually produce prices, instead +of a source that quietly fails on every poll. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +import urllib3.exceptions +from massive import RESTClient +from massive.exceptions import AuthError, BadResponse + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True, slots=True) +class MassiveCapabilities: + """What a given API key is actually allowed to do.""" + + valid: bool # key authenticates at all + realtime: bool # snapshot / last-trade endpoints entitled + end_of_day: bool # aggregate endpoints entitled + detail: str + + +def probe_capabilities(api_key: str) -> MassiveCapabilities: + """Two cheap calls, run once at startup. Costs 2 of the free tier's 5/min budget. + + Never raises — every failure mode is captured in the returned MassiveCapabilities. + """ + try: + client = RESTClient(api_key=api_key, retries=0, read_timeout=5.0) + except AuthError: + return MassiveCapabilities(False, False, False, "no API key configured") + + # 1. Cheapest possible entitlement test for real-time. + try: + client.get_snapshot_all(market_type="stocks", tickers=["AAPL"]) + return MassiveCapabilities(True, True, True, "real-time snapshots entitled") + except BadResponse as e: + if "NOT_AUTHORIZED" not in str(e): + return MassiveCapabilities(False, False, False, f"unexpected: {e}") + except urllib3.exceptions.MaxRetryError as e: + return MassiveCapabilities(False, False, False, f"unreachable/rate-limited: {e}") + except Exception as e: # pragma: no cover - defensive catch-all, never raise + return MassiveCapabilities(False, False, False, f"unexpected: {e}") + + # 2. Snapshots refused — is this a valid key on a lower plan, or a bad key? + try: + client.get_previous_close_agg("AAPL") + return MassiveCapabilities(True, False, True, "end-of-day only (Basic tier)") + except Exception as e: + return MassiveCapabilities(False, False, False, f"key rejected: {e}") diff --git a/backend/app/market/factory.py b/backend/app/market/factory.py index 00360e94f..b6bfcd773 100644 --- a/backend/app/market/factory.py +++ b/backend/app/market/factory.py @@ -2,10 +2,13 @@ from __future__ import annotations +import asyncio import logging import os +from .anchored import AnchoredSimulatorDataSource from .cache import PriceCache +from .capabilities import probe_capabilities from .interface import MarketDataSource from .massive_client import MassiveDataSource from .simulator import SimulatorDataSource @@ -13,19 +16,40 @@ logger = logging.getLogger(__name__) -def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: - """Create the appropriate market data source based on environment variables. +async def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """Select a market data source. Never raises; always returns a working source. - - MASSIVE_API_KEY set and non-empty → MassiveDataSource (real market data) - - Otherwise → SimulatorDataSource (GBM simulation) + Async because, when MASSIVE_API_KEY is set, this makes real network calls to + probe entitlement. Call once, from the FastAPI lifespan handler, before + start() is invoked. - Returns an unstarted source. Caller must await source.start(tickers). + A Massive key can be valid but unable to return live prices (Basic/free + tier is end-of-day only). Selection is therefore driven by a one-time + capability probe rather than by key presence alone: + + - No key -> SimulatorDataSource (synthetic seeds) + - Key, real-time entitled -> MassiveDataSource (live snapshots) + - Key, end-of-day only -> AnchoredSimulatorDataSource (real closes, synthetic motion) + - Key, invalid/rejected -> SimulatorDataSource (never boot into a broken state) """ api_key = os.environ.get("MASSIVE_API_KEY", "").strip() - if api_key: - logger.info("Market data source: Massive API (real data)") - return MassiveDataSource(api_key=api_key, price_cache=price_cache) - else: - logger.info("Market data source: GBM Simulator") - return SimulatorDataSource(price_cache=price_cache) + if not api_key: + logger.info("No MASSIVE_API_KEY — using GBM simulator") + return SimulatorDataSource(price_cache) + + caps = await asyncio.to_thread(probe_capabilities, api_key) + + if caps.realtime: + logger.info("Massive: real-time entitled — using live snapshots") + return MassiveDataSource(api_key, price_cache, poll_interval=5.0) + + if caps.end_of_day: + logger.warning( + "Massive key is end-of-day only (Basic tier). Anchoring the simulator " + "to real closing prices — displayed prices are simulated, not live." + ) + return AnchoredSimulatorDataSource(api_key, price_cache) + + logger.error("MASSIVE_API_KEY rejected (%s) — falling back to simulator", caps.detail) + return SimulatorDataSource(price_cache, status_detail=f"key rejected: {caps.detail}") diff --git a/backend/app/market/interface.py b/backend/app/market/interface.py index 0f3b7d8c9..911eb4c6b 100644 --- a/backend/app/market/interface.py +++ b/backend/app/market/interface.py @@ -4,6 +4,8 @@ from abc import ABC, abstractmethod +from .models import PricePoint, SourceStatus + class MarketDataSource(ABC): """Contract for market data providers. @@ -13,7 +15,7 @@ class MarketDataSource(ABC): it reads from the cache. Lifecycle: - source = create_market_data_source(cache) + source = await create_market_data_source(cache) await source.start(["AAPL", "GOOGL", ...]) # ... app runs ... await source.add_ticker("TSLA") @@ -55,3 +57,15 @@ async def remove_ticker(self, ticker: str) -> None: @abstractmethod def get_tickers(self) -> list[str]: """Return the current list of actively tracked tickers.""" + + @abstractmethod + def describe(self) -> SourceStatus: + """Introspection for GET /api/health. Must never raise.""" + + async def get_history(self, ticker: str, points: int = 120) -> list[PricePoint]: + """Historical series for the chart's first paint. + + Default implementation returns [] — the frontend then accumulates from + SSE. Sources with real or synthesized history override this. + """ + return [] diff --git a/backend/app/market/massive_client.py b/backend/app/market/massive_client.py index 00bc7b2aa..5be85d98d 100644 --- a/backend/app/market/massive_client.py +++ b/backend/app/market/massive_client.py @@ -1,18 +1,30 @@ -"""Massive (Polygon.io) API client for real market data.""" +"""Massive (Polygon.io) API client for real market data. + +Only usable by keys entitled to real-time snapshots (Advanced tier or above, +$199/mo). Lower tiers authenticate but get NOT_AUTHORIZED on every snapshot +call — factory.py probes entitlement once at startup and routes those keys to +AnchoredSimulatorDataSource instead. See planning/MASSIVE_API.md. +""" from __future__ import annotations import asyncio +import datetime import logging +import urllib3.exceptions from massive import RESTClient +from massive.exceptions import AuthError, BadResponse from massive.rest.models import SnapshotMarketType from .cache import PriceCache from .interface import MarketDataSource +from .models import PricePoint, SourceStatus, epoch_to_iso logger = logging.getLogger(__name__) +NANOSECONDS_PER_SECOND = 1e9 + class MassiveDataSource(MarketDataSource): """MarketDataSource backed by the Massive (Polygon.io) REST API. @@ -20,27 +32,31 @@ class MassiveDataSource(MarketDataSource): Polls GET /v2/snapshot/locale/us/markets/stocks/tickers for all watched tickers in a single API call, then writes results to the PriceCache. - Rate limits: - - Free tier: 5 req/min → poll every 15s (default) - - Paid tiers: higher limits → poll every 2-5s + `retries=0` on the client deliberately: the SDK's default retry/backoff is + entirely inside the same rate-limit window, so a retry on 429 is guaranteed + to fail too — it just burns budget. Poll-level backoff (the interval itself) + is what matters. """ def __init__( self, api_key: str, price_cache: PriceCache, - poll_interval: float = 15.0, + poll_interval: float = 5.0, ) -> None: self._api_key = api_key self._cache = price_cache self._interval = poll_interval self._tickers: list[str] = [] + self._unknown_tickers: set[str] = set() self._task: asyncio.Task | None = None self._client: RESTClient | None = None + self._live = False + self._last_error: str | None = None async def start(self, tickers: list[str]) -> None: - self._client = RESTClient(api_key=self._api_key) - self._tickers = list(tickers) + self._client = RESTClient(api_key=self._api_key, retries=0) + self._tickers = [t.upper().strip() for t in tickers] # Do an immediate first poll so the cache has data right away await self._poll_once() @@ -72,12 +88,42 @@ async def add_ticker(self, ticker: str) -> None: async def remove_ticker(self, ticker: str) -> None: ticker = ticker.upper().strip() self._tickers = [t for t in self._tickers if t != ticker] + self._unknown_tickers.discard(ticker) self._cache.remove(ticker) logger.info("Massive: removed ticker %s", ticker) def get_tickers(self) -> list[str]: return list(self._tickers) + def describe(self) -> SourceStatus: + detail = self._last_error or "real-time snapshots" + if self._unknown_tickers: + detail += f"; not returned by last poll: {', '.join(sorted(self._unknown_tickers))}" + return SourceStatus( + name="massive", + live=self._live, + detail=detail, + tickers=len(self._tickers), + cache_populated=len(self._cache) > 0, + ) + + async def get_history(self, ticker: str, points: int = 120) -> list[PricePoint]: + """Real intraday minute bars for the current session (paid-tier only).""" + if not self._client: + return [] + try: + bars = await asyncio.to_thread( + self._fetch_today_minute_bars, ticker, points + ) + except Exception as e: + logger.warning("Massive get_history failed for %s: %s", ticker, e) + return [] + return [PricePoint(epoch_to_iso(bar.timestamp / 1000), bar.close) for bar in bars[-points:]] + + def _fetch_today_minute_bars(self, ticker: str, points: int) -> list: + today = datetime.date.today().isoformat() + return self._client.get_aggs(ticker, 1, "minute", today, today, limit=max(points, 1)) + # --- Internal --- async def _poll_loop(self) -> None: @@ -95,30 +141,65 @@ async def _poll_once(self) -> None: # The Massive RESTClient is synchronous — run in a thread to # avoid blocking the event loop. snapshots = await asyncio.to_thread(self._fetch_snapshots) - processed = 0 - for snap in snapshots: - try: - price = snap.last_trade.price - # Massive timestamps are Unix milliseconds → convert to seconds - timestamp = snap.last_trade.timestamp / 1000.0 - self._cache.update( - ticker=snap.ticker, - price=price, - timestamp=timestamp, - ) - processed += 1 - except (AttributeError, TypeError) as e: - logger.warning( - "Skipping snapshot for %s: %s", - getattr(snap, "ticker", "???"), - e, - ) - logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers)) - + except AuthError as e: + self._live = False + self._last_error = f"auth error: {e}" + logger.error("Massive poll failed (auth): %s", e) + return + except urllib3.exceptions.MaxRetryError as e: + # Rate limited or unreachable. Do not retry within this cycle — the + # next scheduled poll is the backoff. + self._live = False + self._last_error = "rate limited or unreachable" + logger.warning("Massive poll rate-limited/unreachable: %s", e) + return + except BadResponse as e: + self._live = False + self._last_error = str(e) + logger.error("Massive poll failed (bad response): %s", e) + return except Exception as e: + self._live = False + self._last_error = str(e) logger.error("Massive poll failed: %s", e) - # Don't re-raise — the loop will retry on the next interval. - # Common failures: 401 (bad key), 429 (rate limit), network errors. + return + + processed = 0 + seen: set[str] = set() + for snap in snapshots: + ticker = getattr(snap, "ticker", None) + if ticker: + seen.add(ticker) + try: + last_trade = snap.last_trade + if last_trade is None: + raise AttributeError("snapshot has no last_trade") + price = last_trade.price + # Massive snapshot timestamps are Unix NANOSECONDS on sip_timestamp + # (not `timestamp`, and not milliseconds — see planning/MASSIVE_API.md §6). + timestamp = last_trade.sip_timestamp / NANOSECONDS_PER_SECOND + open_price = snap.prev_day.close if snap.prev_day else None + self._cache.update( + ticker=ticker, + price=price, + timestamp=timestamp, + open_price=open_price, + ) + processed += 1 + except (AttributeError, TypeError) as e: + logger.warning( + "Skipping snapshot for %s: %s", + ticker or "???", + e, + ) + + # Tickers we asked for but that never appeared in the response — + # surfaced via describe() rather than vanishing silently. + self._unknown_tickers = {t for t in self._tickers if t not in seen} + self._live = processed > 0 + if processed: + self._last_error = None + logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers)) def _fetch_snapshots(self) -> list: """Synchronous call to the Massive REST API. Runs in a thread.""" diff --git a/backend/app/market/models.py b/backend/app/market/models.py index de81b1dbc..1f370cd7f 100644 --- a/backend/app/market/models.py +++ b/backend/app/market/models.py @@ -4,46 +4,111 @@ import time from dataclasses import dataclass, field +from datetime import datetime, timezone + + +def epoch_to_iso(timestamp: float) -> str: + """Convert a Unix epoch (seconds) to an ISO 8601 UTC string, e.g. '...T17:42:11.413700Z'. + + Always includes microseconds (unlike `datetime.isoformat()`, which omits them + when exactly zero) so every timestamp has a fixed-width format — plain string + comparison then agrees with chronological order. + """ + dt = datetime.fromtimestamp(timestamp, tz=timezone.utc) + return dt.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z" @dataclass(frozen=True, slots=True) class PriceUpdate: - """Immutable snapshot of a single ticker's price at a point in time.""" + """Immutable snapshot of a single ticker's price at a point in time. + + `previous_price` is the previous *tick* (drives the flash animation only). + `open_price` is fixed for the session (drives the daily change column) — + the seed price for the simulator, the anchor close for the anchored + simulator, or the previous day's close for Massive. + """ ticker: str price: float previous_price: float + open_price: float timestamp: float = field(default_factory=time.time) # Unix seconds @property def change(self) -> float: - """Absolute price change from previous update.""" + """Absolute price change from the previous tick.""" return round(self.price - self.previous_price, 4) @property def change_percent(self) -> float: - """Percentage change from previous update.""" + """Percentage change from the previous tick.""" 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'.""" + def tick_direction(self) -> str: + """'up', 'down', or 'flat' — drives the CSS flash class.""" if self.price > self.previous_price: return "up" elif self.price < self.previous_price: return "down" return "flat" + @property + def change_today(self) -> float: + """Absolute change since the session open/anchor.""" + return round(self.price - self.open_price, 4) + + @property + def change_percent_today(self) -> float: + """Percentage change since the session open/anchor — drives the daily % column.""" + if self.open_price == 0: + return 0.0 + return round((self.price - self.open_price) / self.open_price * 100, 4) + def to_dict(self) -> dict: - """Serialize for JSON / SSE transmission.""" + """Serialize for JSON / SSE transmission. Timestamp is ISO 8601 UTC on the wire.""" return { "ticker": self.ticker, "price": self.price, "previous_price": self.previous_price, - "timestamp": self.timestamp, + "open_price": self.open_price, + "timestamp": epoch_to_iso(self.timestamp), + "tick_direction": self.tick_direction, "change": self.change, "change_percent": self.change_percent, - "direction": self.direction, + "change_today": self.change_today, + "change_percent_today": self.change_percent_today, + } + + +@dataclass(frozen=True, slots=True) +class PricePoint: + """A single point in a historical price series.""" + + timestamp: str # ISO 8601 UTC + price: float + + def to_dict(self) -> dict: + return {"timestamp": self.timestamp, "price": self.price} + + +@dataclass(frozen=True, slots=True) +class SourceStatus: + """Introspection for a MarketDataSource, surfaced via GET /api/health.""" + + name: str # "simulator" | "massive" | "anchored-simulator" + live: bool # True only when prices reflect the real current market + detail: str # human-readable, surfaced verbatim in /api/health + tickers: int + cache_populated: bool + + def to_dict(self) -> dict: + return { + "name": self.name, + "live": self.live, + "detail": self.detail, + "tickers": self.tickers, + "cache_populated": self.cache_populated, } diff --git a/backend/app/market/seed_prices.py b/backend/app/market/seed_prices.py index 69586df03..9d70a19ff 100644 --- a/backend/app/market/seed_prices.py +++ b/backend/app/market/seed_prices.py @@ -1,17 +1,24 @@ -"""Seed prices and per-ticker parameters for the market simulator.""" +"""Seed prices and per-ticker parameters for the market simulator. -# Realistic starting prices for the default watchlist (as of project creation) +These are the no-key fallback only. When a Massive API key is available, +AnchoredSimulatorDataSource replaces these levels with real closing prices — +see planning/MARKET_INTERFACE.md. Refreshed to real 2026-09-02 closes so the +no-key demo does not show implausible levels (e.g. NVDA and NFLX are badly +stale after their 2024 splits). +""" + +# Starting prices for the default watchlist, sourced from real 2026-09-02 closes. 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, + "AAPL": 324.96, + "GOOGL": 337.12, + "MSFT": 496.82, + "AMZN": 254.98, + "TSLA": 357.01, + "NVDA": 224.41, + "META": 592.85, + "JPM": 356.22, + "V": 378.40, + "NFLX": 82.73, } # Per-ticker GBM parameters diff --git a/backend/app/market/simulator.py b/backend/app/market/simulator.py index b6803f592..2b0946706 100644 --- a/backend/app/market/simulator.py +++ b/backend/app/market/simulator.py @@ -3,14 +3,19 @@ from __future__ import annotations import asyncio +import hashlib import logging import math import random +import time +from collections import deque +from dataclasses import dataclass import numpy as np from .cache import PriceCache from .interface import MarketDataSource +from .models import PricePoint, SourceStatus, epoch_to_iso from .seed_prices import ( CORRELATION_GROUPS, CROSS_GROUP_CORR, @@ -24,6 +29,33 @@ logger = logging.getLogger(__name__) +# Deterministic synthetic seed range for unknown tickers, e.g. $20.00-$500.00. +SYNTHETIC_SEED_MIN = 20.0 +SYNTHETIC_SEED_SPAN_CENTS = 48_000 # (SEED_MAX - SEED_MIN) * 100 + + +def synthesize_seed(ticker: str) -> float: + """Stable pseudo-price for an unknown symbol. Same ticker, same price, always. + + Unlike a random draw, this survives container restarts — a held position's + cost basis and the P&L chart stay consistent instead of jumping around. + """ + digest = int(hashlib.sha256(ticker.encode()).hexdigest()[:8], 16) + return round(SYNTHETIC_SEED_MIN + (digest % SYNTHETIC_SEED_SPAN_CENTS) / 100.0, 2) + + +@dataclass +class _Shock: + """A transient, decaying price overlay simulating a sudden intraday move. + + Modeled as an additive overlay on top of the GBM path (not a permanent + level shift), so a shock is visible as a spike that bleeds off over + roughly a minute without permanently distorting the calibrated volatility. + """ + + magnitude: float # signed fraction, e.g. -0.03 + decay: float = 0.985 # per-tick multiplier; ~half-life 46 ticks (~23s at 500ms) + class GBMSimulator: """Geometric Brownian Motion simulator for correlated stock prices. @@ -47,19 +79,27 @@ class GBMSimulator: TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8 + # Default event probability, calibrated so shocks add drama without + # swamping the calibrated sigma: ~1.2 events per 10-minute demo across a + # 10-ticker watchlist. See planning/MARKET_SIMULATOR.md §6. + DEFAULT_EVENT_PROBABILITY = 1e-4 + def __init__( self, tickers: list[str], dt: float = DEFAULT_DT, - event_probability: float = 0.001, + event_probability: float = DEFAULT_EVENT_PROBABILITY, + seed_overrides: dict[str, float] | None = None, ) -> None: self._dt = dt self._event_prob = event_probability + self._seed_overrides = dict(seed_overrides or {}) # Per-ticker state self._tickers: list[str] = [] self._prices: dict[str, float] = {} self._params: dict[str, dict[str, float]] = {} + self._shocks: dict[str, _Shock] = {} # Cholesky decomposition of the correlation matrix (for correlated moves) self._cholesky: np.ndarray | None = None @@ -96,24 +136,29 @@ def step(self) -> dict[str, float]: sigma = params["sigma"] # GBM: S(t+dt) = S(t) * exp((mu - 0.5*sigma^2)*dt + sigma*sqrt(dt)*Z) + # This is the underlying, correctly-calibrated price. Shocks (below) + # are a separate overlay and never touch this value directly. drift = (mu - 0.5 * sigma**2) * self._dt diffusion = sigma * math.sqrt(self._dt) * z_correlated[i] self._prices[ticker] *= math.exp(drift + diffusion) - # Random event: ~0.1% chance per tick per ticker - # With 10 tickers at 2 ticks/sec, expect an event ~every 50 seconds + # Random event: a transient, decaying overlay — not a permanent + # level shift. See _Shock docstring. 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", - ) + magnitude = random.uniform(0.015, 0.04) * random.choice([-1, 1]) + self._shocks[ticker] = _Shock(magnitude=magnitude) + logger.debug("Shock event on %s: %.2f%%", ticker, magnitude * 100) - result[ticker] = round(self._prices[ticker], 2) + shock = self._shocks.get(ticker) + if shock is not None: + displayed = self._prices[ticker] * (1 + shock.magnitude) + shock.magnitude *= shock.decay + if abs(shock.magnitude) < 1e-4: + del self._shocks[ticker] + else: + displayed = self._prices[ticker] + + result[ticker] = round(displayed, 2) return result @@ -131,6 +176,7 @@ def remove_ticker(self, ticker: str) -> None: self._tickers.remove(ticker) del self._prices[ticker] del self._params[ticker] + self._shocks.pop(ticker, None) self._rebuild_cholesky() def get_price(self, ticker: str) -> float | None: @@ -141,6 +187,16 @@ def get_tickers(self) -> list[str]: """Return the list of currently tracked tickers.""" return list(self._tickers) + def reset_price(self, ticker: str, price: float) -> None: + """Reset a ticker's price and clear any active shock, without touching correlation. + + Used to rewind the simulator after running it forward to prefill history, + so live trading starts from the true seed rather than the prefill's endpoint. + """ + if ticker in self._prices: + self._prices[ticker] = price + self._shocks.pop(ticker, None) + # --- Internals --- def _add_ticker_internal(self, ticker: str) -> None: @@ -148,13 +204,18 @@ def _add_ticker_internal(self, ticker: str) -> None: if ticker in self._prices: return self._tickers.append(ticker) - self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) + seed = self._seed_overrides.get(ticker) + if seed is None: + seed = SEED_PRICES.get(ticker) + if seed is None: + seed = synthesize_seed(ticker) + self._prices[ticker] = seed self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) def _rebuild_cholesky(self) -> None: """Rebuild the Cholesky decomposition of the ticker correlation matrix. - Called whenever tickers are added or removed. O(n^2) but n < 50. + Called whenever tickers are added or removed. O(n^2) but n < 100. """ n = len(self._tickers) if n <= 1: @@ -169,7 +230,19 @@ def _rebuild_cholesky(self) -> None: corr[i, j] = rho corr[j, i] = rho - self._cholesky = np.linalg.cholesky(corr) + try: + self._cholesky = np.linalg.cholesky(corr) + except np.linalg.LinAlgError: + # The shipped correlation structure is verified positive-definite at + # any size (block-equicorrelated, min eigenvalue = 1 - rho_max), but a + # future change to the structure could break that. Degrade to + # independent moves rather than 500ing a watchlist add. + logger.error( + "Correlation matrix not positive-definite for %d tickers; " + "falling back to uncorrelated moves", + n, + ) + self._cholesky = np.eye(n) @staticmethod def _pairwise_correlation(t1: str, t2: str) -> float: @@ -204,28 +277,47 @@ class SimulatorDataSource(MarketDataSource): `update_interval` seconds and writes results to the PriceCache. """ + # 600 ticks x 500ms = 5 minutes of history per ticker. ~480KB for 50 tickers. + HISTORY_POINTS = 600 + def __init__( self, price_cache: PriceCache, update_interval: float = 0.5, - event_probability: float = 0.001, + event_probability: float = GBMSimulator.DEFAULT_EVENT_PROBABILITY, + seed_overrides: dict[str, float] | None = None, + status_detail: str | None = None, ) -> None: self._cache = price_cache self._interval = update_interval self._event_prob = event_probability + self._seed_overrides = dict(seed_overrides or {}) + self._status_detail = status_detail self._sim: GBMSimulator | None = None self._task: asyncio.Task | None = None + self._history: dict[str, deque[tuple[float, float]]] = {} + self._open_prices: dict[str, float] = {} async def start(self, tickers: list[str]) -> None: self._sim = GBMSimulator( tickers=tickers, event_probability=self._event_prob, + seed_overrides=self._seed_overrides, ) + for ticker in tickers: + price = self._sim.get_price(ticker) + if price is not None: + self._open_prices[ticker] = price + + self._prefill_history(tickers) + # Seed the cache with initial prices so SSE has data immediately for ticker in tickers: price = self._sim.get_price(ticker) if price is not None: - self._cache.update(ticker=ticker, price=price) + self._cache.update( + ticker=ticker, price=price, open_price=self._open_prices.get(ticker, price) + ) self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") logger.info("Simulator started with %d tickers", len(tickers)) @@ -242,29 +334,86 @@ async def stop(self) -> None: async def add_ticker(self, ticker: str) -> None: if self._sim: self._sim.add_ticker(ticker) - # Seed cache immediately so the ticker has a price right away price = self._sim.get_price(ticker) if price is not None: - self._cache.update(ticker=ticker, price=price) + self._open_prices.setdefault(ticker, price) + self._history.setdefault(ticker, deque(maxlen=self.HISTORY_POINTS)) + self._cache.update( + ticker=ticker, price=price, open_price=self._open_prices[ticker] + ) 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) + self._history.pop(ticker, None) + self._open_prices.pop(ticker, None) logger.info("Simulator: removed ticker %s", ticker) def get_tickers(self) -> list[str]: return self._sim.get_tickers() if self._sim else [] + def describe(self) -> SourceStatus: + return SourceStatus( + name="simulator", + live=False, + detail=self._status_detail or "synthetic GBM simulation", + tickers=len(self.get_tickers()), + cache_populated=len(self._cache) > 0, + ) + + def set_status_detail(self, detail: str | None) -> None: + """Allow a wrapping source (e.g. AnchoredSimulatorDataSource) to update the + health detail after re-anchoring, without exposing the private attribute.""" + self._status_detail = detail + + async def get_history(self, ticker: str, points: int = 120) -> list[PricePoint]: + series = self._history.get(ticker, ()) + return [PricePoint(epoch_to_iso(t), p) for t, p in list(series)[-points:]] + 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() + now = time.time() for ticker, price in prices.items(): - self._cache.update(ticker=ticker, price=price) + self._cache.update( + ticker=ticker, + price=price, + open_price=self._open_prices.get(ticker, price), + ) + hist = self._history.setdefault( + ticker, deque(maxlen=self.HISTORY_POINTS) + ) + hist.append((now, price)) except Exception: - logger.exception("Simulator step failed") + logger.exception("Simulator step failed") # never let one bad tick kill the loop await asyncio.sleep(self._interval) + + def _prefill_history(self, tickers: list[str]) -> None: + """Run the simulator forward HISTORY_POINTS steps with no sleeping, recording + the path, then reset every price to its seed. So the very first chart paint has + five minutes of plausible history instead of a single dot. + """ + if self._sim is None or not tickers: + return + + now = time.time() + paths: dict[str, deque[tuple[float, float]]] = { + t: deque(maxlen=self.HISTORY_POINTS) for t in tickers + } + for i in range(self.HISTORY_POINTS): + prices = self._sim.step() + ts = now - (self.HISTORY_POINTS - i) * self._interval + for ticker in tickers: + if ticker in prices: + paths[ticker].append((ts, prices[ticker])) + + for ticker in tickers: + self._history[ticker] = paths[ticker] + seed = self._open_prices.get(ticker) + if seed is not None: + self._sim.reset_price(ticker, seed) diff --git a/backend/app/market/stream.py b/backend/app/market/stream.py index 7fd974b7c..14b96c3e2 100644 --- a/backend/app/market/stream.py +++ b/backend/app/market/stream.py @@ -5,6 +5,7 @@ import asyncio import json import logging +import time from collections.abc import AsyncGenerator from fastapi import APIRouter, Request @@ -16,6 +17,12 @@ router = APIRouter(prefix="/api/stream", tags=["streaming"]) +# How long the stream may stay silent before sending an SSE comment ping. +# Matters most in Massive mode: a 15s poll interval (free tier) or a closed +# market can otherwise leave the connection silent long enough for a proxy or +# a laptop sleep to drop it without either side noticing. +HEARTBEAT_INTERVAL_SECONDS = 10.0 + def create_stream_router(price_cache: PriceCache) -> APIRouter: """Create the SSE streaming router with a reference to the price cache. @@ -52,16 +59,22 @@ async def _generate_events( price_cache: PriceCache, request: Request, interval: float = 0.5, + heartbeat_interval: float = HEARTBEAT_INTERVAL_SECONDS, ) -> AsyncGenerator[str, None]: """Async generator that yields SSE-formatted price events. - Sends all prices every `interval` seconds. Stops when the client - disconnects (detected via request.is_disconnected()). + Sends the full price map whenever the cache version changes (deliberately + not a delta protocol — at 10-50 tickers the payload is a couple of KB, and + a delta would need reconnection-resync logic for no measurable gain). + Sends a ": ping" comment if nothing has changed for `heartbeat_interval` + seconds, so proxies and idle connections don't silently die. Stops when + the client disconnects (detected via request.is_disconnected()). """ # Tell the client to retry after 1 second if the connection drops yield "retry: 1000\n\n" last_version = -1 + last_sent = time.monotonic() client_ip = request.client.host if request.client else "unknown" logger.info("SSE client connected: %s", client_ip) @@ -81,6 +94,10 @@ async def _generate_events( data = {ticker: update.to_dict() for ticker, update in prices.items()} payload = json.dumps(data) yield f"data: {payload}\n\n" + last_sent = time.monotonic() + elif time.monotonic() - last_sent > heartbeat_interval: + yield ": ping\n\n" # SSE comment — ignored by EventSource, keeps the pipe alive + last_sent = time.monotonic() await asyncio.sleep(interval) except asyncio.CancelledError: diff --git a/backend/market_data_demo.py b/backend/market_data_demo.py index 7414416c4..5a617311d 100644 --- a/backend/market_data_demo.py +++ b/backend/market_data_demo.py @@ -79,10 +79,10 @@ def build_table( continue # Direction styling - if update.direction == "up": + if update.tick_direction == "up": color = "green" arrow = "[bold green]\u25b2[/]" - elif update.direction == "down": + elif update.tick_direction == "down": color = "red" arrow = "[bold red]\u25bc[/]" else: @@ -248,8 +248,8 @@ async def run() -> None: # Log notable moves if abs(update.change_percent) > 1.0: - direction = "\u25b2" if update.direction == "up" else "\u25bc" - color = "green" if update.direction == "up" else "red" + direction = "\u25b2" if update.tick_direction == "up" else "\u25bc" + color = "green" if update.tick_direction == "up" else "red" timestamp = time.strftime("%H:%M:%S") events.appendleft( f"[bright_black]{timestamp}[/] " diff --git a/backend/tests/market/test_anchored.py b/backend/tests/market/test_anchored.py new file mode 100644 index 000000000..e7b1160d3 --- /dev/null +++ b/backend/tests/market/test_anchored.py @@ -0,0 +1,235 @@ +"""Tests for AnchoredSimulatorDataSource. No test hits the network.""" + +from unittest.mock import MagicMock, patch + +import pytest + +from app.market.anchored import AnchoredSimulatorDataSource +from app.market.cache import PriceCache +from app.market.models import SourceStatus + + +def _bar(ticker: str, close: float) -> MagicMock: + bar = MagicMock() + bar.ticker = ticker + bar.close = close + return bar + + +@pytest.mark.asyncio +class TestAnchoredSimulatorDataSource: + async def test_start_anchors_to_grouped_daily_close(self): + """A single get_grouped_daily_aggs call should price the whole watchlist.""" + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.return_value = [ + _bar("AAPL", 324.96), + _bar("GOOGL", 337.12), + _bar("OTHERTICKER", 12.34), + ] + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0 + ) + await source.start(["AAPL", "GOOGL"]) + + assert source._anchors == {"AAPL": 324.96, "GOOGL": 337.12} + assert mock_client.get_grouped_daily_aggs.call_count == 1 + + await source.stop() + + async def test_anchored_prices_seed_the_cache(self): + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)] + cache = PriceCache() + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=cache, reanchor_interval=0 + ) + await source.start(["AAPL"]) + + update = cache.get("AAPL") + assert update.price == 324.96 + assert update.open_price == 324.96 + + await source.stop() + + async def test_fetch_anchors_walks_back_over_a_weekend(self): + """Saturday and Sunday return empty results; Friday's close should be used.""" + mock_client = MagicMock() + # today() - 1 day = Saturday (empty), -2 = Friday (populated) + mock_client.get_grouped_daily_aggs.side_effect = [ + [], # Saturday + [_bar("AAPL", 324.96)], # Friday + ] + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache() + ) + anchors, anchor_date = source._fetch_anchors(["AAPL"]) + + assert anchors == {"AAPL": 324.96} + assert anchor_date is not None + assert mock_client.get_grouped_daily_aggs.call_count == 2 + + async def test_fetch_anchors_falls_back_after_repeated_failures(self): + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.side_effect = Exception("network error") + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache() + ) + anchors, anchor_date = source._fetch_anchors(["AAPL"]) + + assert anchors == {} + assert anchor_date is None + assert mock_client.get_grouped_daily_aggs.call_count == 7 # MAX_ANCHOR_LOOKBACK_DAYS + + async def test_start_falls_back_to_static_seeds_when_anchor_fetch_fails(self): + """If Massive is unreachable at startup, the simulator must still start + (with the static seed table) rather than dead-ending the app.""" + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.side_effect = Exception("network error") + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0 + ) + await source.start(["AAPL"]) + + assert "AAPL" in source.get_tickers() + assert source.describe().detail == "simulated from static seed prices (anchor fetch failed)" + + await source.stop() + + async def test_describe_reports_anchor_date_and_count(self): + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)] + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0 + ) + await source.start(["AAPL"]) + + status = source.describe() + assert isinstance(status, SourceStatus) + assert status.name == "anchored-simulator" + assert status.live is False # simulated, never presented as live + assert "1 anchored" in status.detail + + await source.stop() + + async def test_add_ticker_delegates_to_underlying_simulator(self): + """Adding an unanchored symbol must never dead-end — the underlying + simulator synthesizes a seed for it.""" + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)] + cache = PriceCache() + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=cache, reanchor_interval=0 + ) + await source.start(["AAPL"]) + await source.add_ticker("ZZZZ") + + assert "ZZZZ" in source.get_tickers() + assert cache.get("ZZZZ") is not None + + await source.stop() + + async def test_remove_ticker_delegates_to_underlying_simulator(self): + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)] + cache = PriceCache() + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=cache, reanchor_interval=0 + ) + await source.start(["AAPL"]) + await source.remove_ticker("AAPL") + + assert "AAPL" not in source.get_tickers() + assert cache.get("AAPL") is None + + await source.stop() + + async def test_get_history_uses_real_minute_bars(self): + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)] + minute_bar = MagicMock(timestamp=1757000000000, close=325.10) + mock_client.get_aggs.return_value = [minute_bar] + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0 + ) + await source.start(["AAPL"]) + history = await source.get_history("AAPL") + + assert len(history) == 1 + assert history[0].price == 325.10 + + await source.stop() + + async def test_get_history_empty_when_no_anchor_date(self): + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.side_effect = Exception("network error") + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0 + ) + await source.start(["AAPL"]) + history = await source.get_history("AAPL") + + assert history == [] + + await source.stop() + + async def test_stop_is_idempotent(self): + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)] + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0 + ) + await source.start(["AAPL"]) + await source.stop() + await source.stop() # should not raise + + async def test_reanchor_task_started_when_interval_positive(self): + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)] + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache(), reanchor_interval=3600.0 + ) + await source.start(["AAPL"]) + + assert source._reanchor_task is not None + assert not source._reanchor_task.done() + + await source.stop() + assert source._reanchor_task is None + + async def test_no_reanchor_task_when_interval_zero(self): + mock_client = MagicMock() + mock_client.get_grouped_daily_aggs.return_value = [_bar("AAPL", 324.96)] + + with patch("app.market.anchored.RESTClient", return_value=mock_client): + source = AnchoredSimulatorDataSource( + api_key="basic-key", price_cache=PriceCache(), reanchor_interval=0 + ) + await source.start(["AAPL"]) + + assert source._reanchor_task is None + + await source.stop() diff --git a/backend/tests/market/test_cache.py b/backend/tests/market/test_cache.py index b5ab3d55d..81b537a91 100644 --- a/backend/tests/market/test_cache.py +++ b/backend/tests/market/test_cache.py @@ -15,26 +15,32 @@ def test_update_and_get(self): assert cache.get("AAPL") == update def test_first_update_is_flat(self): - """Test that the first update has flat direction.""" + """Test that the first update has flat tick_direction.""" cache = PriceCache() update = cache.update("AAPL", 190.50) - assert update.direction == "flat" + assert update.tick_direction == "flat" assert update.previous_price == 190.50 + def test_first_update_open_price_defaults_to_price(self): + """On first write, open_price defaults to price when not given.""" + cache = PriceCache() + update = cache.update("AAPL", 190.50) + assert update.open_price == 190.50 + def test_direction_up(self): - """Test price update with upward direction.""" + """Test price update with upward tick_direction.""" cache = PriceCache() cache.update("AAPL", 190.00) update = cache.update("AAPL", 191.00) - assert update.direction == "up" + assert update.tick_direction == "up" assert update.change == 1.00 def test_direction_down(self): - """Test price update with downward direction.""" + """Test price update with downward tick_direction.""" cache = PriceCache() cache.update("AAPL", 190.00) update = cache.update("AAPL", 189.00) - assert update.direction == "down" + assert update.tick_direction == "down" assert update.change == -1.00 def test_remove(self): @@ -101,3 +107,29 @@ def test_price_rounding(self): cache = PriceCache() update = cache.update("AAPL", 190.12345) assert update.price == 190.12 + + def test_explicit_open_price(self): + """Test that an explicit open_price is stored as given.""" + cache = PriceCache() + update = cache.update("AAPL", 190.50, open_price=180.00) + assert update.open_price == 180.00 + + def test_open_price_persists_across_updates(self): + """open_price should stay fixed for the session unless explicitly changed.""" + cache = PriceCache() + cache.update("AAPL", 190.00, open_price=180.00) + update = cache.update("AAPL", 195.00) # no open_price given + assert update.open_price == 180.00 + + def test_open_price_can_be_updated_explicitly(self): + """A source may explicitly move the baseline (e.g. Massive's prev_day.close).""" + cache = PriceCache() + cache.update("AAPL", 190.00, open_price=180.00) + update = cache.update("AAPL", 195.00, open_price=185.00) + assert update.open_price == 185.00 + + def test_open_price_rounded(self): + """Test that open_price is rounded to 2 decimal places.""" + cache = PriceCache() + update = cache.update("AAPL", 190.00, open_price=180.126) + assert update.open_price == 180.13 diff --git a/backend/tests/market/test_capabilities.py b/backend/tests/market/test_capabilities.py new file mode 100644 index 000000000..de411c4f4 --- /dev/null +++ b/backend/tests/market/test_capabilities.py @@ -0,0 +1,111 @@ +"""Tests for probe_capabilities — the Massive entitlement probe. + +No test hits the network: the RESTClient constructor and its methods are +always mocked. +""" + +from unittest.mock import MagicMock, patch + +import urllib3.exceptions +from massive.exceptions import AuthError, BadResponse + +from app.market.capabilities import probe_capabilities + + +class TestProbeCapabilities: + def test_no_key_is_invalid(self): + with patch("app.market.capabilities.RESTClient", side_effect=AuthError("no key")): + caps = probe_capabilities("") + + assert caps.valid is False + assert caps.realtime is False + assert caps.end_of_day is False + + def test_realtime_entitled_key(self): + """Advanced tier: snapshot call succeeds outright.""" + mock_client = MagicMock() + mock_client.get_snapshot_all.return_value = [] + + with patch("app.market.capabilities.RESTClient", return_value=mock_client): + caps = probe_capabilities("advanced-key") + + assert caps.valid is True + assert caps.realtime is True + assert caps.end_of_day is True + + def test_end_of_day_only_key(self): + """Basic/free tier: snapshot NOT_AUTHORIZED, but aggregate call succeeds. + This is the tier nearly every student will have — the whole point of the + capability probe is to detect it accurately.""" + mock_client = MagicMock() + mock_client.get_snapshot_all.side_effect = BadResponse( + "NOT_AUTHORIZED: Please upgrade your plan" + ) + mock_client.get_previous_close_agg.return_value = [MagicMock(close=324.96)] + + with patch("app.market.capabilities.RESTClient", return_value=mock_client): + caps = probe_capabilities("basic-key") + + assert caps.valid is True + assert caps.realtime is False + assert caps.end_of_day is True + assert "end-of-day" in caps.detail.lower() + + def test_invalid_key_rejected_by_both_calls(self): + mock_client = MagicMock() + mock_client.get_snapshot_all.side_effect = BadResponse("NOT_AUTHORIZED") + mock_client.get_previous_close_agg.side_effect = AuthError("invalid key") + + with patch("app.market.capabilities.RESTClient", return_value=mock_client): + caps = probe_capabilities("bad-key") + + assert caps.valid is False + assert caps.realtime is False + assert caps.end_of_day is False + assert "rejected" in caps.detail.lower() + + def test_snapshot_error_not_not_authorized_is_treated_as_unexpected(self): + """A BadResponse that isn't NOT_AUTHORIZED (e.g. a genuine server error) + should not be silently reinterpreted as an entitlement gap.""" + mock_client = MagicMock() + mock_client.get_snapshot_all.side_effect = BadResponse("INTERNAL_SERVER_ERROR") + + with patch("app.market.capabilities.RESTClient", return_value=mock_client): + caps = probe_capabilities("some-key") + + assert caps.valid is False + assert caps.realtime is False + assert caps.end_of_day is False + + def test_rate_limited_on_first_call(self): + """MaxRetryError (rate limiting), not BadResponse — planning/MASSIVE_API.md §7.""" + mock_client = MagicMock() + error = urllib3.exceptions.MaxRetryError(pool=MagicMock(), url="/x", reason="429") + mock_client.get_snapshot_all.side_effect = error + + with patch("app.market.capabilities.RESTClient", return_value=mock_client): + caps = probe_capabilities("some-key") + + assert caps.valid is False + assert "rate" in caps.detail.lower() or "unreachable" in caps.detail.lower() + + def test_never_raises(self): + """probe_capabilities must never raise — the factory awaits it directly + with no try/except of its own.""" + mock_client = MagicMock() + mock_client.get_snapshot_all.side_effect = RuntimeError("something exploded") + + with patch("app.market.capabilities.RESTClient", return_value=mock_client): + caps = probe_capabilities("some-key") # must not raise + + assert caps.valid is False + + def test_probe_uses_no_retries(self): + mock_client = MagicMock() + mock_client.get_snapshot_all.return_value = [] + + with patch("app.market.capabilities.RESTClient", return_value=mock_client) as mock_cls: + probe_capabilities("some-key") + + _, kwargs = mock_cls.call_args + assert kwargs.get("retries") == 0 diff --git a/backend/tests/market/test_factory.py b/backend/tests/market/test_factory.py index 5ff5dd49e..e04d93039 100644 --- a/backend/tests/market/test_factory.py +++ b/backend/tests/market/test_factory.py @@ -1,79 +1,131 @@ -"""Tests for market data source factory.""" +"""Tests for the market data source factory.""" import os from unittest.mock import patch +import pytest + +from app.market.anchored import AnchoredSimulatorDataSource from app.market.cache import PriceCache +from app.market.capabilities import MassiveCapabilities from app.market.factory import create_market_data_source from app.market.massive_client import MassiveDataSource from app.market.simulator import SimulatorDataSource +@pytest.mark.asyncio class TestFactory: - """Tests for create_market_data_source factory.""" + """Tests for create_market_data_source. Never hits the network — probe_capabilities + is always mocked.""" - def test_creates_simulator_when_no_api_key(self): - """Test that simulator is created when MASSIVE_API_KEY is not set.""" + async def test_creates_simulator_when_no_api_key(self): + """No key → simulator, and the (network-calling) probe must not even run.""" cache = PriceCache() with patch.dict(os.environ, {}, clear=True): - source = create_market_data_source(cache) + with patch("app.market.factory.probe_capabilities") as mock_probe: + source = await create_market_data_source(cache) assert isinstance(source, SimulatorDataSource) + mock_probe.assert_not_called() - def test_creates_simulator_when_api_key_empty(self): - """Test that simulator is created when MASSIVE_API_KEY is empty.""" + async def test_creates_simulator_when_api_key_empty(self): cache = PriceCache() with patch.dict(os.environ, {"MASSIVE_API_KEY": ""}, clear=True): - source = create_market_data_source(cache) + with patch("app.market.factory.probe_capabilities"): + source = await create_market_data_source(cache) assert isinstance(source, SimulatorDataSource) - def test_creates_simulator_when_api_key_whitespace(self): - """Test that simulator is created when MASSIVE_API_KEY is whitespace.""" + async def test_creates_simulator_when_api_key_whitespace(self): cache = PriceCache() with patch.dict(os.environ, {"MASSIVE_API_KEY": " "}, clear=True): - source = create_market_data_source(cache) + with patch("app.market.factory.probe_capabilities"): + source = await create_market_data_source(cache) assert isinstance(source, SimulatorDataSource) - def test_creates_massive_when_api_key_set(self): - """Test that Massive client is created when MASSIVE_API_KEY is set.""" + async def test_creates_massive_when_realtime_entitled(self): cache = PriceCache() + caps = MassiveCapabilities(True, True, True, "real-time snapshots entitled") with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True): - source = create_market_data_source(cache) + with patch("app.market.factory.probe_capabilities", return_value=caps): + source = await create_market_data_source(cache) assert isinstance(source, MassiveDataSource) + assert source._api_key == "test-key" - def test_massive_receives_api_key(self): - """Test that Massive client receives the API key.""" + async def test_creates_anchored_simulator_when_end_of_day_only(self): + """The free-tier case: a valid key that cannot return live prices must + route to the anchored simulator, not a MassiveDataSource that would + poll forever and never populate the cache.""" cache = PriceCache() + caps = MassiveCapabilities(True, False, True, "end-of-day only (Basic tier)") - with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key-123"}, clear=True): - source = create_market_data_source(cache) + with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True): + with patch("app.market.factory.probe_capabilities", return_value=caps): + with patch("app.market.anchored.RESTClient"): + source = await create_market_data_source(cache) - assert isinstance(source, MassiveDataSource) - assert source._api_key == "test-key-123" + assert isinstance(source, AnchoredSimulatorDataSource) + + async def test_falls_back_to_simulator_when_key_rejected(self): + """An invalid/revoked key must never boot into a broken state — it + degrades to the simulator instead of a source that produces nothing.""" + cache = PriceCache() + caps = MassiveCapabilities(False, False, False, "key rejected: 401") - def test_simulator_receives_cache(self): - """Test that simulator receives the cache reference.""" + with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True): + with patch("app.market.factory.probe_capabilities", return_value=caps): + source = await create_market_data_source(cache) + + assert isinstance(source, SimulatorDataSource) + assert "key rejected" in source.describe().detail + + async def test_probe_called_with_the_configured_key(self): + cache = PriceCache() + caps = MassiveCapabilities(True, True, True, "real-time snapshots entitled") + + with patch.dict(os.environ, {"MASSIVE_API_KEY": "specific-key-123"}, clear=True): + with patch( + "app.market.factory.probe_capabilities", return_value=caps + ) as mock_probe: + await create_market_data_source(cache) + + mock_probe.assert_called_once_with("specific-key-123") + + async def test_simulator_receives_cache(self): cache = PriceCache() with patch.dict(os.environ, {}, clear=True): - source = create_market_data_source(cache) + source = await create_market_data_source(cache) assert isinstance(source, SimulatorDataSource) assert source._cache is cache - def test_massive_receives_cache(self): - """Test that Massive client receives the cache reference.""" + async def test_massive_receives_cache(self): cache = PriceCache() + caps = MassiveCapabilities(True, True, True, "real-time snapshots entitled") with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True): - source = create_market_data_source(cache) + with patch("app.market.factory.probe_capabilities", return_value=caps): + source = await create_market_data_source(cache) assert isinstance(source, MassiveDataSource) assert source._cache is cache + + async def test_never_raises_on_probe_exception_paths(self): + """probe_capabilities itself never raises (it catches everything), but + the factory must still resolve to a usable source for every capability + combination it can return.""" + cache = PriceCache() + caps = MassiveCapabilities(False, False, False, "unreachable/rate-limited: timeout") + + with patch.dict(os.environ, {"MASSIVE_API_KEY": "test-key"}, clear=True): + with patch("app.market.factory.probe_capabilities", return_value=caps): + source = await create_market_data_source(cache) + + assert isinstance(source, SimulatorDataSource) diff --git a/backend/tests/market/test_interface.py b/backend/tests/market/test_interface.py new file mode 100644 index 000000000..cbd698b12 --- /dev/null +++ b/backend/tests/market/test_interface.py @@ -0,0 +1,60 @@ +"""Tests for the MarketDataSource abstract contract.""" + +import pytest + +from app.market.interface import MarketDataSource + + +class _MinimalSource(MarketDataSource): + """Implements only the abstract members, to exercise the default get_history().""" + + async def start(self, tickers): + pass + + async def stop(self): + pass + + async def add_ticker(self, ticker): + pass + + async def remove_ticker(self, ticker): + pass + + def get_tickers(self): + return [] + + def describe(self): + from app.market.models import SourceStatus + + return SourceStatus(name="minimal", live=False, detail="", tickers=0, cache_populated=False) + + +class TestMarketDataSource: + def test_cannot_instantiate_directly(self): + with pytest.raises(TypeError): + MarketDataSource() + + def test_subclass_missing_describe_cannot_instantiate(self): + class _Incomplete(MarketDataSource): + async def start(self, tickers): + pass + + async def stop(self): + pass + + async def add_ticker(self, ticker): + pass + + async def remove_ticker(self, ticker): + pass + + def get_tickers(self): + return [] + + with pytest.raises(TypeError): + _Incomplete() + + @pytest.mark.asyncio + async def test_default_get_history_returns_empty_list(self): + source = _MinimalSource() + assert await source.get_history("AAPL") == [] diff --git a/backend/tests/market/test_massive.py b/backend/tests/market/test_massive.py index cdd7dbd24..2e7b23f17 100644 --- a/backend/tests/market/test_massive.py +++ b/backend/tests/market/test_massive.py @@ -1,20 +1,30 @@ -"""Tests for MassiveDataSource (mocked).""" +"""Tests for MassiveDataSource (mocked). No test hits the network.""" from unittest.mock import MagicMock, patch import pytest +import urllib3.exceptions +from massive.exceptions import AuthError, BadResponse from app.market.cache import PriceCache from app.market.massive_client import MassiveDataSource +from app.market.models import SourceStatus -def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock: - """Create a mock Massive snapshot object.""" +def _make_snapshot( + ticker: str, price: float, sip_timestamp_ns: int, prev_close: float | None = None +) -> MagicMock: + """Create a mock Massive snapshot object matching the real SDK's field names.""" snap = MagicMock() snap.ticker = ticker snap.last_trade = MagicMock() snap.last_trade.price = price - snap.last_trade.timestamp = timestamp_ms + snap.last_trade.sip_timestamp = sip_timestamp_ns + if prev_close is not None: + snap.prev_day = MagicMock() + snap.prev_day.close = prev_close + else: + snap.prev_day = None return snap @@ -34,8 +44,8 @@ async def test_poll_updates_cache(self): source._client = MagicMock() # Satisfy the _poll_once guard mock_snapshots = [ - _make_snapshot("AAPL", 190.50, 1707580800000), - _make_snapshot("GOOGL", 175.25, 1707580800000), + _make_snapshot("AAPL", 190.50, 1707580800_000000000), + _make_snapshot("GOOGL", 175.25, 1707580800_000000000), ] with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): @@ -44,6 +54,57 @@ async def test_poll_updates_cache(self): assert cache.get_price("AAPL") == 190.50 assert cache.get_price("GOOGL") == 175.25 + async def test_nanosecond_timestamp_conversion(self): + """Massive snapshot timestamps are sip_timestamp in NANOSECONDS, not + `timestamp` in milliseconds. Regression guard for planning/MASSIVE_API.md + §6: a naive /1000.0 on the wrong field is wrong by a factor of a million.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + mock_snapshots = [_make_snapshot("AAPL", 190.50, 1605192894630916600)] + + with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): + await source._poll_once() + + update = cache.get("AAPL") + assert update.timestamp == pytest.approx(1605192894.63, abs=0.01) + + async def test_open_price_captured_from_prev_day_close(self): + """Regression guard: the daily change column has nowhere to get its + baseline from unless prev_day.close is captured as open_price.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + mock_snapshots = [ + _make_snapshot("AAPL", 325.41, 1707580800_000000000, prev_close=324.96) + ] + + with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): + await source._poll_once() + + update = cache.get("AAPL") + assert update.open_price == 324.96 + + async def test_missing_prev_day_leaves_open_price_defaulted(self): + """When prev_day is unavailable, open_price should fall back to the + cache's own default (price on first write) rather than raising.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800_000000000)] + + with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): + await source._poll_once() + + update = cache.get("AAPL") + assert update.open_price == 190.50 + async def test_malformed_snapshot_skipped(self): """Test that malformed snapshots are skipped gracefully.""" cache = PriceCache() @@ -55,7 +116,7 @@ async def test_malformed_snapshot_skipped(self): source._tickers = ["AAPL", "BAD"] source._client = MagicMock() # Satisfy the _poll_once guard - good_snap = _make_snapshot("AAPL", 190.50, 1707580800000) + good_snap = _make_snapshot("AAPL", 190.50, 1707580800_000000000) bad_snap = MagicMock() bad_snap.ticker = "BAD" bad_snap.last_trade = None # Will cause AttributeError @@ -67,8 +128,51 @@ async def test_malformed_snapshot_skipped(self): assert cache.get_price("AAPL") == 190.50 assert cache.get_price("BAD") is None + async def test_auth_error_marks_not_live(self): + cache = PriceCache() + source = MassiveDataSource(api_key="bad-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + with patch.object(source, "_fetch_snapshots", side_effect=AuthError("invalid key")): + await source._poll_once() + + assert source.describe().live is False + assert "auth" in source.describe().detail.lower() + + async def test_rate_limit_error_classified_distinctly(self): + """Rate limiting arrives as urllib3.MaxRetryError, not BadResponse — + planning/MASSIVE_API.md §7. A handler written as `except BadResponse` + alone would miss it entirely.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + error = urllib3.exceptions.MaxRetryError(pool=MagicMock(), url="/x", reason="429") + with patch.object(source, "_fetch_snapshots", side_effect=error): + await source._poll_once() # must not raise + + status = source.describe() + assert status.live is False + assert "rate limit" in status.detail.lower() or "unreachable" in status.detail.lower() + + async def test_bad_response_error_does_not_crash(self): + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + with patch.object( + source, "_fetch_snapshots", side_effect=BadResponse("NOT_AUTHORIZED") + ): + await source._poll_once() # Should not raise + + assert cache.get_price("AAPL") is None + assert source.describe().live is False + async def test_api_error_does_not_crash(self): - """Test that API errors don't crash the poller.""" + """Test that unexpected errors don't crash the poller.""" cache = PriceCache() source = MassiveDataSource( api_key="test-key", @@ -83,25 +187,33 @@ async def test_api_error_does_not_crash(self): assert cache.get_price("AAPL") is None # No update happened - async def test_timestamp_conversion(self): - """Test that timestamps are converted from milliseconds to seconds.""" + async def test_successful_poll_marks_live(self): cache = PriceCache() - source = MassiveDataSource( - api_key="test-key", - price_cache=cache, - poll_interval=60.0, - ) + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) source._tickers = ["AAPL"] - source._client = MagicMock() # Satisfy the _poll_once guard + source._client = MagicMock() - mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)] + with patch.object( + source, "_fetch_snapshots", return_value=[_make_snapshot("AAPL", 190.50, 1)] + ): + await source._poll_once() - with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): + assert source.describe().live is True + + async def test_unknown_tickers_tracked(self): + """Tickers requested but absent from the response should be surfaced, + not silently dropped — planning/MARKET_INTERFACE.md §7.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL", "ZZZZ"] + source._client = MagicMock() + + with patch.object( + source, "_fetch_snapshots", return_value=[_make_snapshot("AAPL", 190.50, 1)] + ): await source._poll_once() - update = cache.get("AAPL") - assert update is not None - assert update.timestamp == 1707580800.0 # Converted to seconds + assert "ZZZZ" in source.describe().detail async def test_add_ticker(self): """Test adding a ticker.""" @@ -189,7 +301,7 @@ async def test_start_immediate_poll(self): cache = PriceCache() source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) - mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)] + mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800_000000000)] with patch("app.market.massive_client.RESTClient"): with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): @@ -199,3 +311,27 @@ async def test_start_immediate_poll(self): assert cache.get_price("AAPL") == 190.50 await source.stop() + + async def test_start_constructs_client_with_no_retries(self): + """retries=0 is deliberate: the SDK's default backoff is inside the + same rate-limit window, so a retry on 429 is guaranteed to fail too — + it only burns budget. planning/MASSIVE_API.md §7.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + + with patch("app.market.massive_client.RESTClient") as mock_client_cls: + with patch.object(source, "_fetch_snapshots", return_value=[]): + await source.start(["AAPL"]) + + _, kwargs = mock_client_cls.call_args + assert kwargs.get("retries") == 0 + + await source.stop() + + async def test_describe_before_start(self): + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache) + status = source.describe() + assert isinstance(status, SourceStatus) + assert status.name == "massive" + assert status.live is False diff --git a/backend/tests/market/test_models.py b/backend/tests/market/test_models.py index 21600dfd6..dafa749af 100644 --- a/backend/tests/market/test_models.py +++ b/backend/tests/market/test_models.py @@ -1,8 +1,8 @@ -"""Tests for PriceUpdate dataclass.""" +"""Tests for market data models: PriceUpdate, PricePoint, SourceStatus.""" import pytest -from app.market.models import PriceUpdate +from app.market.models import PricePoint, PriceUpdate, SourceStatus, epoch_to_iso class TestPriceUpdate: @@ -10,68 +10,207 @@ class TestPriceUpdate: def test_price_update_creation(self): """Test basic PriceUpdate creation.""" - update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", + price=190.50, + previous_price=190.00, + open_price=189.00, + timestamp=1234567890.0, + ) assert update.ticker == "AAPL" assert update.price == 190.50 assert update.previous_price == 190.00 + assert update.open_price == 189.00 assert update.timestamp == 1234567890.0 def test_change_calculation(self): - """Test price change calculation.""" - update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0) + """Test tick-to-tick price change calculation.""" + update = PriceUpdate( + ticker="AAPL", price=190.50, previous_price=190.00, open_price=190.00 + ) assert update.change == 0.50 def test_change_negative(self): - """Test negative price change.""" - update = PriceUpdate(ticker="AAPL", price=189.50, previous_price=190.00, timestamp=1234567890.0) + """Test negative tick-to-tick price change.""" + update = PriceUpdate( + ticker="AAPL", price=189.50, previous_price=190.00, open_price=190.00 + ) assert update.change == -0.50 def test_change_percent_up(self): - """Test percentage change calculation (up).""" - update = PriceUpdate(ticker="AAPL", price=190.00, previous_price=100.00, timestamp=1234567890.0) + """Test tick-to-tick percentage change calculation (up).""" + update = PriceUpdate( + ticker="AAPL", price=190.00, previous_price=100.00, open_price=100.00 + ) assert update.change_percent == 90.0 def test_change_percent_down(self): - """Test percentage change calculation (down).""" - update = PriceUpdate(ticker="AAPL", price=100.00, previous_price=200.00, timestamp=1234567890.0) + """Test tick-to-tick percentage change calculation (down).""" + update = PriceUpdate( + ticker="AAPL", price=100.00, previous_price=200.00, open_price=200.00 + ) assert update.change_percent == -50.0 def test_change_percent_zero_previous(self): """Test percentage change with zero previous price.""" - update = PriceUpdate(ticker="AAPL", price=100.00, previous_price=0.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=100.00, previous_price=0.00, open_price=100.00 + ) assert update.change_percent == 0.0 - def test_direction_up(self): - """Test direction calculation (up).""" - update = PriceUpdate(ticker="AAPL", price=191.00, previous_price=190.00, timestamp=1234567890.0) - assert update.direction == "up" - - def test_direction_down(self): - """Test direction calculation (down).""" - update = PriceUpdate(ticker="AAPL", price=189.00, previous_price=190.00, timestamp=1234567890.0) - assert update.direction == "down" - - def test_direction_flat(self): - """Test direction calculation (flat).""" - update = PriceUpdate(ticker="AAPL", price=190.00, previous_price=190.00, timestamp=1234567890.0) - assert update.direction == "flat" + def test_tick_direction_up(self): + """Test tick_direction calculation (up).""" + update = PriceUpdate( + ticker="AAPL", price=191.00, previous_price=190.00, open_price=190.00 + ) + assert update.tick_direction == "up" + + def test_tick_direction_down(self): + """Test tick_direction calculation (down).""" + update = PriceUpdate( + ticker="AAPL", price=189.00, previous_price=190.00, open_price=190.00 + ) + assert update.tick_direction == "down" + + def test_tick_direction_flat(self): + """Test tick_direction calculation (flat).""" + update = PriceUpdate( + ticker="AAPL", price=190.00, previous_price=190.00, open_price=190.00 + ) + assert update.tick_direction == "flat" + + def test_change_today(self): + """Test the daily change is measured against open_price, not previous tick.""" + update = PriceUpdate( + ticker="AAPL", price=195.00, previous_price=194.99, open_price=190.00 + ) + assert update.change_today == 5.00 + + def test_change_percent_today(self): + """Test the daily percentage change is measured against open_price.""" + update = PriceUpdate( + ticker="AAPL", price=209.00, previous_price=208.99, open_price=190.00 + ) + assert update.change_percent_today == pytest.approx(10.0, abs=0.001) + + def test_change_percent_today_zero_open(self): + """Test daily percentage change with a zero open_price (guard against ZeroDivisionError).""" + update = PriceUpdate( + ticker="AAPL", price=100.00, previous_price=100.00, open_price=0.00 + ) + assert update.change_percent_today == 0.0 + + def test_change_today_independent_of_tick_change(self): + """A ticker can be flat tick-to-tick but still up/down on the day.""" + update = PriceUpdate( + ticker="AAPL", price=200.00, previous_price=200.00, open_price=190.00 + ) + assert update.tick_direction == "flat" + assert update.change_today == 10.00 + assert update.change_percent_today == pytest.approx(5.263, abs=0.001) def test_to_dict(self): """Test serialization to dictionary.""" - update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", + price=190.50, + previous_price=190.00, + open_price=180.00, + timestamp=1234567890.0, + ) result = update.to_dict() assert result["ticker"] == "AAPL" assert result["price"] == 190.50 assert result["previous_price"] == 190.00 - assert result["timestamp"] == 1234567890.0 + assert result["open_price"] == 180.00 + assert result["timestamp"] == epoch_to_iso(1234567890.0) assert result["change"] == 0.50 assert result["change_percent"] == 0.2632 # (0.50 / 190.00) * 100 - assert result["direction"] == "up" + assert result["tick_direction"] == "up" + assert result["change_today"] == 10.50 + assert "change_percent_today" in result + + def test_to_dict_timestamp_is_iso_string(self): + """Wire format is ISO 8601 UTC, not a raw float, per PLAN.md §13.1 item 6.""" + update = PriceUpdate( + ticker="AAPL", + price=190.50, + previous_price=190.00, + open_price=190.00, + timestamp=1234567890.0, + ) + result = update.to_dict() + assert isinstance(result["timestamp"], str) + assert result["timestamp"].endswith("Z") def test_immutability(self): """Test that PriceUpdate is immutable.""" - update = PriceUpdate(ticker="AAPL", price=190.50, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=190.50, previous_price=190.00, open_price=190.00 + ) with pytest.raises(AttributeError): update.price = 200.00 # Should raise error + + +class TestEpochToIso: + """Unit tests for the epoch_to_iso wire-format helper.""" + + def test_returns_utc_z_suffix(self): + assert epoch_to_iso(0).startswith("1970-01-01T00:00:00") + assert epoch_to_iso(0).endswith("Z") + + def test_preserves_subsecond_precision(self): + result = epoch_to_iso(1234567890.4137) + assert "413700" in result or "413" in result + + +class TestPricePoint: + """Unit tests for the PricePoint model.""" + + def test_creation(self): + point = PricePoint(timestamp="2026-09-03T17:42:11Z", price=325.41) + assert point.timestamp == "2026-09-03T17:42:11Z" + assert point.price == 325.41 + + def test_to_dict(self): + point = PricePoint(timestamp="2026-09-03T17:42:11Z", price=325.41) + assert point.to_dict() == {"timestamp": "2026-09-03T17:42:11Z", "price": 325.41} + + def test_immutability(self): + point = PricePoint(timestamp="2026-09-03T17:42:11Z", price=325.41) + with pytest.raises(AttributeError): + point.price = 100.0 + + +class TestSourceStatus: + """Unit tests for the SourceStatus model.""" + + def test_creation(self): + status = SourceStatus( + name="simulator", live=False, detail="synthetic", tickers=10, cache_populated=True + ) + assert status.name == "simulator" + assert status.live is False + assert status.tickers == 10 + assert status.cache_populated is True + + def test_to_dict(self): + status = SourceStatus( + name="massive", live=True, detail="real-time", tickers=5, cache_populated=True + ) + assert status.to_dict() == { + "name": "massive", + "live": True, + "detail": "real-time", + "tickers": 5, + "cache_populated": True, + } + + def test_immutability(self): + status = SourceStatus( + name="simulator", live=False, detail="x", tickers=0, cache_populated=False + ) + with pytest.raises(AttributeError): + status.live = True diff --git a/backend/tests/market/test_simulator.py b/backend/tests/market/test_simulator.py index 1845ec16b..2b3492d9f 100644 --- a/backend/tests/market/test_simulator.py +++ b/backend/tests/market/test_simulator.py @@ -1,7 +1,13 @@ """Tests for GBMSimulator.""" -from app.market.seed_prices import SEED_PRICES -from app.market.simulator import GBMSimulator +import math +import random + +import numpy as np +import pytest + +from app.market.seed_prices import SEED_PRICES, TICKER_PARAMS +from app.market.simulator import GBMSimulator, synthesize_seed class TestGBMSimulator: @@ -52,13 +58,6 @@ def test_remove_nonexistent_is_noop(self): sim = GBMSimulator(tickers=["AAPL"]) sim.remove_ticker("NOPE") # Should not raise - def test_unknown_ticker_gets_random_seed_price(self): - """Test that unknown tickers get random seed prices.""" - sim = GBMSimulator(tickers=["ZZZZ"]) - price = sim.get_price("ZZZZ") - assert price is not None - assert 50.0 <= price <= 300.0 - def test_empty_step(self): """Test stepping with no tickers.""" sim = GBMSimulator(tickers=[]) @@ -89,6 +88,34 @@ def test_cholesky_none_with_one_ticker(self): sim = GBMSimulator(tickers=["AAPL"]) assert sim._cholesky is None + def test_cholesky_survives_many_unknown_tickers(self): + """The correlation matrix must stay positive-definite as tickers are added + one at a time (the incremental path, not the batch constructor) — a failure + here would 500 a watchlist add.""" + sim = GBMSimulator(tickers=["AAPL"]) + for i in range(100): + sim.add_ticker(f"ZZ{i:03d}") + assert sim._cholesky is not None + result = sim.step() + assert len(result) == 101 + + def test_cholesky_linalg_error_falls_back_to_identity(self, monkeypatch): + """A non-positive-definite matrix must degrade to uncorrelated moves, + not raise and crash the caller (e.g. a watchlist add).""" + + def raise_linalg_error(_matrix): + raise np.linalg.LinAlgError("not positive definite") + + sim = GBMSimulator(tickers=["AAPL"]) + monkeypatch.setattr(np.linalg, "cholesky", raise_linalg_error) + sim.add_ticker("GOOGL") # triggers _rebuild_cholesky with 2 tickers + + assert sim._cholesky is not None + np.testing.assert_array_equal(sim._cholesky, np.eye(2)) + # Simulator must still be usable. + result = sim.step() + assert set(result.keys()) == {"AAPL", "GOOGL"} + def test_get_price_returns_none_for_unknown(self): """Test that get_price returns None for unknown ticker.""" sim = GBMSimulator(tickers=["AAPL"]) @@ -120,12 +147,261 @@ def test_default_dt_is_reasonable(self): """Test that default dt is a reasonable small value.""" assert 0 < GBMSimulator.DEFAULT_DT < 0.0001 + def test_default_event_probability_is_calibrated(self): + """Regression guard: event_probability must stay at the calibrated 1e-4, + not the old 0.001 which inflated realized volatility ~20x + (planning/MARKET_SIMULATOR.md §6).""" + assert GBMSimulator.DEFAULT_EVENT_PROBABILITY == 1e-4 + def test_prices_rounded_to_two_decimals(self): """Test that prices are rounded to 2 decimal places.""" sim = GBMSimulator(tickers=["AAPL"]) result = sim.step() price_str = str(result["AAPL"]) # Check that we have at most 2 decimal places - if '.' in price_str: - decimal_part = price_str.split('.')[1] + if "." in price_str: + decimal_part = price_str.split(".")[1] assert len(decimal_part) <= 2 + + def test_reset_price(self): + """reset_price rewinds the price and clears any active shock.""" + sim = GBMSimulator(tickers=["AAPL"]) + sim.step() + sim.reset_price("AAPL", 999.99) + assert sim.get_price("AAPL") == 999.99 + assert "AAPL" not in sim._shocks + + def test_reset_price_unknown_ticker_is_noop(self): + sim = GBMSimulator(tickers=["AAPL"]) + sim.reset_price("NOPE", 1.0) # should not raise + + +class TestSynthesizeSeed: + """Unit tests for deterministic unknown-ticker seed synthesis. + + Regression guard for planning/MARKET_SIMULATOR.md §3: unknown tickers used + to get `random.uniform(50, 300)`, a different price every restart, which + made a held position's cost basis jump and lied to the P&L chart. + """ + + def test_deterministic_across_calls(self): + assert synthesize_seed("ZZZZ") == synthesize_seed("ZZZZ") + + def test_deterministic_across_fresh_instances(self): + """Same ticker, same price, across two independently constructed simulators.""" + sim1 = GBMSimulator(tickers=["ZZZZ"]) + sim2 = GBMSimulator(tickers=["ZZZZ"]) + assert sim1.get_price("ZZZZ") == sim2.get_price("ZZZZ") + + def test_different_tickers_differ(self): + # Not a hard guarantee for arbitrary hashes, but true for these symbols + # and a good smoke test that the hash actually depends on the ticker. + assert synthesize_seed("ZZZZ") != synthesize_seed("QQQQ") + + def test_in_expected_range(self): + for ticker in ["ZZZZ", "QQQQ", "FOOBAR", "X"]: + price = synthesize_seed(ticker) + assert 20.0 <= price <= 500.0 + + def test_unknown_ticker_uses_synthesized_seed(self): + sim = GBMSimulator(tickers=["ZZZZ"]) + assert sim.get_price("ZZZZ") == synthesize_seed("ZZZZ") + + def test_unknown_ticker_uses_default_params(self): + sim = GBMSimulator(tickers=["ZZZZ"]) + assert sim._params["ZZZZ"]["sigma"] == 0.25 + assert sim._params["ZZZZ"]["mu"] == 0.05 + + +class TestSeedOverrides: + """Unit tests for anchoring the simulator to real (Massive-sourced) prices.""" + + def test_seed_override_takes_priority_over_static_table(self): + sim = GBMSimulator(tickers=["AAPL"], seed_overrides={"AAPL": 999.99}) + assert sim.get_price("AAPL") == 999.99 + + def test_seed_override_used_for_unknown_ticker(self): + sim = GBMSimulator(tickers=["ZZZZ"], seed_overrides={"ZZZZ": 42.42}) + assert sim.get_price("ZZZZ") == 42.42 + + def test_missing_override_falls_back_to_static_table(self): + sim = GBMSimulator(tickers=["AAPL", "GOOGL"], seed_overrides={"AAPL": 999.99}) + assert sim.get_price("GOOGL") == SEED_PRICES["GOOGL"] + + def test_missing_override_and_missing_table_entry_synthesizes(self): + sim = GBMSimulator(tickers=["ZZZZ"], seed_overrides={"AAPL": 999.99}) + assert sim.get_price("ZZZZ") == synthesize_seed("ZZZZ") + + +class TestShockDynamics: + """Unit tests for the decaying shock overlay. + + Regression guard for the historical bug: shocks used to be a *permanent* + multiplicative level shift, which — at the old event_probability — inflated + realized volatility roughly 20x and made every per-ticker sigma decorative + (planning/MARKET_SIMULATOR.md §6). Shocks must now decay back toward the + underlying (still-calibrated) GBM path. + """ + + def test_shock_is_applied_on_trigger(self, monkeypatch): + monkeypatch.setattr(random, "random", lambda: 0.0) # always triggers + monkeypatch.setattr(random, "uniform", lambda a, b: 0.04) + monkeypatch.setattr(random, "choice", lambda seq: 1) + + sim = GBMSimulator(tickers=["AAPL"], event_probability=1.0) + seed = sim.get_price("AAPL") + result = sim.step() + + assert "AAPL" in sim._shocks + # Displayed price reflects the +4% shock (approximately, modulo the + # underlying GBM drift/diffusion for this one tick). + assert result["AAPL"] > seed + + def test_underlying_price_is_never_permanently_shifted_by_a_shock(self, monkeypatch): + """The GBM path (`_prices`) must stay independent of the shock overlay — + only the *displayed* price reflects the shock.""" + monkeypatch.setattr(random, "random", lambda: 0.0) + monkeypatch.setattr(random, "uniform", lambda a, b: 0.04) + monkeypatch.setattr(random, "choice", lambda seq: 1) + + sim = GBMSimulator(tickers=["AAPL"], event_probability=1.0) + sim.step() + underlying = sim._prices["AAPL"] + displayed = sim.step()["AAPL"] + + # The shock inflates the displayed price above the underlying GBM price. + assert displayed != round(underlying, 2) + + def test_shock_decays_and_does_not_grow(self, monkeypatch): + monkeypatch.setattr(random, "random", lambda: 0.0) + monkeypatch.setattr(random, "uniform", lambda a, b: 0.04) + monkeypatch.setattr(random, "choice", lambda seq: 1) + + sim = GBMSimulator(tickers=["AAPL"], event_probability=1.0) + sim.step() + initial_magnitude = abs(sim._shocks["AAPL"].magnitude) + + # Stop triggering new shocks; let the existing one decay. + sim._event_prob = 0.0 + for _ in range(50): + sim.step() + + remaining = sim._shocks.get("AAPL") + if remaining is not None: + assert abs(remaining.magnitude) < initial_magnitude + + def test_shock_eventually_expires(self, monkeypatch): + monkeypatch.setattr(random, "random", lambda: 0.0) + monkeypatch.setattr(random, "uniform", lambda a, b: 0.03) + monkeypatch.setattr(random, "choice", lambda seq: 1) + + sim = GBMSimulator(tickers=["AAPL"], event_probability=1.0) + sim.step() + sim._event_prob = 0.0 # no further triggers + for _ in range(500): + sim.step() + + assert "AAPL" not in sim._shocks + + +class TestStatisticalCalibration: + """Lighter-weight statistical regression tests. + + These are smoke tests, not the full empirical study in + planning/MARKET_SIMULATOR.md §2/§6 (200 one-day runs is too slow for a + unit test) — but they use generous, one-sided bounds specifically chosen + to catch a regression back to the old, uncalibrated event process + (0.001 probability, permanent shifts), which produced roughly a + 10%-per-hour realized sd against a ~0.5% theoretical target. + """ + + def test_pure_gbm_daily_volatility_is_within_theory(self): + """With shocks disabled, realized log-return sd over many short runs + should track sigma * sqrt(steps * dt).""" + np.random.seed(42) + sigma = TICKER_PARAMS["AAPL"]["sigma"] + steps = 2000 + runs = 60 + + log_returns = [] + for _ in range(runs): + sim = GBMSimulator(tickers=["AAPL"], event_probability=0.0) + start = sim.get_price("AAPL") + for _ in range(steps): + sim.step() + end = sim.get_price("AAPL") + log_returns.append(math.log(end / start)) + + realized_sd = float(np.std(log_returns)) + theoretical_sd = sigma * math.sqrt(steps * GBMSimulator.DEFAULT_DT) + + # Generous tolerance band (statistical test, not an exact bound). + assert theoretical_sd * 0.4 < realized_sd < theoretical_sd * 2.5 + + def test_default_event_probability_does_not_blow_up_volatility(self): + """The historical bug inflated realized volatility ~20x. At the + calibrated default, realized sd should stay within a small multiple of + the pure-GBM theoretical value — nowhere near the old ~20x blowup.""" + np.random.seed(7) + random.seed(7) + sigma = TICKER_PARAMS["AAPL"]["sigma"] + steps = 3600 # ~30 minutes of trading time + runs = 40 + + log_returns = [] + for _ in range(runs): + sim = GBMSimulator(tickers=["AAPL"]) # production default event_probability + start = sim.get_price("AAPL") + for _ in range(steps): + sim.step() + end = sim.get_price("AAPL") + log_returns.append(math.log(end / start)) + + realized_sd = float(np.std(log_returns)) + theoretical_sd = sigma * math.sqrt(steps * GBMSimulator.DEFAULT_DT) + + # The old bug produced ~20x theory; the fix should stay under ~6x even + # with sampling noise from a modest run count. + assert realized_sd < theoretical_sd * 6 + + def test_correlation_is_applied_not_bypassed(self): + """Realized correlation between two tech tickers should be + substantially higher than between a tech and a finance ticker — + confirms the Cholesky decomposition is actually wired in.""" + np.random.seed(11) + steps = 3000 + + sim = GBMSimulator(tickers=["AAPL", "GOOGL", "JPM"], event_probability=0.0) + paths = {"AAPL": [], "GOOGL": [], "JPM": []} + for _ in range(steps): + prices = sim.step() + for t in paths: + paths[t].append(prices[t]) + + def log_returns(series): + arr = np.array(series, dtype=float) + return np.diff(np.log(arr)) + + tech_corr = float(np.corrcoef(log_returns(paths["AAPL"]), log_returns(paths["GOOGL"]))[0, 1]) + cross_corr = float(np.corrcoef(log_returns(paths["AAPL"]), log_returns(paths["JPM"]))[0, 1]) + + assert tech_corr > cross_corr + assert tech_corr > 0.3 + + +class TestVisibleTickRate: + """Regression guard for planning/MARKET_SIMULATOR.md §2: at real anchored + price levels, the 2-decimal displayed price must actually change on most + ticks, or the flash animation and sparklines look dead.""" + + @pytest.mark.parametrize("ticker", list(SEED_PRICES.keys())) + def test_visible_tick_rate_exceeds_fifty_percent(self, ticker): + np.random.seed(3) + sim = GBMSimulator(tickers=[ticker], event_probability=0.0) + prices = [sim.get_price(ticker)] + for _ in range(1500): + prices.append(sim.step()[ticker]) + + visible = sum(1 for a, b in zip(prices, prices[1:]) if a != b) + rate = visible / (len(prices) - 1) + assert rate > 0.5, f"{ticker} visible-tick rate {rate:.2%} too low" diff --git a/backend/tests/market/test_simulator_source.py b/backend/tests/market/test_simulator_source.py index 515ce7290..6940da371 100644 --- a/backend/tests/market/test_simulator_source.py +++ b/backend/tests/market/test_simulator_source.py @@ -5,6 +5,7 @@ import pytest from app.market.cache import PriceCache +from app.market.models import SourceStatus from app.market.simulator import SimulatorDataSource @@ -136,3 +137,149 @@ async def test_custom_event_probability(self): # Just verify it starts and stops cleanly await asyncio.sleep(0.2) await source.stop() + + async def test_open_price_seeded_from_start_price(self): + """The first cache write's open_price should be the session seed.""" + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + + update = cache.get("AAPL") + assert update.open_price == update.price # nothing has moved yet + + await source.stop() + + async def test_open_price_stable_across_ticks(self): + """open_price must stay fixed for the session even as price ticks.""" + cache = PriceCache() + source = SimulatorDataSource( + price_cache=cache, update_interval=0.02, event_probability=0.0 + ) + await source.start(["AAPL"]) + first_open = cache.get("AAPL").open_price + + await asyncio.sleep(0.15) + + assert cache.get("AAPL").open_price == first_open + await source.stop() + + async def test_seed_overrides_used_for_open_price(self): + """Anchored (real-close) seeds should be used verbatim as the session open.""" + cache = PriceCache() + source = SimulatorDataSource( + price_cache=cache, update_interval=0.1, seed_overrides={"AAPL": 500.00} + ) + await source.start(["AAPL"]) + + update = cache.get("AAPL") + assert update.price == 500.00 + assert update.open_price == 500.00 + + await source.stop() + + async def test_describe_returns_source_status(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL", "GOOGL"]) + + status = source.describe() + assert isinstance(status, SourceStatus) + assert status.name == "simulator" + assert status.live is False + assert status.tickers == 2 + assert status.cache_populated is True + + await source.stop() + + async def test_describe_before_start_reports_empty(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache) + status = source.describe() + assert status.tickers == 0 + assert status.cache_populated is False + + async def test_describe_uses_custom_status_detail(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, status_detail="key rejected: bad key") + status = source.describe() + assert status.detail == "key rejected: bad key" + + async def test_set_status_detail_updates_describe(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache) + source.set_status_detail("re-anchored") + assert source.describe().detail == "re-anchored" + + async def test_get_history_empty_before_start(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache) + history = await source.get_history("AAPL") + assert history == [] + + async def test_get_history_prefilled_on_start(self): + """The chart must not be empty on first paint — start() prefills a + ring buffer of history rather than waiting for live ticks.""" + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + + history = await source.get_history("AAPL") + assert len(history) > 1 + # Timestamps should be non-decreasing and end at/near "now". + assert all(a.timestamp <= b.timestamp for a, b in zip(history, history[1:])) + + await source.stop() + + async def test_get_history_respects_points_limit(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + + history = await source.get_history("AAPL", points=5) + assert len(history) == 5 + + await source.stop() + + async def test_prefill_resets_live_price_to_seed(self): + """The prefill run must not leave the simulator's live price at wherever + the prefill's random walk happened to end up.""" + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + + update = cache.get("AAPL") + assert update.price == update.open_price + + await source.stop() + + async def test_get_history_unknown_ticker_is_empty(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + + assert await source.get_history("NOPE") == [] + + await source.stop() + + async def test_add_ticker_history_starts_tracking(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.05) + await source.start(["AAPL"]) + + await source.add_ticker("TSLA") + await asyncio.sleep(0.2) + + history = await source.get_history("TSLA") + assert len(history) >= 1 + + await source.stop() + + async def test_remove_ticker_clears_history(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + await source.remove_ticker("AAPL") + + assert await source.get_history("AAPL") == [] + + await source.stop() diff --git a/backend/tests/market/test_stream.py b/backend/tests/market/test_stream.py new file mode 100644 index 000000000..5b1be6065 --- /dev/null +++ b/backend/tests/market/test_stream.py @@ -0,0 +1,121 @@ +"""Tests for the SSE streaming generator. + +planning/MARKET_INTERFACE.md §8 (PLAN.md §13.4 item 31) drops the Playwright +"disconnect and verify reconnection" E2E case in favor of covering the server +side here — an EventSource's auto-reconnect is a browser behavior we can't +usefully re-test, but the heartbeat and disconnect-detection logic in +`_generate_events` are ours to verify directly. +""" + +from unittest.mock import MagicMock + +import pytest + +from app.market.cache import PriceCache +from app.market.stream import _generate_events + + +def _make_request(disconnected_after: int | None = None) -> MagicMock: + """A fake Request whose is_disconnected() returns True after N calls.""" + request = MagicMock() + request.client = MagicMock(host="127.0.0.1") + + calls = {"n": 0} + + async def is_disconnected(): + calls["n"] += 1 + if disconnected_after is None: + return False + return calls["n"] > disconnected_after + + request.is_disconnected = is_disconnected + return request + + +@pytest.mark.asyncio +class TestGenerateEvents: + async def test_first_event_is_retry_directive(self): + cache = PriceCache() + request = _make_request(disconnected_after=0) + + events = [e async for e in _generate_events(cache, request, interval=0.01)] + assert events[0] == "retry: 1000\n\n" + + async def test_sends_data_when_cache_populated(self): + cache = PriceCache() + cache.update("AAPL", 190.50) + request = _make_request(disconnected_after=1) + + events = [e async for e in _generate_events(cache, request, interval=0.01)] + data_events = [e for e in events if e.startswith("data:")] + assert len(data_events) == 1 + assert "AAPL" in data_events[0] + + async def test_no_data_event_when_cache_empty(self): + cache = PriceCache() + request = _make_request(disconnected_after=1) + + events = [e async for e in _generate_events(cache, request, interval=0.01)] + data_events = [e for e in events if e.startswith("data:")] + assert data_events == [] + + async def test_stops_on_disconnect(self): + cache = PriceCache() + request = _make_request(disconnected_after=0) + + events = [e async for e in _generate_events(cache, request, interval=0.01)] + # Only the initial retry directive — the loop must exit before sleeping/looping. + assert events == ["retry: 1000\n\n"] + + async def test_only_sends_once_per_version_change(self): + """Re-fetching an unchanged cache must not re-emit the same data event.""" + cache = PriceCache() + cache.update("AAPL", 190.50) + request = _make_request(disconnected_after=3) + + events = [e async for e in _generate_events(cache, request, interval=0.01)] + data_events = [e for e in events if e.startswith("data:")] + assert len(data_events) == 1 + + async def test_heartbeat_sent_after_silence(self): + """No price change for longer than heartbeat_interval → an SSE comment ping, + so a proxy or idle Massive poll doesn't silently drop the connection.""" + cache = PriceCache() + request = _make_request(disconnected_after=2) + + events = [ + e + async for e in _generate_events( + cache, request, interval=0.01, heartbeat_interval=0.0 + ) + ] + assert any(e == ": ping\n\n" for e in events) + + async def test_no_heartbeat_before_interval_elapses(self): + cache = PriceCache() + request = _make_request(disconnected_after=2) + + events = [ + e + async for e in _generate_events( + cache, request, interval=0.01, heartbeat_interval=1000.0 + ) + ] + assert not any(e == ": ping\n\n" for e in events) + + async def test_new_data_resets_heartbeat_clock(self): + """A version bump should count as activity — no ping should immediately + follow a fresh data event.""" + cache = PriceCache() + cache.update("AAPL", 190.50) + request = _make_request(disconnected_after=1) + + events = [ + e + async for e in _generate_events( + cache, request, interval=0.01, heartbeat_interval=0.0 + ) + ] + # First event after retry is the data event, not a ping, even though + # heartbeat_interval is 0 — a version change takes priority. + assert events[1].startswith("data:") diff --git a/planning/MARKET_DATA_SUMMARY.md b/planning/MARKET_DATA_SUMMARY.md new file mode 100644 index 000000000..346e9b83c --- /dev/null +++ b/planning/MARKET_DATA_SUMMARY.md @@ -0,0 +1,53 @@ +# Market Data — Summary + +**Status:** Complete. Implements the design in `MARKET_INTERFACE.md`, `MARKET_SIMULATOR.md`, +and `MASSIVE_API.md`. Lives entirely in `backend/app/market/`, with tests in +`backend/tests/market/`. + +## What's There + +- **`PriceCache`** (`cache.py`) — thread-safe in-memory store, one `PriceUpdate` per ticker, + with a monotonic `version` counter for SSE change detection. +- **`PriceUpdate` / `PricePoint` / `SourceStatus`** (`models.py`) — `PriceUpdate` carries both + a tick-to-tick delta (`previous_price`, `change`, `change_percent`, `tick_direction` — drives + the flash animation) and a session-baseline delta (`open_price`, `change_today`, + `change_percent_today` — drives the daily % column). Wire format is ISO 8601 UTC. +- **`MarketDataSource`** (`interface.py`) — the shared abstract contract: `start`/`stop`/ + `add_ticker`/`remove_ticker`/`get_tickers`, plus `describe()` (health introspection) and + `get_history()` (chart first-paint backfill). +- **Three implementations**, selected by `create_market_data_source()` (`factory.py`, async) + based on a one-time Massive entitlement probe (`capabilities.py`) rather than key presence + alone — a free/Basic-tier key authenticates but can't return a live price, so it is routed + away from `MassiveDataSource` instead of silently producing an empty watchlist: + - **`SimulatorDataSource`** (`simulator.py`) — no key. GBM price motion, correlated across a + sector structure, with a deterministic (hash-based) seed for unknown tickers and a + decaying (not permanent) shock overlay for occasional drama. Keeps a 5-minute ring buffer + per ticker, prefilled at startup so charts aren't empty on first paint. + - **`AnchoredSimulatorDataSource`** (`anchored.py`) — key present but end-of-day only (the + free tier). One API call (`get_grouped_daily_aggs`) anchors the whole watchlist to real + closing prices; the GBM simulator then supplies tick-to-tick motion on top. Re-anchors + hourly. Badged as simulated (`SourceStatus.live == False`). + - **`MassiveDataSource`** (`massive_client.py`) — key entitled to real-time snapshots (paid + tiers). Polls `get_snapshot_all` on an interval; corrected nanosecond `sip_timestamp` + handling, captures `prev_day.close` as `open_price`, classifies rate-limit errors + distinctly from bad responses, and surfaces tickers missing from a poll via `describe()`. + - Any key state that can't be classified falls back to `SimulatorDataSource` — the app never + boots into a broken, empty-watchlist state. +- **SSE streaming** (`stream.py`) — `GET /api/stream/prices`, full price map on every cache + version change, with an SSE comment heartbeat (`: ping`) when the stream would otherwise sit + silent (matters most against a 15s Massive poll interval or a closed market). +- **`seed_prices.py`** — static fallback seeds/volatility/correlation params for the no-key + case, kept current against real closing prices. + +## Testing + +`backend/tests/market/` covers all of the above with mocked Massive calls (no test hits the +network) plus statistical smoke tests for the GBM calibration, the shock decay behavior, and +correlation. Run with `uv run --extra dev pytest -v` from `backend/`. + +## Details + +See `MARKET_INTERFACE.md` (source selection, `PriceUpdate`/`SourceStatus` design, SSE wire +format), `MARKET_SIMULATOR.md` (GBM math, calibration, shock design, seeding), and +`MASSIVE_API.md` (entitlement research, endpoint reference, rate-limit gotchas) for the full +design rationale.