From 446b350c9bc447956bc32455dc5163c8f96cb084 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Mon, 3 Aug 2026 18:54:01 +0000 Subject: [PATCH] Complete the market data backend: session baseline, backfill, heartbeat Implements the one remaining piece of backend/app/market/ called out in PLAN.md section 6/13 and detailed in planning/MARKET_DATA_DESIGN.md: - Session baseline: PriceUpdate/PriceCache gain open_price and change_from_open_percent, pinned on each ticker's first tick and carried forward across later updates. - Bounded per-ticker history (deque, one point/minute) with seed_history()/get_history(), so /api/watchlist can serve populated sparklines on first paint. PriceCache.version now only advances on a real price change, keeping SSE quiet in an idle market. - Deterministic parameter synthesis (SHA-256 of the symbol) for unknown tickers, replacing random.uniform(), which repriced a held position on every restart. - GBMSimulator.backfill_history() manufactures ~60 points of prior history ending at the live price; wired into SimulatorDataSource.start()/add_ticker(). - MassiveDataSource: fixed the nanosecond/millisecond timestamp bug (prices were landing ~50,000 years in the future), added a last_trade -> min -> day quote fallback chain, exponential backoff on poll failure, and per-ticker history backfill via get_aggs(). Pinned massive==2.2.0 to match the documented model shapes. - SSE stream: 15s heartbeat comment frame so the frontend can tell "quiet market" from "backend stalled"; moved APIRouter construction inside create_stream_router() to stop double route registration. - New app/market/tickers.py: single normalize_ticker()/TICKER_PATTERN shared by the manual and LLM watchlist paths (not yet built). - wait_for_price() helper for the just-added-ticker trade race. - Updated/added unit tests across every changed module, including a conformance suite that runs the MarketDataSource lifecycle contract against both the simulator and (mocked) Massive implementations. Co-authored-by: Essam Hasin <123895080+EnigmaticFuel@users.noreply.github.com> --- backend/app/market/__init__.py | 13 +- backend/app/market/cache.py | 131 +++++++++-- backend/app/market/interface.py | 40 ++-- backend/app/market/massive_client.py | 206 ++++++++++++----- backend/app/market/models.py | 37 +++- backend/app/market/seed_prices.py | 51 +++-- backend/app/market/simulator.py | 195 +++++++++------- backend/app/market/stream.py | 49 +++-- backend/app/market/tickers.py | 22 ++ backend/market_data_demo.py | 31 +-- backend/pyproject.toml | 2 +- backend/tests/market/test_cache.py | 138 +++++++++++- backend/tests/market/test_conformance.py | 87 ++++++++ backend/tests/market/test_massive.py | 208 +++++++++++++++--- backend/tests/market/test_models.py | 85 ++++++- backend/tests/market/test_simulator.py | 80 ++++++- backend/tests/market/test_simulator_source.py | 49 +++++ backend/tests/market/test_stream.py | 102 +++++++++ backend/tests/market/test_tickers.py | 62 ++++++ 19 files changed, 1291 insertions(+), 297 deletions(-) create mode 100644 backend/app/market/tickers.py create mode 100644 backend/tests/market/test_conformance.py create mode 100644 backend/tests/market/test_stream.py create mode 100644 backend/tests/market/test_tickers.py diff --git a/backend/app/market/__init__.py b/backend/app/market/__init__.py index 57ad0a121..cf30c4dcc 100644 --- a/backend/app/market/__init__.py +++ b/backend/app/market/__init__.py @@ -3,21 +3,28 @@ Public API: PriceUpdate - Immutable price snapshot dataclass PriceCache - Thread-safe in-memory price store + wait_for_price - Poll the cache for a first tick on a new ticker MarketDataSource - Abstract interface for data providers create_market_data_source - Factory that selects simulator or Massive create_stream_router - FastAPI router factory for SSE endpoint + TICKER_PATTERN - Shared ticker validation regex + normalize_ticker - Uppercase, strip, and validate a ticker symbol """ -from .cache import PriceCache +from .cache import PriceCache, wait_for_price from .factory import create_market_data_source from .interface import MarketDataSource from .models import PriceUpdate from .stream import create_stream_router +from .tickers import TICKER_PATTERN, normalize_ticker __all__ = [ - "PriceUpdate", - "PriceCache", "MarketDataSource", + "PriceCache", + "PriceUpdate", + "TICKER_PATTERN", "create_market_data_source", "create_stream_router", + "normalize_ticker", + "wait_for_price", ] diff --git a/backend/app/market/cache.py b/backend/app/market/cache.py index 4d0215778..20267ec92 100644 --- a/backend/app/market/cache.py +++ b/backend/app/market/cache.py @@ -2,68 +2,123 @@ from __future__ import annotations +import asyncio import time +from collections import deque from threading import Lock from .models import PriceUpdate +HISTORY_POINTS = 60 # Points retained per ticker for sparklines +HISTORY_INTERVAL_SECONDS = 60.0 # Minimum spacing between recorded points + class PriceCache: - """Thread-safe in-memory cache of the latest price for each ticker. + """Latest price, session baseline and recent history for each tracked ticker. - Writers: SimulatorDataSource or MassiveDataSource (one at a time). - Readers: SSE streaming endpoint, portfolio valuation, trade execution. + Writers: exactly one MarketDataSource (simulator or Massive poller). + Readers: SSE stream, portfolio valuation, trade execution, watchlist. """ - def __init__(self) -> None: + def __init__( + self, + history_points: int = HISTORY_POINTS, + history_interval: float = HISTORY_INTERVAL_SECONDS, + ) -> None: self._prices: dict[str, PriceUpdate] = {} + self._history: dict[str, deque[float]] = {} + self._history_at: dict[str, float] = {} + self._history_points = history_points + self._history_interval = history_interval self._lock = Lock() - self._version: int = 0 # Monotonically increasing; bumped on every update + self._version = 0 + + # --- Writing --- def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: - """Record a new price for a ticker. Returns the created PriceUpdate. + """Record a new price. Returns the stored PriceUpdate. - Automatically computes direction and change from the previous price. - If this is the first update for the ticker, previous_price == price (direction='flat'). + Derives previous_price and open_price from the entry being replaced, so + sources stay dumb: they produce a number, the cache supplies the meaning. + The first update for a ticker pins its session baseline — + previous_price == open_price == price, direction 'flat', both changes 0. """ with self._lock: - ts = timestamp or time.time() - prev = self._prices.get(ticker) - previous_price = prev.price if prev else price + ts = time.time() if timestamp is None else timestamp + price = round(price, 2) + previous = self._prices.get(ticker) update = PriceUpdate( ticker=ticker, - price=round(price, 2), - previous_price=round(previous_price, 2), + price=price, + previous_price=previous.price if previous else price, + open_price=previous.open_price if previous else price, timestamp=ts, ) self._prices[ticker] = update - self._version += 1 + + # Version tracks *visible* change. A repeated price refreshes the + # timestamp (so /api/health still sees a live feed) without waking + # every SSE client to re-send an identical payload. + if previous is None or previous.price != price: + self._version += 1 + + self._record_history(ticker, update) return update + def seed_history( + self, ticker: str, prices: list[float], timestamp: float | None = None + ) -> None: + """Install backfilled history for a ticker, replacing anything present. + + Called by a source at startup and when a ticker is added, so sparklines + are populated on first paint rather than filling in over 30 seconds. + """ + with self._lock: + self._history[ticker] = deque( + (round(p, 2) for p in prices[-self._history_points :]), + maxlen=self._history_points, + ) + self._history_at[ticker] = time.time() if timestamp is None else timestamp + + def remove(self, ticker: str) -> None: + """Forget a ticker entirely — price, baseline and history.""" + with self._lock: + self._prices.pop(ticker, None) + self._history.pop(ticker, None) + self._history_at.pop(ticker, None) + + # --- Reading --- + def get(self, ticker: str) -> PriceUpdate | None: - """Get the latest price for a single ticker, or None if unknown.""" with self._lock: return self._prices.get(ticker) + def get_price(self, ticker: str) -> float | None: + update = self.get(ticker) + return update.price if update else None + def get_all(self) -> dict[str, PriceUpdate]: - """Snapshot of all current prices. Returns a shallow copy.""" + """Shallow copy of every current price. Safe to iterate without the lock.""" with self._lock: return dict(self._prices) - def get_price(self, ticker: str) -> float | None: - """Convenience: get just the price float, or None.""" - update = self.get(ticker) - return update.price if update else None + def get_history(self, ticker: str) -> list[float]: + """Recent prices, oldest first, up to history_points. Empty if unknown.""" + with self._lock: + history = self._history.get(ticker) + return list(history) if history else [] - def remove(self, ticker: str) -> None: - """Remove a ticker from the cache (e.g., when removed from watchlist).""" + def newest_timestamp(self) -> float | None: + """Timestamp of the most recently written price, for /api/health.""" with self._lock: - self._prices.pop(ticker, None) + if not self._prices: + return None + return max(update.timestamp for update in self._prices.values()) @property def version(self) -> int: - """Current version counter. Useful for SSE change detection.""" + """Monotonic counter, bumped whenever a price actually changes.""" return self._version def __len__(self) -> int: @@ -73,3 +128,31 @@ def __len__(self) -> int: def __contains__(self, ticker: str) -> bool: with self._lock: return ticker in self._prices + + # --- Internal (callers already hold the lock) --- + + def _record_history(self, ticker: str, update: PriceUpdate) -> None: + history = self._history.get(ticker) + if history is None: + history = self._history[ticker] = deque(maxlen=self._history_points) + + last_at = self._history_at.get(ticker) + if last_at is None or update.timestamp - last_at >= self._history_interval: + history.append(update.price) + self._history_at[ticker] = update.timestamp + + +async def wait_for_price(cache: PriceCache, ticker: str, timeout: float = 2.0) -> float: + """Return the current price, waiting up to `timeout` for a first tick. + + Raises ValueError with a user-facing message if no price arrives. Callers + translate that into a 400 (PLAN.md section 8). + """ + deadline = time.monotonic() + timeout + while True: + price = cache.get_price(ticker) + if price is not None: + return price + if time.monotonic() >= deadline: + raise ValueError(f"No price available for {ticker} yet, please try again") + await asyncio.sleep(0.2) diff --git a/backend/app/market/interface.py b/backend/app/market/interface.py index 0f3b7d8c9..efcfe9943 100644 --- a/backend/app/market/interface.py +++ b/backend/app/market/interface.py @@ -8,50 +8,42 @@ class MarketDataSource(ABC): """Contract for market data providers. - Implementations push price updates into a shared PriceCache on their own - schedule. Downstream code never calls the data source directly for prices — - it reads from the cache. + Implementations push prices into a shared PriceCache on their own schedule. + Downstream code never asks a source for a price — it reads the cache. Lifecycle: source = create_market_data_source(cache) - await source.start(["AAPL", "GOOGL", ...]) - # ... app runs ... + await source.start(["AAPL", "GOOGL", ...]) # cache populated on return await source.add_ticker("TSLA") await source.remove_ticker("GOOGL") - # ... app shutting down ... - await source.stop() + await source.stop() # idempotent """ + @property + @abstractmethod + def source_name(self) -> str: + """Short identifier for logs and /api/health: 'simulator' or 'massive'.""" + @abstractmethod async def start(self, tickers: list[str]) -> None: - """Begin producing price updates for the given tickers. + """Begin producing prices for `tickers`. - Starts a background task that periodically writes to the PriceCache. - Must be called exactly once. Calling start() twice is undefined behavior. + Must populate the cache (prices and seeded history) before returning, so + the first HTTP request never sees an empty cache. Called exactly once. """ @abstractmethod async def stop(self) -> None: - """Stop the background task and release resources. - - Safe to call multiple times. After stop(), the source will not write - to the cache again. - """ + """Stop the background task. Idempotent, and never writes afterwards.""" @abstractmethod async def add_ticker(self, ticker: str) -> None: - """Add a ticker to the active set. No-op if already present. - - The next update cycle will include this ticker. - """ + """Track a ticker. No-op if already tracked. Seeds price and history.""" @abstractmethod async def remove_ticker(self, ticker: str) -> None: - """Remove a ticker from the active set. No-op if not present. - - Also removes the ticker from the PriceCache. - """ + """Stop tracking a ticker and remove it from the cache. No-op if absent.""" @abstractmethod def get_tickers(self) -> list[str]: - """Return the current list of actively tracked tickers.""" + """Currently tracked tickers. Synchronous — called from request handlers.""" diff --git a/backend/app/market/massive_client.py b/backend/app/market/massive_client.py index 00bc7b2aa..cd387589a 100644 --- a/backend/app/market/massive_client.py +++ b/backend/app/market/massive_client.py @@ -1,28 +1,33 @@ -"""Massive (Polygon.io) API client for real market data.""" +"""Massive (Polygon.io) REST client for real market data.""" from __future__ import annotations import asyncio +import datetime as dt import logging +import time from massive import RESTClient from massive.rest.models import SnapshotMarketType -from .cache import PriceCache +from .cache import HISTORY_POINTS, PriceCache from .interface import MarketDataSource logger = logging.getLogger(__name__) +MAX_BACKOFF_MULTIPLIER = 8.0 +BACKFILL_LOOKBACK_DAYS = 7 + class MassiveDataSource(MarketDataSource): - """MarketDataSource backed by the Massive (Polygon.io) REST API. + """MarketDataSource backed by the Massive REST API. - Polls GET /v2/snapshot/locale/us/markets/stocks/tickers for all watched - tickers in a single API call, then writes results to the PriceCache. + Polls /v2/snapshot/locale/us/markets/stocks/tickers for the union of watched + tickers in one request, then writes each result to the PriceCache. Rate limits: - - Free tier: 5 req/min → poll every 15s (default) - - Paid tiers: higher limits → poll every 2-5s + Basic (free): 5 req/min -> poll_interval 15.0 (default) + Paid tiers: unlimited -> poll_interval 2.0-5.0 """ def __init__( @@ -30,44 +35,63 @@ def __init__( api_key: str, price_cache: PriceCache, poll_interval: float = 15.0, + backfill_history: bool = True, ) -> None: self._api_key = api_key self._cache = price_cache self._interval = poll_interval + self._backfill_enabled = backfill_history self._tickers: list[str] = [] - self._task: asyncio.Task | None = None self._client: RESTClient | None = None + self._task: asyncio.Task | None = None + self._backfill_task: asyncio.Task | None = None + self._backoff = 1.0 + + @property + def source_name(self) -> str: + return "massive" + + # --- Lifecycle --- async def start(self, tickers: list[str]) -> None: self._client = RESTClient(api_key=self._api_key) - self._tickers = list(tickers) + self._tickers = [t.upper() for t in tickers] - # Do an immediate first poll so the cache has data right away - await self._poll_once() + await self._log_market_status() + await self._poll_once() # Cache has prices before start() returns self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") + if self._backfill_enabled: + self._backfill_task = asyncio.create_task( + self._backfill_all(list(self._tickers)), name="massive-backfill" + ) logger.info( "Massive poller started: %d tickers, %.1fs interval", - len(tickers), + len(self._tickers), self._interval, ) async def stop(self) -> None: - if self._task and not self._task.done(): - self._task.cancel() - try: - await self._task - except asyncio.CancelledError: - pass + for task in (self._task, self._backfill_task): + if task and not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass self._task = None + self._backfill_task = None self._client = None logger.info("Massive poller stopped") async def add_ticker(self, ticker: str) -> None: ticker = ticker.upper().strip() - if ticker not in self._tickers: - self._tickers.append(ticker) - logger.info("Massive: added ticker %s (will appear on next poll)", ticker) + if ticker in self._tickers: + return + self._tickers.append(ticker) + logger.info("Massive: added ticker %s (will appear on next poll)", ticker) + if self._backfill_enabled: + await self._backfill_one(ticker) async def remove_ticker(self, ticker: str) -> None: ticker = ticker.upper().strip() @@ -78,16 +102,16 @@ async def remove_ticker(self, ticker: str) -> None: def get_tickers(self) -> list[str]: return list(self._tickers) - # --- Internal --- + # --- Polling --- async def _poll_loop(self) -> None: - """Poll on interval. First poll already happened in start().""" + """Poll on interval, widening under sustained failure.""" while True: - await asyncio.sleep(self._interval) + await asyncio.sleep(self._interval * self._backoff) await self._poll_once() async def _poll_once(self) -> None: - """Execute one poll cycle: fetch snapshots, update cache.""" + """Execute one poll cycle. Never raises — the loop must survive every failure.""" if not self._tickers or not self._client: return @@ -95,34 +119,116 @@ 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 Exception as e: - logger.error("Massive poll failed: %s", e) - # Don't re-raise — the loop will retry on the next interval. - # Common failures: 401 (bad key), 429 (rate limit), network errors. + except Exception as exc: + # 401 bad key, 403 not in plan, 429 rate limited, network, timeout. + self._backoff = min(self._backoff * 2, MAX_BACKOFF_MULTIPLIER) + logger.error( + "Massive poll failed (%s); backing off to %.1fs", + exc, + self._interval * self._backoff, + ) + return + + self._backoff = 1.0 + updated = 0 + for snap in snapshots: + quote = self._extract_quote(snap) + if quote is None: + logger.warning( + "No usable price in snapshot for %s", getattr(snap, "ticker", "???") + ) + continue + price, timestamp = quote + self._cache.update(ticker=snap.ticker, price=price, timestamp=timestamp) + updated += 1 + + logger.debug("Massive poll: updated %d/%d tickers", updated, len(self._tickers)) def _fetch_snapshots(self) -> list: - """Synchronous call to the Massive REST API. Runs in a thread.""" + """Synchronous SDK call. Runs on a worker thread.""" + assert self._client is not None return self._client.get_snapshot_all( market_type=SnapshotMarketType.STOCKS, tickers=self._tickers, ) + + @staticmethod + def _extract_quote(snap) -> tuple[float, float] | None: + """Best available price and its Unix-seconds timestamp, or None. + + TIMESTAMP UNITS ARE NOT UNIFORM IN THIS API: + last_trade.timestamp, last_quote.timestamp, snapshot.updated -> NANOseconds + Agg.timestamp (aggregate bars), min.timestamp -> MILLIseconds + + Massive's own sample lastTrade.t of 1605195918306274000 is 2020-11-12 + when divided by 1e9, and out of range for any other unit. Dividing by + 1e3 puts every price roughly 50,000 years in the future, silently. + """ + trade = getattr(snap, "last_trade", None) + if trade is not None and getattr(trade, "price", None): + return float(trade.price), float(trade.timestamp) / 1_000_000_000.0 + + minute = getattr(snap, "min", None) + if minute is not None and getattr(minute, "close", None): + return float(minute.close), float(minute.timestamp) / 1_000.0 + + day = getattr(snap, "day", None) + if day is not None and getattr(day, "close", None): + return float(day.close), time.time() + + return None + + # --- History backfill --- + + async def _backfill_all(self, tickers: list[str]) -> None: + """Seed sparkline history, one ticker at a time, spaced to respect the + rate limit. Off the poll loop: once at startup, once per added ticker. + + On the free tier this shares a 5 req/min budget with the poller, so + sparklines fill in over the first minutes rather than instantly. That is + the correct trade — a burst of ten requests at startup earns a 429 and + no history at all. + """ + for ticker in tickers: + await self._backfill_one(ticker) + await asyncio.sleep(self._interval) + + async def _backfill_one(self, ticker: str) -> None: + if self._client is None: + return + try: + history = await asyncio.to_thread(self._fetch_history, ticker) + except Exception as exc: + logger.warning("History backfill failed for %s: %s", ticker, exc) + return + if history: + self._cache.seed_history(ticker, history) + logger.debug("Backfilled %d history points for %s", len(history), ticker) + + def _fetch_history(self, ticker: str) -> list[float]: + """Most recent ~60 one-minute closes. Synchronous; runs on a thread.""" + assert self._client is not None + today = dt.date.today() + bars = self._client.get_aggs( + ticker=ticker, + multiplier=1, + timespan="minute", + from_=(today - dt.timedelta(days=BACKFILL_LOOKBACK_DAYS)).isoformat(), + to=today.isoformat(), + limit=HISTORY_POINTS, + sort="desc", + ) + return [float(bar.close) for bar in reversed(list(bars))] # Oldest first + + # --- Diagnostics --- + + async def _log_market_status(self) -> None: + """Log whether the market is open. 'Prices are not moving' is the most + likely support question on this path, and this line is the answer.""" + if self._client is None: + return + try: + status = await asyncio.to_thread(self._client.get_market_status) + logger.info("Massive market status: %s", getattr(status, "market", "unknown")) + except Exception as exc: + logger.warning("Could not read market status: %s", exc) diff --git a/backend/app/market/models.py b/backend/app/market/models.py index de81b1dbc..08da701b7 100644 --- a/backend/app/market/models.py +++ b/backend/app/market/models.py @@ -8,42 +8,65 @@ @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. + + Constructed only by PriceCache.update(), which supplies previous_price and + open_price from the entry it is replacing. Sources never build one directly. + """ ticker: str price: float previous_price: float - timestamp: float = field(default_factory=time.time) # Unix seconds + open_price: float + timestamp: float = field(default_factory=time.time) # Unix epoch seconds + + # --- Derived: tick over tick --- @property def change(self) -> float: - """Absolute price change from previous update.""" + """Absolute price change since the previous tick.""" return round(self.price - self.previous_price, 4) @property def change_percent(self) -> float: - """Percentage change from previous update.""" + """Percent change since the previous tick. Drives the flash animation only.""" if self.previous_price == 0: return 0.0 return round((self.price - self.previous_price) / self.previous_price * 100, 4) @property def direction(self) -> str: - """'up', 'down', or 'flat'.""" + """'up', 'down' or 'flat' since the previous tick.""" if self.price > self.previous_price: return "up" - elif self.price < self.previous_price: + if self.price < self.previous_price: return "down" return "flat" + # --- Derived: against the session baseline --- + + @property + def change_from_open(self) -> float: + """Absolute price change since the session open.""" + return round(self.price - self.open_price, 4) + + @property + def change_from_open_percent(self) -> float: + """Percent change since the session open. This is the user-facing 'change %'.""" + if self.open_price == 0: + return 0.0 + return round((self.price - self.open_price) / self.open_price * 100, 4) + def to_dict(self) -> dict: - """Serialize for JSON / SSE transmission.""" + """Serialize for JSON / SSE. Keys match the payload in PLAN.md section 6.""" return { "ticker": self.ticker, "price": self.price, "previous_price": self.previous_price, + "open_price": self.open_price, "timestamp": self.timestamp, "change": self.change, "change_percent": self.change_percent, + "change_from_open_percent": self.change_from_open_percent, "direction": self.direction, } diff --git a/backend/app/market/seed_prices.py b/backend/app/market/seed_prices.py index 69586df03..cc6da7de7 100644 --- a/backend/app/market/seed_prices.py +++ b/backend/app/market/seed_prices.py @@ -1,6 +1,10 @@ -"""Seed prices and per-ticker parameters for the market simulator.""" +"""Seed prices, GBM parameters and correlation groups for the simulator.""" -# Realistic starting prices for the default watchlist (as of project creation) +from __future__ import annotations + +import hashlib + +# Recognisable rather than current. This is a simulation with pretend money. SEED_PRICES: dict[str, float] = { "AAPL": 190.00, "GOOGL": 175.00, @@ -14,9 +18,7 @@ "NFLX": 600.00, } -# Per-ticker GBM parameters -# sigma: annualized volatility (higher = more price movement) -# mu: annualized drift / expected return +# sigma: annualised volatility. mu: annualised drift. TICKER_PARAMS: dict[str, dict[str, float]] = { "AAPL": {"sigma": 0.22, "mu": 0.05}, "GOOGL": {"sigma": 0.25, "mu": 0.05}, @@ -30,18 +32,37 @@ "NFLX": {"sigma": 0.35, "mu": 0.05}, } -# Default parameters for tickers not in the list above (dynamically added) -DEFAULT_PARAMS: dict[str, float] = {"sigma": 0.25, "mu": 0.05} - -# Correlation groups for the simulator's Cholesky decomposition -# Tickers in the same group have higher intra-group correlation +# Correlation groups. TSLA is deliberately in neither: nominally tech, famously +# does its own thing, and its independence gives the watchlist texture. CORRELATION_GROUPS: dict[str, set[str]] = { "tech": {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"}, "finance": {"JPM", "V"}, } -# Correlation coefficients -INTRA_TECH_CORR = 0.6 # Tech stocks move together -INTRA_FINANCE_CORR = 0.5 # Finance stocks move together -CROSS_GROUP_CORR = 0.3 # Between sectors / unknown tickers -TSLA_CORR = 0.3 # TSLA does its own thing +INTRA_TECH_CORR = 0.6 # Tech names move together +INTRA_FINANCE_CORR = 0.5 # Finance names move together +CROSS_GROUP_CORR = 0.3 # Across sectors, TSLA, and synthesised tickers + + +def synthesize_params(ticker: str) -> tuple[float, dict[str, float]]: + """Derive a stable seed price and GBM parameters from the symbol itself. + + Deterministic by construction: SHA-256 of the symbol, never random. A user + holding 10 shares of PYPL bought at $73 must not restart the container and + find PYPL trading at $412 — their P&L would be nonsense. + + Ranges are chosen so every synthesised ticker looks like an ordinary + large-cap: $20-$500, sigma 0.15-0.50, mu 0.02-0.08. + """ + digest = hashlib.sha256(ticker.encode()).digest() + price = 20.0 + (int.from_bytes(digest[0:4], "big") % 48_000) / 100.0 + sigma = 0.15 + (digest[4] / 255.0) * 0.35 + mu = 0.02 + (digest[5] / 255.0) * 0.06 + return round(price, 2), {"sigma": round(sigma, 4), "mu": round(mu, 4)} + + +def params_for(ticker: str) -> tuple[float, dict[str, float]]: + """Seed price and GBM parameters for any well-formed ticker.""" + if ticker in SEED_PRICES: + return SEED_PRICES[ticker], dict(TICKER_PARAMS[ticker]) + return synthesize_params(ticker) diff --git a/backend/app/market/simulator.py b/backend/app/market/simulator.py index b6803f592..5b8a6a458 100644 --- a/backend/app/market/simulator.py +++ b/backend/app/market/simulator.py @@ -9,24 +9,27 @@ import numpy as np -from .cache import PriceCache +from .cache import HISTORY_POINTS, PriceCache from .interface import MarketDataSource from .seed_prices import ( CORRELATION_GROUPS, CROSS_GROUP_CORR, - DEFAULT_PARAMS, INTRA_FINANCE_CORR, INTRA_TECH_CORR, - SEED_PRICES, - TICKER_PARAMS, - TSLA_CORR, + params_for, ) logger = logging.getLogger(__name__) +# 500ms expressed as a fraction of a trading year +# 252 trading days * 6.5 hours/day * 3600 seconds/hour = 5,896,800 seconds +TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 +DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8 for a 500ms tick +HISTORY_STEP_TICKS = 120 # One backfill point per simulated minute + class GBMSimulator: - """Geometric Brownian Motion simulator for correlated stock prices. + """Correlated geometric Brownian motion over a set of tickers. Math: S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) @@ -35,17 +38,16 @@ class GBMSimulator: S(t) = current price mu = annualized drift (expected return) sigma = annualized volatility - dt = time step as fraction of a trading year + dt = time step as a fraction of a trading year Z = correlated standard normal random variable - The tiny dt (~8.5e-8 for 500ms ticks over 252 trading days * 6.5h/day) - produces sub-cent moves per tick that accumulate naturally over time. + Pure and synchronous: holds prices, parameters and the Cholesky factor, and + knows nothing about the cache, asyncio or FastAPI. Tests drive step() + directly with a large dt instead of sleeping. """ - # 500ms expressed as a fraction of a trading year - # 252 trading days * 6.5 hours/day * 3600 seconds/hour = 5,896,800 seconds - TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 - DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8 + TRADING_SECONDS_PER_YEAR = TRADING_SECONDS_PER_YEAR + DEFAULT_DT = DEFAULT_DT def __init__( self, @@ -64,64 +66,58 @@ def __init__( # Cholesky decomposition of the correlation matrix (for correlated moves) self._cholesky: np.ndarray | None = None - # Initialize all starting tickers for ticker in tickers: - self._add_ticker_internal(ticker) + self._add_internal(ticker) self._rebuild_cholesky() # --- Public API --- def step(self) -> dict[str, float]: - """Advance all tickers by one time step. Returns {ticker: new_price}. + """Advance every ticker one tick. Returns {ticker: rounded price}. - This is the hot path — called every 500ms. Keep it fast. + The hot path: called every 500ms forever. All n normals are drawn in one + numpy call and correlated with one matrix multiply rather than looping. """ n = len(self._tickers) if n == 0: return {} - # Generate n independent standard normal draws - z_independent = np.random.standard_normal(n) - - # Apply Cholesky to get correlated draws + z = np.random.standard_normal(n) if self._cholesky is not None: - z_correlated = self._cholesky @ z_independent - else: - z_correlated = z_independent + z = self._cholesky @ z - result: dict[str, float] = {} + prices: dict[str, float] = {} for i, ticker in enumerate(self._tickers): params = self._params[ticker] - mu = params["mu"] - sigma = params["sigma"] + mu, sigma = params["mu"], params["sigma"] - # GBM: S(t+dt) = S(t) * exp((mu - 0.5*sigma^2)*dt + sigma*sqrt(dt)*Z) drift = (mu - 0.5 * sigma**2) * self._dt - diffusion = sigma * math.sqrt(self._dt) * z_correlated[i] + diffusion = sigma * math.sqrt(self._dt) * z[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 shock: GBM alone is smooth, real markets jump. ~0.1% per + # tick per ticker is an event every ~50s across ten tickers — often + # enough to see in a minute, rare enough to stay an event. if random.random() < self._event_prob: - shock_magnitude = random.uniform(0.02, 0.05) - shock_sign = random.choice([-1, 1]) - self._prices[ticker] *= 1 + shock_magnitude * shock_sign + magnitude = random.uniform(0.02, 0.05) + sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + magnitude * sign logger.debug( - "Random event on %s: %.1f%% %s", + "Shock event on %s: %.1f%% %s", ticker, - shock_magnitude * 100, - "up" if shock_sign > 0 else "down", + magnitude * 100, + "up" if sign > 0 else "down", ) - result[ticker] = round(self._prices[ticker], 2) + prices[ticker] = round(self._prices[ticker], 2) - return result + return prices def add_ticker(self, ticker: str) -> None: """Add a ticker to the simulation. Rebuilds the correlation matrix.""" if ticker in self._prices: return - self._add_ticker_internal(ticker) + self._add_internal(ticker) self._rebuild_cholesky() def remove_ticker(self, ticker: str) -> None: @@ -135,73 +131,90 @@ def remove_ticker(self, ticker: str) -> None: def get_price(self, ticker: str) -> float | None: """Current price for a ticker, or None if not tracked.""" - return self._prices.get(ticker) + price = self._prices.get(ticker) + return round(price, 2) if price is not None else None def get_tickers(self) -> list[str]: """Return the list of currently tracked tickers.""" return list(self._tickers) + def backfill_history(self, ticker: str, points: int = HISTORY_POINTS) -> list[float]: + """Manufacture plausible prior history ending at the current price. + + Runs the GBM recurrence *backwards* — dividing rather than multiplying — + so the series ends at the live price and joins the stream continuously. + The coarser dt gives a per-minute cadence, so 60 points is an hour of + price action with visible shape rather than 30 seconds of flat line. + """ + params = self._params.get(ticker) + if params is None or points < 1: + return [] + + mu, sigma = params["mu"], params["sigma"] + dt = self._dt * HISTORY_STEP_TICKS + drift = (mu - 0.5 * sigma**2) * dt + scale = sigma * math.sqrt(dt) + + price = self._prices[ticker] + history = [round(price, 2)] + for z in np.random.standard_normal(points - 1): + price /= math.exp(drift + scale * z) + history.append(round(price, 2)) + + history.reverse() # Oldest first, ending at the current price + return history + # --- Internals --- - def _add_ticker_internal(self, ticker: str) -> None: - """Add a ticker without rebuilding Cholesky (for batch initialization).""" + def _add_internal(self, ticker: str) -> None: + """Add without rebuilding Cholesky, for batch initialization.""" if ticker in self._prices: return + price, params = params_for(ticker) self._tickers.append(ticker) - self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) - self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) + self._prices[ticker] = price + self._params[ticker] = 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. - """ + """Refactor the correlation matrix. O(n^3), called only on add/remove.""" n = len(self._tickers) if n <= 1: self._cholesky = None return - # Build the correlation matrix corr = np.eye(n) for i in range(n): for j in range(i + 1, n): rho = self._pairwise_correlation(self._tickers[i], self._tickers[j]) - corr[i, j] = rho - corr[j, i] = rho - - self._cholesky = np.linalg.cholesky(corr) + corr[i, j] = corr[j, i] = rho + + try: + self._cholesky = np.linalg.cholesky(corr) + except np.linalg.LinAlgError: + # Should be unreachable with the block structure in seed_prices.py. + # Degrade to independent draws rather than taking the price feed + # down over a correlation constant. + logger.error("Correlation matrix not positive definite; using independent draws") + self._cholesky = None @staticmethod def _pairwise_correlation(t1: str, t2: str) -> float: - """Determine correlation between two tickers based on sector grouping. - - Correlation structure: - - Same tech sector: 0.6 - - Same finance sector: 0.5 - - TSLA with anything: 0.3 (it does its own thing) - - Cross-sector: 0.3 - - Unknown tickers: 0.3 - """ + """Sector-based correlation: tech 0.6, finance 0.5, everything else 0.3.""" tech = CORRELATION_GROUPS["tech"] finance = CORRELATION_GROUPS["finance"] - # TSLA is in tech set but behaves independently - if t1 == "TSLA" or t2 == "TSLA": - return TSLA_CORR - if t1 in tech and t2 in tech: return INTRA_TECH_CORR if t1 in finance and t2 in finance: return INTRA_FINANCE_CORR - return CROSS_GROUP_CORR class SimulatorDataSource(MarketDataSource): """MarketDataSource backed by the GBM simulator. - Runs a background asyncio task that calls GBMSimulator.step() every - `update_interval` seconds and writes results to the PriceCache. + Owns one asyncio task that steps the simulation every `update_interval` + seconds and writes each price to the cache. No maths, no I/O. """ def __init__( @@ -216,16 +229,21 @@ def __init__( self._sim: GBMSimulator | None = None self._task: asyncio.Task | None = None + @property + def source_name(self) -> str: + return "simulator" + async def start(self, tickers: list[str]) -> None: self._sim = GBMSimulator( tickers=tickers, event_probability=self._event_prob, ) - # Seed the cache with initial prices so SSE has data immediately + # Populate the cache *before* the loop starts, so start() returns with + # prices and sparklines already available and the first HTTP request + # never sees an empty cache. for ticker in tickers: - price = self._sim.get_price(ticker) - if price is not None: - self._cache.update(ticker=ticker, price=price) + self._seed(ticker) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") logger.info("Simulator started with %d tickers", len(tickers)) @@ -240,13 +258,13 @@ async def stop(self) -> None: logger.info("Simulator stopped") async def add_ticker(self, ticker: str) -> None: - if self._sim: - self._sim.add_ticker(ticker) - # Seed cache immediately so the ticker has a price right away - price = self._sim.get_price(ticker) - if price is not None: - self._cache.update(ticker=ticker, price=price) - logger.info("Simulator: added ticker %s", ticker) + if self._sim is None: + return + if ticker in self._sim.get_tickers(): + return + self._sim.add_ticker(ticker) + self._seed(ticker) + logger.info("Simulator: added ticker %s", ticker) async def remove_ticker(self, ticker: str) -> None: if self._sim: @@ -257,6 +275,18 @@ async def remove_ticker(self, ticker: str) -> None: def get_tickers(self) -> list[str]: return self._sim.get_tickers() if self._sim else [] + # --- Internals --- + + def _seed(self, ticker: str) -> None: + """Backfill history then publish the first price. Order matters: the + history must be in place before the price the sparkline ends at.""" + assert self._sim is not None + price = self._sim.get_price(ticker) + if price is None: + return + self._cache.seed_history(ticker, self._sim.backfill_history(ticker)) + self._cache.update(ticker=ticker, price=price) + async def _run_loop(self) -> None: """Core loop: step the simulation, write to cache, sleep.""" while True: @@ -266,5 +296,8 @@ async def _run_loop(self) -> None: for ticker, price in prices.items(): self._cache.update(ticker=ticker, price=price) except Exception: + # A background task that raises dies silently and takes the + # whole price feed with it, leaving a UI that looks connected + # and frozen. Log and take the next tick. logger.exception("Simulator step failed") await asyncio.sleep(self._interval) diff --git a/backend/app/market/stream.py b/backend/app/market/stream.py index 7fd974b7c..16a5f088b 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 @@ -14,14 +15,17 @@ logger = logging.getLogger(__name__) -router = APIRouter(prefix="/api/stream", tags=["streaming"]) +POLL_INTERVAL = 0.5 # How often the generator looks at the cache +HEARTBEAT_INTERVAL = 15.0 # Comment frame cadence, price activity or not def create_stream_router(price_cache: PriceCache) -> APIRouter: - """Create the SSE streaming router with a reference to the price cache. + """Build the /api/stream router bound to a specific cache. - This factory pattern lets us inject the PriceCache without globals. + The router is created inside the factory, not at module level, so calling + this twice (an app plus a test app) does not register the route twice. """ + router = APIRouter(prefix="/api/stream", tags=["streaming"]) @router.get("/prices") async def stream_prices(request: Request) -> StreamingResponse: @@ -51,37 +55,44 @@ async def stream_prices(request: Request) -> StreamingResponse: async def _generate_events( price_cache: PriceCache, request: Request, - interval: float = 0.5, + interval: float = POLL_INTERVAL, + heartbeat: float = HEARTBEAT_INTERVAL, ) -> AsyncGenerator[str, None]: - """Async generator that yields SSE-formatted price events. + """Yield SSE frames until the client disconnects. - Sends all prices every `interval` seconds. Stops when the client - disconnects (detected via request.is_disconnected()). + Emits one data event carrying every tracked ticker, keyed by symbol, and + only when the cache version has moved. A heartbeat comment goes out every + `heartbeat` seconds regardless, so silence is legible to the frontend. """ + client = request.client.host if request.client else "unknown" + logger.info("SSE client connected: %s", client) + # Tell the client to retry after 1 second if the connection drops yield "retry: 1000\n\n" last_version = -1 - client_ip = request.client.host if request.client else "unknown" - logger.info("SSE client connected: %s", client_ip) + last_beat = time.monotonic() try: while True: - # Check for client disconnect if await request.is_disconnected(): - logger.info("SSE client disconnected: %s", client_ip) + logger.info("SSE client disconnected: %s", client) break - current_version = price_cache.version - if current_version != last_version: - last_version = current_version + version = price_cache.version + if version != last_version: + last_version = version prices = price_cache.get_all() - if prices: - data = {ticker: update.to_dict() for ticker, update in prices.items()} - payload = json.dumps(data) - yield f"data: {payload}\n\n" + payload = {ticker: update.to_dict() for ticker, update in prices.items()} + yield f"data: {json.dumps(payload)}\n\n" + + now = time.monotonic() + if now - last_beat >= heartbeat: + yield ": ping\n\n" + last_beat = now await asyncio.sleep(interval) except asyncio.CancelledError: - logger.info("SSE stream cancelled for: %s", client_ip) + logger.info("SSE stream cancelled: %s", client) + raise diff --git a/backend/app/market/tickers.py b/backend/app/market/tickers.py new file mode 100644 index 000000000..4e078574c --- /dev/null +++ b/backend/app/market/tickers.py @@ -0,0 +1,22 @@ +"""Ticker symbol validation — one rule, shared by every caller.""" + +from __future__ import annotations + +import re + +TICKER_PATTERN = re.compile(r"^[A-Z]{1,5}$") + + +def normalize_ticker(raw: str) -> str: + """Uppercase, strip, and validate a ticker symbol. + + Raises ValueError if the symbol is not 1-5 A-Z characters. Callers turn + that into a 400 with the message shown to the user verbatim. + + This validates shape, not existence: the simulator accepts any well-formed + symbol and synthesizes parameters for it (see seed_prices.py). + """ + ticker = raw.strip().upper() + if not TICKER_PATTERN.match(ticker): + raise ValueError(f"Invalid ticker symbol: {raw!r}") + return ticker diff --git a/backend/market_data_demo.py b/backend/market_data_demo.py index 7414416c4..cd033a313 100644 --- a/backend/market_data_demo.py +++ b/backend/market_data_demo.py @@ -53,7 +53,6 @@ def format_price(price: float) -> str: def build_table( cache: PriceCache, - history: dict[str, deque], elapsed: float, ) -> Table: """Build the price table.""" @@ -68,7 +67,7 @@ def build_table( table.add_column("Ticker", style="bold bright_white", width=8) table.add_column("Price", justify="right", width=10) table.add_column("Change", justify="right", width=9) - table.add_column("Chg %", justify="right", width=8) + table.add_column("Chg % (open)", justify="right", width=12) table.add_column("", width=3) # arrow table.add_column("Sparkline", width=42, no_wrap=True) @@ -91,10 +90,12 @@ def build_table( price_str = f"[{color}]${format_price(update.price)}[/]" change_str = f"[{color}]{update.change:+.2f}[/]" - pct_str = f"[{color}]{update.change_percent:+.2f}%[/]" + # The user-facing "change %" column compares against the session open, + # not the previous tick \u2014 see PLAN.md section 6. + pct_str = f"[{color}]{update.change_from_open_percent:+.2f}%[/]" - # Sparkline from history - vals = list(history.get(ticker, [])) + # Sparkline straight from the cache's own backfilled/live history + vals = cache.get_history(ticker) spark_str = f"[bright_cyan]{sparkline(vals)}[/]" if len(vals) > 1 else "" table.add_row(ticker, price_str, change_str, pct_str, arrow, spark_str) @@ -120,7 +121,6 @@ def build_event_log(events: deque) -> Panel: def build_dashboard( cache: PriceCache, - history: dict[str, deque], events: deque, start_time: float, ) -> Layout: @@ -153,7 +153,7 @@ def build_dashboard( # Body: price table layout["body"].update( Panel( - build_table(cache, history, elapsed), + build_table(cache, elapsed), title="[bold bright_white]Live Prices[/]", border_style="bright_black", ) @@ -209,24 +209,15 @@ async def run() -> None: cache = PriceCache() source = SimulatorDataSource(price_cache=cache, update_interval=0.5) - # Per-ticker price history for sparklines - history: dict[str, deque] = {t: deque(maxlen=40) for t in TICKERS} - # Recent event log events: deque = deque(maxlen=12) await source.start(TICKERS) start_time = time.time() - # Seed initial history points - for ticker in TICKERS: - update = cache.get(ticker) - if update: - history[ticker].append(update.price) - try: with Live( - build_dashboard(cache, history, events, start_time), + build_dashboard(cache, events, start_time), refresh_per_second=4, screen=True, ) as live: @@ -239,14 +230,12 @@ async def run() -> None: continue last_version = cache.version - # Record history & detect events + # Detect notable moves for the event log for ticker in TICKERS: update = cache.get(ticker) if update is None: continue - history[ticker].append(update.price) - # 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" @@ -258,7 +247,7 @@ async def run() -> None: f"${format_price(update.price)}" ) - live.update(build_dashboard(cache, history, events, start_time)) + live.update(build_dashboard(cache, events, start_time)) except KeyboardInterrupt: pass diff --git a/backend/pyproject.toml b/backend/pyproject.toml index e172cca22..1dda0ad7f 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -8,7 +8,7 @@ dependencies = [ "fastapi>=0.115.0", "uvicorn[standard]>=0.32.0", "numpy>=2.0.0", - "massive>=1.0.0", + "massive==2.2.0", "rich>=13.0.0", ] diff --git a/backend/tests/market/test_cache.py b/backend/tests/market/test_cache.py index b5ab3d55d..e7bc67c5c 100644 --- a/backend/tests/market/test_cache.py +++ b/backend/tests/market/test_cache.py @@ -1,6 +1,10 @@ """Tests for PriceCache.""" -from app.market.cache import PriceCache +import asyncio + +import pytest + +from app.market.cache import PriceCache, wait_for_price class TestPriceCache: @@ -58,7 +62,7 @@ def test_get_all(self): assert set(all_prices.keys()) == {"AAPL", "GOOGL"} def test_version_increments(self): - """Test that version counter increments.""" + """Test that version counter increments on a real price change.""" cache = PriceCache() v0 = cache.version cache.update("AAPL", 190.00) @@ -96,8 +100,138 @@ def test_custom_timestamp(self): update = cache.update("AAPL", 190.50, timestamp=custom_ts) assert update.timestamp == custom_ts + def test_zero_timestamp_is_not_discarded(self): + """timestamp=0.0 is a legitimate value and must not fall back to time.time().""" + cache = PriceCache() + update = cache.update("AAPL", 190.50, timestamp=0.0) + assert update.timestamp == 0.0 + def test_price_rounding(self): """Test that prices are rounded to 2 decimal places.""" cache = PriceCache() update = cache.update("AAPL", 190.12345) assert update.price == 190.12 + + # --- Session baseline --- + + def test_first_update_pins_the_session_baseline(self): + """The first update for a ticker sets its open_price.""" + cache = PriceCache() + update = cache.update("AAPL", 190.00) + assert update.open_price == 190.00 + assert update.previous_price == 190.00 + assert update.direction == "flat" + assert update.change_from_open_percent == 0.0 + + def test_open_price_survives_later_updates(self): + """open_price stays pinned to the first price seen, not the latest tick.""" + cache = PriceCache() + cache.update("AAPL", 190.00) + cache.update("AAPL", 191.00) + update = cache.update("AAPL", 192.00) + assert update.open_price == 190.00 + assert update.previous_price == 191.00 + assert update.change_from_open_percent == pytest.approx(1.0526, abs=1e-3) + + def test_open_price_resets_after_remove_and_readd(self): + """A ticker re-added after removal gets a fresh session baseline.""" + cache = PriceCache() + cache.update("AAPL", 190.00) + cache.remove("AAPL") + update = cache.update("AAPL", 250.00) + assert update.open_price == 250.00 + + def test_repeated_price_does_not_bump_version(self): + """Emitting the same price twice must not look like a change to SSE.""" + cache = PriceCache() + cache.update("AAPL", 190.00) + version = cache.version + cache.update("AAPL", 190.00) + assert cache.version == version + assert cache.get("AAPL").timestamp > 0 + + def test_repeated_price_still_refreshes_timestamp(self): + """Even with no visible change, the feed should look alive to /api/health.""" + cache = PriceCache() + cache.update("AAPL", 190.00, timestamp=100.0) + cache.update("AAPL", 190.00, timestamp=200.0) + assert cache.get("AAPL").timestamp == 200.0 + + # --- History --- + + def test_history_is_bounded_and_ordered(self): + cache = PriceCache(history_points=5, history_interval=0.0) + for price in range(100, 110): + cache.update("AAPL", float(price)) + assert cache.get_history("AAPL") == [105.0, 106.0, 107.0, 108.0, 109.0] + + def test_history_respects_the_minimum_interval(self): + cache = PriceCache(history_points=10, history_interval=60.0) + cache.update("AAPL", 100.0, timestamp=0.0) + cache.update("AAPL", 101.0, timestamp=10.0) # too soon, not recorded + cache.update("AAPL", 102.0, timestamp=70.0) # 70s later, recorded + assert cache.get_history("AAPL") == [100.0, 102.0] + + def test_get_history_unknown_ticker_is_empty(self): + cache = PriceCache() + assert cache.get_history("NOPE") == [] + + def test_seed_history_then_remove_clears_everything(self): + cache = PriceCache() + cache.seed_history("AAPL", [1.0, 2.0, 3.0]) + cache.update("AAPL", 4.0) + cache.remove("AAPL") + assert cache.get("AAPL") is None + assert cache.get_history("AAPL") == [] + + def test_seed_history_replaces_existing_history(self): + cache = PriceCache(history_interval=0.0) + cache.update("AAPL", 1.0) + cache.update("AAPL", 2.0) + cache.seed_history("AAPL", [10.0, 20.0, 30.0]) + assert cache.get_history("AAPL") == [10.0, 20.0, 30.0] + + def test_seed_history_truncates_to_history_points(self): + cache = PriceCache(history_points=3) + cache.seed_history("AAPL", [1.0, 2.0, 3.0, 4.0, 5.0]) + assert cache.get_history("AAPL") == [3.0, 4.0, 5.0] + + # --- Health --- + + def test_newest_timestamp_empty_cache(self): + cache = PriceCache() + assert cache.newest_timestamp() is None + + def test_newest_timestamp_reflects_latest_write(self): + cache = PriceCache() + cache.update("AAPL", 190.00, timestamp=100.0) + cache.update("GOOGL", 175.00, timestamp=200.0) + assert cache.newest_timestamp() == 200.0 + + +@pytest.mark.asyncio +class TestWaitForPrice: + """Unit tests for the wait_for_price helper.""" + + async def test_returns_immediately_when_price_present(self): + cache = PriceCache() + cache.update("AAPL", 190.00) + price = await wait_for_price(cache, "AAPL", timeout=1.0) + assert price == 190.00 + + async def test_waits_for_a_price_that_arrives_late(self): + cache = PriceCache() + + async def seed_later(): + await asyncio.sleep(0.05) + cache.update("AAPL", 190.00) + + task = asyncio.create_task(seed_later()) + price = await wait_for_price(cache, "AAPL", timeout=1.0) + assert price == 190.00 + await task + + async def test_raises_on_timeout(self): + cache = PriceCache() + with pytest.raises(ValueError, match="AAPL"): + await wait_for_price(cache, "AAPL", timeout=0.1) diff --git a/backend/tests/market/test_conformance.py b/backend/tests/market/test_conformance.py new file mode 100644 index 000000000..0535f8fce --- /dev/null +++ b/backend/tests/market/test_conformance.py @@ -0,0 +1,87 @@ +"""Conformance suite: the lifecycle contract must hold for every MarketDataSource. + +Anything only one implementation passes is a leak in the abstraction that +downstream code (SSE, portfolio valuation, trade execution) would otherwise +have to special-case. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from app.market.cache import PriceCache +from app.market.massive_client import MassiveDataSource +from app.market.simulator import SimulatorDataSource + + +def _snapshot(ticker: str, price: float) -> MagicMock: + snap = MagicMock() + snap.ticker = ticker + snap.last_trade = MagicMock(price=price, timestamp=1605195918306274000) + snap.min = None + snap.day = None + return snap + + +@pytest.fixture(params=["simulator", "massive"]) +def source_and_cache(request): + cache = PriceCache() + if request.param == "simulator": + yield SimulatorDataSource(cache, update_interval=0.05), cache + else: + source = MassiveDataSource( + "test-key", cache, poll_interval=60.0, backfill_history=False + ) + with ( + patch.object( + source, + "_fetch_snapshots", + return_value=[_snapshot("AAPL", 190.50), _snapshot("GOOGL", 175.25)], + ), + patch.object(source, "_log_market_status", new=AsyncMock()), + patch("app.market.massive_client.RESTClient"), + ): + yield source, cache + + +@pytest.mark.asyncio +class TestSourceConformance: + async def test_source_name_is_a_short_identifier(self, source_and_cache): + source, _ = source_and_cache + assert source.source_name in {"simulator", "massive"} + + async def test_start_populates_cache_before_returning(self, source_and_cache): + source, cache = source_and_cache + await source.start(["AAPL", "GOOGL"]) + assert cache.get_price("AAPL") is not None + assert cache.get_price("GOOGL") is not None + await source.stop() + + async def test_stop_is_idempotent(self, source_and_cache): + source, _cache = source_and_cache + await source.start(["AAPL"]) + await source.stop() + await source.stop() # must not raise + + async def test_remove_ticker_clears_the_cache(self, source_and_cache): + source, cache = source_and_cache + await source.start(["AAPL", "GOOGL"]) + await source.remove_ticker("AAPL") + assert "AAPL" not in cache + assert "AAPL" not in source.get_tickers() + await source.stop() + + async def test_get_tickers_reflects_start(self, source_and_cache): + source, _cache = source_and_cache + await source.start(["AAPL", "GOOGL"]) + assert set(source.get_tickers()) == {"AAPL", "GOOGL"} + await source.stop() + + async def test_empty_start_is_not_an_error(self, source_and_cache): + source, cache = source_and_cache + await source.start([]) + assert len(cache) == 0 + assert source.get_tickers() == [] + await source.stop() diff --git a/backend/tests/market/test_massive.py b/backend/tests/market/test_massive.py index cdd7dbd24..f5b0717c6 100644 --- a/backend/tests/market/test_massive.py +++ b/backend/tests/market/test_massive.py @@ -1,20 +1,27 @@ """Tests for MassiveDataSource (mocked).""" -from unittest.mock import MagicMock, patch +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch import pytest from app.market.cache import PriceCache from app.market.massive_client import MassiveDataSource +# Massive's own sample lastTrade.t value: 2020-11-12T15:45:18.306274 UTC in nanoseconds. +SAMPLE_TIMESTAMP_NS = 1605195918306274000 +SAMPLE_TIMESTAMP_S = 1605195918.306274 -def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock: - """Create a mock Massive snapshot object.""" + +def _make_snapshot(ticker: str, price: float, timestamp_ns: int = SAMPLE_TIMESTAMP_NS) -> MagicMock: + """Create a mock Massive snapshot object with a last_trade quote.""" snap = MagicMock() snap.ticker = ticker snap.last_trade = MagicMock() snap.last_trade.price = price - snap.last_trade.timestamp = timestamp_ms + snap.last_trade.timestamp = timestamp_ns + snap.min = None + snap.day = None return snap @@ -22,6 +29,11 @@ def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock: class TestMassiveDataSource: """Unit tests for MassiveDataSource with mocked API.""" + async def test_source_name(self): + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache) + assert source.source_name == "massive" + async def test_poll_updates_cache(self): """Test that polling updates the cache.""" cache = PriceCache() @@ -34,8 +46,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), + _make_snapshot("GOOGL", 175.25), ] with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): @@ -44,6 +56,18 @@ 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_timestamps_convert_to_seconds(self): + """last_trade.timestamp is nanoseconds, not milliseconds.""" + 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", return_value=[_make_snapshot("AAPL", 190.5)]): + await source._poll_once() + + assert cache.get("AAPL").timestamp == pytest.approx(SAMPLE_TIMESTAMP_S) + async def test_malformed_snapshot_skipped(self): """Test that malformed snapshots are skipped gracefully.""" cache = PriceCache() @@ -55,10 +79,12 @@ 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) bad_snap = MagicMock() bad_snap.ticker = "BAD" - bad_snap.last_trade = None # Will cause AttributeError + bad_snap.last_trade = None + bad_snap.min = None + bad_snap.day = None with patch.object(source, "_fetch_snapshots", return_value=[good_snap, bad_snap]): await source._poll_once() @@ -67,6 +93,61 @@ async def test_malformed_snapshot_skipped(self): assert cache.get_price("AAPL") == 190.50 assert cache.get_price("BAD") is None + async def test_falls_back_to_minute_bar_when_no_last_trade(self): + """Off-hours snapshots may have no last_trade; fall back to the minute bar.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + snap = MagicMock() + snap.ticker = "AAPL" + snap.last_trade = None + snap.min = MagicMock(close=188.25, timestamp=1707580800000) # milliseconds + snap.day = None + + with patch.object(source, "_fetch_snapshots", return_value=[snap]): + await source._poll_once() + + assert cache.get_price("AAPL") == 188.25 + assert cache.get("AAPL").timestamp == pytest.approx(1707580800.0) + + async def test_falls_back_to_day_close_when_only_day_available(self): + """The last-resort fallback uses the daily close and the wall clock.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + snap = MagicMock() + snap.ticker = "AAPL" + snap.last_trade = None + snap.min = None + snap.day = MagicMock(close=187.10) + + with patch.object(source, "_fetch_snapshots", return_value=[snap]): + await source._poll_once() + + assert cache.get_price("AAPL") == 187.10 + + async def test_no_usable_quote_is_skipped(self): + """A snapshot with no last_trade, min, or day is skipped, not crashed on.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + + snap = MagicMock() + snap.ticker = "AAPL" + snap.last_trade = None + snap.min = None + snap.day = None + + with patch.object(source, "_fetch_snapshots", return_value=[snap]): + await source._poll_once() + + assert cache.get_price("AAPL") is None + async def test_api_error_does_not_crash(self): """Test that API errors don't crash the poller.""" cache = PriceCache() @@ -83,30 +164,51 @@ 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_poll_failure_backs_off(self): + """Repeated failures should widen the interval rather than hammer the API.""" cache = PriceCache() - source = MassiveDataSource( - api_key="test-key", - price_cache=cache, - poll_interval=60.0, - ) + source = MassiveDataSource(api_key="k", price_cache=cache, poll_interval=15.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", side_effect=RuntimeError("429")): + await source._poll_once() - with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): + assert source._backoff == 2.0 + assert len(cache) == 0 + + async def test_backoff_resets_after_a_success(self): + """A successful poll should undo any prior backoff.""" + cache = PriceCache() + source = MassiveDataSource(api_key="k", price_cache=cache, poll_interval=15.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + source._backoff = 4.0 + + with patch.object(source, "_fetch_snapshots", return_value=[_make_snapshot("AAPL", 190.0)]): await source._poll_once() - update = cache.get("AAPL") - assert update is not None - assert update.timestamp == 1707580800.0 # Converted to seconds + assert source._backoff == 1.0 + + async def test_backoff_is_capped(self): + """Backoff must not grow without bound.""" + cache = PriceCache() + source = MassiveDataSource(api_key="k", price_cache=cache, poll_interval=15.0) + source._tickers = ["AAPL"] + source._client = MagicMock() + source._backoff = 8.0 + + with patch.object(source, "_fetch_snapshots", side_effect=RuntimeError("429")): + await source._poll_once() + + assert source._backoff == 8.0 async def test_add_ticker(self): """Test adding a ticker.""" cache = PriceCache() - source = MassiveDataSource(api_key="test-key", price_cache=cache) + source = MassiveDataSource( + api_key="test-key", price_cache=cache, backfill_history=False + ) await source.add_ticker("AAPL") assert "AAPL" in source.get_tickers() @@ -114,7 +216,9 @@ async def test_add_ticker(self): async def test_add_ticker_uppercase_normalization(self): """Test that tickers are normalized to uppercase.""" cache = PriceCache() - source = MassiveDataSource(api_key="test-key", price_cache=cache) + source = MassiveDataSource( + api_key="test-key", price_cache=cache, backfill_history=False + ) await source.add_ticker("aapl") assert "AAPL" in source.get_tickers() @@ -122,11 +226,22 @@ async def test_add_ticker_uppercase_normalization(self): async def test_add_ticker_strips_whitespace(self): """Test that ticker whitespace is stripped.""" cache = PriceCache() - source = MassiveDataSource(api_key="test-key", price_cache=cache) + source = MassiveDataSource( + api_key="test-key", price_cache=cache, backfill_history=False + ) await source.add_ticker(" AAPL ") assert "AAPL" in source.get_tickers() + async def test_add_duplicate_ticker_is_noop(self): + cache = PriceCache() + source = MassiveDataSource( + api_key="test-key", price_cache=cache, backfill_history=False + ) + await source.add_ticker("AAPL") + await source.add_ticker("AAPL") + assert source.get_tickers() == ["AAPL"] + async def test_remove_ticker(self): """Test removing a ticker.""" cache = PriceCache() @@ -169,12 +284,14 @@ async def test_stop_is_idempotent(self): async def test_stop_cancels_task(self): """Test that stop() cancels the polling task.""" cache = PriceCache() - source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=10.0) + source = MassiveDataSource( + api_key="test-key", price_cache=cache, poll_interval=10.0, backfill_history=False + ) - # Mock the client and start with patch("app.market.massive_client.RESTClient"): with patch.object(source, "_fetch_snapshots", return_value=[]): - await source.start(["AAPL"]) + with patch.object(source, "_log_market_status", new=AsyncMock()): + await source.start(["AAPL"]) # Verify task is running assert source._task is not None @@ -187,15 +304,46 @@ async def test_stop_cancels_task(self): async def test_start_immediate_poll(self): """Test that start() does an immediate poll before starting the loop.""" 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, backfill_history=False + ) - mock_snapshots = [_make_snapshot("AAPL", 190.50, 1707580800000)] + mock_snapshots = [_make_snapshot("AAPL", 190.50)] with patch("app.market.massive_client.RESTClient"): with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): - await source.start(["AAPL"]) + with patch.object(source, "_log_market_status", new=AsyncMock()): + await source.start(["AAPL"]) # Cache should have data immediately from the first poll assert cache.get_price("AAPL") == 190.50 await source.stop() + + async def test_start_backfills_history(self): + """start() kicks off a history backfill task per ticker.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0) + + with patch("app.market.massive_client.RESTClient"): + with patch.object(source, "_fetch_snapshots", return_value=[]): + with patch.object(source, "_log_market_status", new=AsyncMock()): + with patch.object( + source, "_fetch_history", return_value=[100.0, 101.0, 102.0] + ): + await source.start(["AAPL"]) + await asyncio.sleep(0.05) + + assert cache.get_history("AAPL") == [100.0, 101.0, 102.0] + await source.stop() + + async def test_backfill_failure_is_swallowed(self): + """A failed history fetch must not crash the backfill task.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache) + + with patch.object(source, "_fetch_history", side_effect=RuntimeError("boom")): + source._client = MagicMock() + await source._backfill_one("AAPL") # Should not raise + + assert cache.get_history("AAPL") == [] diff --git a/backend/tests/market/test_models.py b/backend/tests/market/test_models.py index 21600dfd6..4c819582b 100644 --- a/backend/tests/market/test_models.py +++ b/backend/tests/market/test_models.py @@ -10,68 +10,131 @@ 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) + 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) + 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) + 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) + 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=0.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) + update = PriceUpdate( + ticker="AAPL", price=191.00, previous_price=190.00, open_price=190.00 + ) assert update.direction == "up" def test_direction_down(self): """Test direction calculation (down).""" - update = PriceUpdate(ticker="AAPL", price=189.00, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=189.00, previous_price=190.00, open_price=190.00 + ) assert update.direction == "down" def test_direction_flat(self): """Test direction calculation (flat).""" - update = PriceUpdate(ticker="AAPL", price=190.00, previous_price=190.00, timestamp=1234567890.0) + update = PriceUpdate( + ticker="AAPL", price=190.00, previous_price=190.00, open_price=190.00 + ) assert update.direction == "flat" + def test_change_from_open_up(self): + """Test change_from_open reflects the session baseline, not the previous tick.""" + update = PriceUpdate( + ticker="AAPL", price=192.00, previous_price=191.50, open_price=190.00 + ) + assert update.change_from_open == 2.00 + + def test_change_from_open_percent(self): + """Test change_from_open_percent uses the open price as the denominator.""" + update = PriceUpdate( + ticker="AAPL", price=190.687, previous_price=190.60, open_price=189.20 + ) + assert update.change_from_open_percent == pytest.approx(0.7855, abs=1e-3) + + def test_change_from_open_percent_zero_open(self): + """Test change_from_open_percent guards against division by zero.""" + update = PriceUpdate(ticker="AAPL", price=100.00, previous_price=100.00, open_price=0.00) + assert update.change_from_open_percent == 0.0 + + def test_change_from_open_percent_differs_from_change_percent(self): + """The two 'change' numbers measure different things and can disagree.""" + # Ticked down from the previous price, but still up on the session. + update = PriceUpdate( + ticker="AAPL", price=190.00, previous_price=191.00, open_price=185.00 + ) + assert update.change_percent < 0 + assert update.change_from_open_percent > 0 + 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=189.20, + timestamp=1234567890.0, + ) result = update.to_dict() assert result["ticker"] == "AAPL" assert result["price"] == 190.50 assert result["previous_price"] == 190.00 + assert result["open_price"] == 189.20 assert result["timestamp"] == 1234567890.0 assert result["change"] == 0.50 assert result["change_percent"] == 0.2632 # (0.50 / 190.00) * 100 + assert result["change_from_open_percent"] == pytest.approx(0.6871, abs=1e-3) assert result["direction"] == "up" 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, + timestamp=1234567890.0, + ) with pytest.raises(AttributeError): update.price = 200.00 # Should raise error diff --git a/backend/tests/market/test_simulator.py b/backend/tests/market/test_simulator.py index 1845ec16b..530678a98 100644 --- a/backend/tests/market/test_simulator.py +++ b/backend/tests/market/test_simulator.py @@ -1,6 +1,10 @@ """Tests for GBMSimulator.""" -from app.market.seed_prices import SEED_PRICES +from unittest.mock import patch + +import numpy as np + +from app.market.seed_prices import SEED_PRICES, synthesize_params from app.market.simulator import GBMSimulator @@ -52,12 +56,18 @@ 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") + def test_unknown_ticker_gets_synthesized_price(self): + """Unknown tickers get a deterministic price in the plausible large-cap range.""" + sim = GBMSimulator(tickers=["ZZZZZ"]) + price = sim.get_price("ZZZZZ") assert price is not None - assert 50.0 <= price <= 300.0 + assert 20.0 <= price <= 500.0 + + def test_unknown_ticker_price_is_deterministic(self): + """Restarting the process must not reprice a held position.""" + sim_a = GBMSimulator(tickers=["PYPL"]) + sim_b = GBMSimulator(tickers=["PYPL"]) + assert sim_a.get_price("PYPL") == sim_b.get_price("PYPL") def test_empty_step(self): """Test stepping with no tickers.""" @@ -89,6 +99,16 @@ def test_cholesky_none_with_one_ticker(self): sim = GBMSimulator(tickers=["AAPL"]) assert sim._cholesky is None + def test_cholesky_failure_degrades_to_independent_draws(self): + """A non positive-definite matrix must not take the whole feed down.""" + sim = GBMSimulator(tickers=["AAPL", "GOOGL"]) + with patch("numpy.linalg.cholesky", side_effect=np.linalg.LinAlgError("singular")): + sim._rebuild_cholesky() + assert sim._cholesky is None + # The simulator keeps working with independent draws. + 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"]) @@ -105,7 +125,7 @@ def test_pairwise_correlation_finance_stocks(self): assert corr == 0.5 def test_pairwise_correlation_tsla(self): - """Test that TSLA has lower correlation with everything.""" + """TSLA is deliberately outside both sector groups.""" corr = GBMSimulator._pairwise_correlation("TSLA", "AAPL") assert corr == 0.3 corr = GBMSimulator._pairwise_correlation("TSLA", "JPM") @@ -126,6 +146,48 @@ def test_prices_rounded_to_two_decimals(self): result = sim.step() price_str = str(result["AAPL"]) # Check that we have at most 2 decimal places - if '.' in price_str: - decimal_part = price_str.split('.')[1] + if "." in price_str: + decimal_part = price_str.split(".")[1] assert len(decimal_part) <= 2 + + # --- History backfill --- + + def test_backfill_returns_requested_points(self): + sim = GBMSimulator(tickers=["AAPL"]) + history = sim.backfill_history("AAPL", points=60) + assert len(history) == 60 + + def test_backfill_ends_at_the_current_price(self): + sim = GBMSimulator(tickers=["AAPL"]) + history = sim.backfill_history("AAPL") + assert history[-1] == sim.get_price("AAPL") + + def test_backfill_is_oldest_first_and_all_positive(self): + sim = GBMSimulator(tickers=["AAPL"]) + history = sim.backfill_history("AAPL", points=10) + assert len(history) == 10 + assert all(p > 0 for p in history) + + def test_backfill_unknown_ticker_returns_empty(self): + sim = GBMSimulator(tickers=["AAPL"]) + assert sim.backfill_history("UNKNOWN") == [] + + def test_backfill_zero_points_returns_empty(self): + sim = GBMSimulator(tickers=["AAPL"]) + assert sim.backfill_history("AAPL", points=0) == [] + + +class TestSynthesizeParams: + """Unit tests for the deterministic unknown-ticker parameter synthesis.""" + + def test_deterministic_across_calls(self): + assert synthesize_params("PYPL") == synthesize_params("PYPL") + + def test_price_and_sigma_within_plausible_ranges(self): + price, params = synthesize_params("PYPL") + assert 20.0 <= price <= 500.0 + assert 0.15 <= params["sigma"] <= 0.50 + assert 0.02 <= params["mu"] <= 0.08 + + def test_different_tickers_get_different_params(self): + assert synthesize_params("AMD") != synthesize_params("DIS") diff --git a/backend/tests/market/test_simulator_source.py b/backend/tests/market/test_simulator_source.py index 515ce7290..6ba7e5238 100644 --- a/backend/tests/market/test_simulator_source.py +++ b/backend/tests/market/test_simulator_source.py @@ -12,6 +12,11 @@ class TestSimulatorDataSource: """Integration tests for the SimulatorDataSource.""" + async def test_source_name(self): + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache) + assert source.source_name == "simulator" + async def test_start_populates_cache(self): """Test that start() immediately populates the cache.""" cache = PriceCache() @@ -24,6 +29,31 @@ async def test_start_populates_cache(self): await source.stop() + async def test_start_backfills_history(self): + """start() must seed sparkline history before returning.""" + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + + history = cache.get_history("AAPL") + assert len(history) == 60 + assert history[-1] == cache.get_price("AAPL") + + await source.stop() + + async def test_add_ticker_backfills_history(self): + """A ticker added after startup also gets seeded history, not silence.""" + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + + await source.add_ticker("TSLA") + history = cache.get_history("TSLA") + assert len(history) == 60 + assert history[-1] == cache.get_price("TSLA") + + await source.stop() + async def test_prices_update_over_time(self): """Test that prices are updated periodically.""" cache = PriceCache() @@ -59,6 +89,25 @@ async def test_add_ticker(self): await source.stop() + async def test_add_ticker_before_start_is_noop(self): + """Adding before start() has no simulator to add to yet.""" + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.add_ticker("AAPL") # Should not raise + assert source.get_tickers() == [] + + async def test_add_duplicate_ticker_is_noop(self): + """Adding a ticker already tracked must not reset its history/price.""" + cache = PriceCache() + source = SimulatorDataSource(price_cache=cache, update_interval=0.1) + await source.start(["AAPL"]) + first_update = cache.get("AAPL") + + await source.add_ticker("AAPL") + + assert cache.get("AAPL") == first_update + await source.stop() + async def test_remove_ticker(self): """Test removing a ticker.""" cache = PriceCache() diff --git a/backend/tests/market/test_stream.py b/backend/tests/market/test_stream.py new file mode 100644 index 000000000..688a8277d --- /dev/null +++ b/backend/tests/market/test_stream.py @@ -0,0 +1,102 @@ +"""Tests for the SSE streaming generator.""" + +import json + +import pytest + +from app.market.cache import PriceCache +from app.market.stream import _generate_events, create_stream_router + + +class _StubRequest: + """Minimal stand-in for a FastAPI Request, driving disconnect after N checks.""" + + client = None + + def __init__(self, disconnect_after: int): + self._calls = 0 + self._limit = disconnect_after + + async def is_disconnected(self) -> bool: + self._calls += 1 + return self._calls > self._limit + + +@pytest.mark.asyncio +class TestGenerateEvents: + """Unit tests for _generate_events, driven directly without an ASGI server.""" + + async def test_opens_with_retry_directive(self): + cache = PriceCache() + frames = [f async for f in _generate_events(cache, _StubRequest(0), interval=0.0)] + assert frames[0] == "retry: 1000\n\n" + + async def test_no_event_when_the_version_is_unchanged(self): + cache = PriceCache() + cache.update("AAPL", 190.00) + frames = [f async for f in _generate_events(cache, _StubRequest(3), interval=0.0)] + data_frames = [f for f in frames if f.startswith("data:")] + assert len(data_frames) == 1 # one payload, then silence + + async def test_no_event_emitted_when_cache_is_empty(self): + cache = PriceCache() + frames = [f async for f in _generate_events(cache, _StubRequest(3), interval=0.0)] + data_frames = [f for f in frames if f.startswith("data:")] + assert data_frames == [] + + async def test_heartbeat_arrives_in_a_quiet_market(self): + cache = PriceCache() + frames = [ + f + async for f in _generate_events(cache, _StubRequest(3), interval=0.0, heartbeat=0.0) + ] + assert ": ping\n\n" in frames + + async def test_no_heartbeat_before_its_interval(self): + cache = PriceCache() + frames = [ + f + async for f in _generate_events( + cache, _StubRequest(3), interval=0.0, heartbeat=3600.0 + ) + ] + assert ": ping\n\n" not in frames + + async def test_payload_is_keyed_by_ticker_and_carries_the_baseline(self): + cache = PriceCache() + cache.update("AAPL", 189.20) + cache.update("AAPL", 190.50) + frames = [f async for f in _generate_events(cache, _StubRequest(1), interval=0.0)] + data_frames = [f for f in frames if f.startswith("data:")] + payload = json.loads(data_frames[0].removeprefix("data: ")) + assert payload["AAPL"]["open_price"] == 189.20 + assert payload["AAPL"]["change_from_open_percent"] == pytest.approx(0.687, abs=1e-3) + + async def test_multiple_tickers_share_one_event(self): + cache = PriceCache() + cache.update("AAPL", 190.00) + cache.update("GOOGL", 175.00) + frames = [f async for f in _generate_events(cache, _StubRequest(1), interval=0.0)] + data_frames = [f for f in frames if f.startswith("data:")] + assert len(data_frames) == 1 + payload = json.loads(data_frames[0].removeprefix("data: ")) + assert set(payload.keys()) == {"AAPL", "GOOGL"} + + async def test_stops_when_client_disconnects_immediately(self): + cache = PriceCache() + frames = [f async for f in _generate_events(cache, _StubRequest(0), interval=0.0)] + # Only the retry directive; the loop exits before ever reading the cache. + assert frames == ["retry: 1000\n\n"] + + +class TestCreateStreamRouter: + """The router factory must not double-register routes when called twice.""" + + def test_creates_independent_routers(self): + cache = PriceCache() + router_a = create_stream_router(cache) + router_b = create_stream_router(cache) + assert router_a is not router_b + paths_a = {route.path for route in router_a.routes} + paths_b = {route.path for route in router_b.routes} + assert paths_a == paths_b == {"/api/stream/prices"} diff --git a/backend/tests/market/test_tickers.py b/backend/tests/market/test_tickers.py new file mode 100644 index 000000000..1543f3310 --- /dev/null +++ b/backend/tests/market/test_tickers.py @@ -0,0 +1,62 @@ +"""Tests for ticker symbol validation.""" + +import pytest + +from app.market.tickers import TICKER_PATTERN, normalize_ticker + + +class TestNormalizeTicker: + """Unit tests for normalize_ticker.""" + + def test_uppercases(self): + assert normalize_ticker("aapl") == "AAPL" + + def test_strips_whitespace(self): + assert normalize_ticker(" AAPL ") == "AAPL" + + def test_accepts_single_letter(self): + assert normalize_ticker("v") == "V" + + def test_accepts_five_letters(self): + assert normalize_ticker("zzzzz") == "ZZZZZ" + + def test_rejects_six_letters(self): + with pytest.raises(ValueError): + normalize_ticker("ZZZZZZ") + + def test_rejects_empty_string(self): + with pytest.raises(ValueError): + normalize_ticker("") + + def test_rejects_whitespace_only(self): + with pytest.raises(ValueError): + normalize_ticker(" ") + + def test_rejects_digits(self): + with pytest.raises(ValueError): + normalize_ticker("12345") + + def test_rejects_words_with_spaces(self): + with pytest.raises(ValueError): + normalize_ticker("hello world") + + def test_rejects_symbols(self): + with pytest.raises(ValueError): + normalize_ticker("AA-PL") + + def test_error_message_includes_original_input(self): + with pytest.raises(ValueError, match="hello world"): + normalize_ticker("hello world") + + +class TestTickerPattern: + """Sanity checks on the shared regex, independent of normalize_ticker.""" + + def test_matches_well_formed_ticker(self): + assert TICKER_PATTERN.match("AAPL") + + def test_does_not_match_lowercase(self): + assert TICKER_PATTERN.match("aapl") is None + + def test_does_not_match_empty(self): + assert TICKER_PATTERN.match("") is None