Complete market data backend: anchor, validation, keepalive, reconciliation

backend/CLAUDE.md

@@ -17,27 +17,33 @@ from app.market import PriceCache, PriceUpdate, MarketDataSource, create_market_
 
 ### 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`, `anchor`, `timestamp`, plus properties `change`/`change_percent`/`direction` ("up"/"down"/"flat", tick-to-tick) and `day_change`/`day_change_percent` (vs. `anchor` — this is the "% change" shown in the watchlist), and `to_dict()` for JSON serialization.
 
 - **`PriceCache`** — Thread-safe in-memory store. Key methods:
-  - `update(ticker, price, timestamp=None) -> PriceUpdate`
+  - `update(ticker, price, timestamp=None, anchor=None) -> PriceUpdate` — `anchor` is captured only on a ticker's first write and stays sticky after that
   - `get(ticker) -> PriceUpdate | None`
   - `get_price(ticker) -> float | None`
+  - `get_anchor(ticker) -> float | None`
   - `get_all() -> dict[str, PriceUpdate]`
-  - `remove(ticker)`
+  - `remove(ticker)` — also drops the ticker's anchor
   - `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()`.
 
 - **`create_market_data_source(cache)`** — Factory. Returns `MassiveDataSource` if `MASSIVE_API_KEY` is set, otherwise `SimulatorDataSource`.
 
+- **`validate_ticker(raw) -> str`** / **`InvalidTickerError`** — normalizes and enforces the `^[A-Z]{1,5}$` ticker format. Call this at every write path that accepts a ticker (watchlist add, trade, LLM-issued actions) before touching the DB or the market source.
+
+- **`get_tracked_tickers(db)`**, **`on_watchlist_add(source, ticker)`**, **`on_watchlist_remove(source, db, ticker)`**, **`on_trade_executed(source, db, ticker)`** — keep the market source's tracked ticker set equal to `watchlist ∪ open positions`. `db` is duck-typed (see `reconcile.py`'s `TrackedTickerStore` protocol) since this module has no DB dependency of its own; call these after each DB write commits.
+
 ### SSE Streaming
 
 ```python
 from app.market import create_stream_router
 
 router = create_stream_router(price_cache)  # Returns FastAPI APIRouter
 # Endpoint: GET /api/stream/prices (text/event-stream)
+# Emits a ": keepalive" comment every ~15s when no price has changed.
 ```
 
 ### Seed Data

backend/app/market/__init__.py

@@ -1,23 +1,39 @@
 """Market data subsystem for FinAlly.
 
 Public API:
-    PriceUpdate         - Immutable price snapshot dataclass
+    PriceUpdate         - Immutable price snapshot dataclass (includes day-change anchor)
     PriceCache          - Thread-safe in-memory price store
     MarketDataSource    - Abstract interface for data providers
     create_market_data_source - Factory that selects simulator or Massive
     create_stream_router - FastAPI router factory for SSE endpoint
+    validate_ticker / InvalidTickerError - Ticker symbol format validation
+    get_tracked_tickers / on_watchlist_add / on_watchlist_remove / on_trade_executed
+        - Keep the market source's tracked set in sync with watchlist ∪ open positions
 """
 
 from .cache import PriceCache
 from .factory import create_market_data_source
 from .interface import MarketDataSource
 from .models import PriceUpdate
+from .reconcile import (
+    get_tracked_tickers,
+    on_trade_executed,
+    on_watchlist_add,
+    on_watchlist_remove,
+)
 from .stream import create_stream_router
+from .validation import InvalidTickerError, validate_ticker
 
 __all__ = [
     "PriceUpdate",
     "PriceCache",
     "MarketDataSource",
     "create_market_data_source",
     "create_stream_router",
+    "get_tracked_tickers",
+    "on_watchlist_add",
+    "on_watchlist_remove",
+    "on_trade_executed",
+    "validate_ticker",
+    "InvalidTickerError",
 ]

backend/app/market/cache.py

@@ -17,24 +17,42 @@ class PriceCache:
 
     def __init__(self) -> None:
         self._prices: dict[str, PriceUpdate] = {}
+        self._anchors: dict[str, float] = {}  # ticker -> day-change baseline
         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,
+        anchor: 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').
+
+        `anchor`, when given, is the day-change baseline for this ticker (e.g. Massive's
+        previous close). It is captured only on the *first* update seen for a ticker — later
+        calls ignore the argument and keep whatever anchor was captured first, so day-change
+        stays stable across a session (it only re-anchors if the ticker is removed and later
+        re-tracked). If no anchor is given (simulator mode, or a briefly missing previous
+        close), the anchor defaults to this call's `price` — "first observed price."
         """
         with self._lock:
             ts = timestamp or time.time()
             prev = self._prices.get(ticker)
             previous_price = prev.price if prev else price
 
+            if ticker not in self._anchors:
+                self._anchors[ticker] = round(anchor if anchor is not None else price, 2)
+
             update = PriceUpdate(
                 ticker=ticker,
                 price=round(price, 2),
                 previous_price=round(previous_price, 2),
+                anchor=self._anchors[ticker],
                 timestamp=ts,
             )
             self._prices[ticker] = update
@@ -46,6 +64,11 @@ def get(self, ticker: str) -> PriceUpdate | None:
         with self._lock:
             return self._prices.get(ticker)
 
+    def get_anchor(self, ticker: str) -> float | None:
+        """The captured day-change baseline for a ticker, or None if untracked."""
+        with self._lock:
+            return self._anchors.get(ticker)
+
     def get_all(self) -> dict[str, PriceUpdate]:
         """Snapshot of all current prices. Returns a shallow copy."""
         with self._lock:
@@ -57,9 +80,14 @@ def get_price(self, ticker: str) -> float | None:
         return update.price if update else None
 
     def remove(self, ticker: str) -> None:
-        """Remove a ticker from the cache (e.g., when removed from watchlist)."""
+        """Remove a ticker from the cache (e.g., when removed from watchlist).
+
+        Also drops its anchor: if the ticker is re-tracked later, it re-anchors fresh
+        from that moment — consistent with "first observed price after tracking started."
+        """
         with self._lock:
             self._prices.pop(ticker, None)
+            self._anchors.pop(ticker, None)
 
     @property
     def version(self) -> int:

backend/app/market/massive_client.py

@@ -101,10 +101,19 @@ async def _poll_once(self) -> None:
                     price = snap.last_trade.price
                     # Massive timestamps are Unix milliseconds → convert to seconds
                     timestamp = snap.last_trade.timestamp / 1000.0
+
+                    # Best-effort previous-close anchor. If the field is missing/None on
+                    # this snapshot (pre-market, a thin plan tier, a transient partial
+                    # response), fall back silently to PriceCache's own "first observed"
+                    # default by passing anchor=None — never let a missing anchor drop
+                    # the price update.
+                    anchor = getattr(getattr(snap, "day", None), "previous_close", None)
+
                     self._cache.update(
                         ticker=snap.ticker,
                         price=price,
                         timestamp=timestamp,
+                        anchor=anchor,
                     )
                     processed += 1
                 except (AttributeError, TypeError) as e:

backend/app/market/models.py

@@ -13,37 +13,57 @@ class PriceUpdate:
     ticker: str
     price: float
     previous_price: float
+    anchor: 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 previous update (tick-to-tick)."""
         return round(self.price - self.previous_price, 4)
 
     @property
     def change_percent(self) -> float:
-        """Percentage change from previous update."""
+        """Percentage change from previous update (tick-to-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'."""
+        """'up', 'down', or 'flat' — tick-to-tick, drives the flash animation."""
         if self.price > self.previous_price:
             return "up"
         elif self.price < self.previous_price:
             return "down"
         return "flat"
 
+    @property
+    def day_change(self) -> float:
+        """Absolute change vs. the day-change anchor (previous close / first observed)."""
+        return round(self.price - self.anchor, 4)
+
+    @property
+    def day_change_percent(self) -> float:
+        """Percentage change vs. the day-change anchor.
+
+        This is the "% change" shown next to each ticker in the watchlist —
+        NOT change_percent, which is tick-to-tick.
+        """
+        if self.anchor == 0:
+            return 0.0
+        return round((self.price - self.anchor) / self.anchor * 100, 4)
+
     def to_dict(self) -> dict:
         """Serialize for JSON / SSE transmission."""
         return {
             "ticker": self.ticker,
             "price": self.price,
             "previous_price": self.previous_price,
+            "anchor": self.anchor,
             "timestamp": self.timestamp,
             "change": self.change,
             "change_percent": self.change_percent,
             "direction": self.direction,
+            "day_change": self.day_change,
+            "day_change_percent": self.day_change_percent,
         }

backend/app/market/reconcile.py

@@ -0,0 +1,78 @@
+"""Keeps a MarketDataSource's tracked ticker set in sync with watchlist ∪ open positions.
+
+The tracked set for price streaming is always the union of watchlist tickers and
+tickers with an open position, so a position is never left unpriced even if its
+ticker is removed from the watchlist. This module owns that invariant in one place;
+platform code (watchlist routes, trade routes, startup) calls these helpers instead
+of calling `MarketDataSource.add_ticker` / `remove_ticker` directly.
+
+These helpers are deliberately DB-agnostic: `db` is any object exposing the four
+async methods used below (`get_watchlist_tickers`, `get_open_position_tickers`,
+`get_position`, `is_on_watchlist`). The platform layer's database module supplies
+the concrete implementation; tests use a lightweight fake.
+"""
+
+from __future__ import annotations
+
+from typing import Protocol
+
+from .interface import MarketDataSource
+
+
+class Position(Protocol):
+    quantity: float
+
+
+class TrackedTickerStore(Protocol):
+    """The subset of the database API this module depends on."""
+
+    async def get_watchlist_tickers(self) -> list[str]: ...
+
+    async def get_open_position_tickers(self) -> list[str]: ...
+
+    async def get_position(self, ticker: str) -> Position | None: ...
+
+    async def is_on_watchlist(self, ticker: str) -> bool: ...
+
+
+async def get_tracked_tickers(db: TrackedTickerStore) -> list[str]:
+    """The full tracked set: every watchlist ticker plus every ticker with an open
+    position, deduplicated. Used at startup and anywhere the full set needs
+    recomputing from scratch."""
+    watchlist = await db.get_watchlist_tickers()
+    positions = await db.get_open_position_tickers()
+    return sorted(set(watchlist) | set(positions))
+
+
+async def on_watchlist_add(source: MarketDataSource, ticker: str) -> None:
+    """Call after inserting a new watchlist row. Idempotent — add_ticker() on both
+    sources is already a no-op if the ticker is already tracked (e.g. via an open
+    position)."""
+    await source.add_ticker(ticker)
+
+
+async def on_watchlist_remove(source: MarketDataSource, db: TrackedTickerStore, ticker: str) -> None:
+    """Call after deleting a watchlist row. Only stops tracking if there is no open
+    position for this ticker — an open position keeps it priced even off the
+    watchlist."""
+    position = await db.get_position(ticker)
+    if position is None or position.quantity == 0:
+        await source.remove_ticker(ticker)
+
+
+async def on_trade_executed(source: MarketDataSource, db: TrackedTickerStore, ticker: str) -> None:
+    """Call after every trade commits (buy or sell). Covers two edge cases the
+    watchlist-add/remove hooks alone don't handle:
+
+    1. A buy opens a *new* ticker not on the watchlist -> it must start being tracked.
+    2. A sell reduces a ticker's quantity to 0, and that ticker had already been
+       removed from the watchlist earlier while the position was still open -> now
+       that nothing references it, stop tracking it.
+    """
+    position = await db.get_position(ticker)
+    on_watchlist = await db.is_on_watchlist(ticker)
+
+    if position and position.quantity > 0:
+        await source.add_ticker(ticker)  # covers case 1; no-op if already tracked
+    elif not on_watchlist:
+        await source.remove_ticker(ticker)  # covers case 2

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,8 @@
 
 router = APIRouter(prefix="/api/stream", tags=["streaming"])
 
+KEEPALIVE_INTERVAL = 15.0  # seconds
+
 
 def create_stream_router(price_cache: PriceCache) -> APIRouter:
     """Create the SSE streaming router with a reference to the price cache.
@@ -52,16 +55,21 @@ async def _generate_events(
     price_cache: PriceCache,
     request: Request,
     interval: float = 0.5,
+    keepalive_interval: float = KEEPALIVE_INTERVAL,
 ) -> 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()).
+    disconnects (detected via request.is_disconnected()). Sends a
+    `: keepalive` comment whenever `keepalive_interval` seconds pass with
+    no price data sent, so the client can distinguish an idle stream
+    (nothing changed) from a dropped connection.
     """
     # Tell the client to retry after 1 second if the connection drops
     yield "retry: 1000\n\n"
 
     last_version = -1
+    last_send = time.monotonic()
     client_ip = request.client.host if request.client else "unknown"
     logger.info("SSE client connected: %s", client_ip)
 
@@ -72,6 +80,7 @@ async def _generate_events(
                 logger.info("SSE client disconnected: %s", client_ip)
                 break
 
+            now = time.monotonic()
             current_version = price_cache.version
             if current_version != last_version:
                 last_version = current_version
@@ -81,6 +90,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_send = now
+            elif now - last_send >= keepalive_interval:
+                yield ": keepalive\n\n"
+                last_send = now
 
             await asyncio.sleep(interval)
     except asyncio.CancelledError:

backend/app/market/validation.py

@@ -0,0 +1,29 @@
+"""Ticker symbol validation shared by every write path that accepts a ticker."""
+
+from __future__ import annotations
+
+import re
+
+_TICKER_RE = re.compile(r"^[A-Z]{1,5}$")
+
+
+class InvalidTickerError(ValueError):
+    """Raised when a ticker does not match the 1-5 uppercase letter format."""
+
+
+def validate_ticker(raw: str) -> str:
+    """Normalize and validate a ticker symbol.
+
+    Uppercases and strips whitespace, then requires 1-5 letters A-Z.
+    Returns the normalized ticker on success; raises InvalidTickerError otherwise.
+
+    Callers (every write path that accepts a ticker):
+        - POST /api/watchlist              (manual watchlist add)
+        - POST /api/portfolio/trade         (manual trade)
+        - LLM `trades[].ticker`             (chat-initiated trade)
+        - LLM `watchlist_changes[].ticker`  (chat-initiated watchlist change)
+    """
+    ticker = raw.strip().upper()
+    if not _TICKER_RE.match(ticker):
+        raise InvalidTickerError(f"Invalid ticker '{raw}': must be 1-5 letters (A-Z).")
+    return ticker