diff --git a/backend/CLAUDE.md b/backend/CLAUDE.md
index 612ff18f5..492a6438c 100644
--- a/backend/CLAUDE.md
+++ b/backend/CLAUDE.md
@@ -17,20 +17,25 @@ 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
@@ -38,6 +43,7 @@ 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
diff --git a/backend/app/market/__init__.py b/backend/app/market/__init__.py
index 57ad0a121..011a222fb 100644
--- a/backend/app/market/__init__.py
+++ b/backend/app/market/__init__.py
@@ -1,18 +1,28 @@
 """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",
@@ -20,4 +30,10 @@
     "MarketDataSource",
     "create_market_data_source",
     "create_stream_router",
+    "get_tracked_tickers",
+    "on_watchlist_add",
+    "on_watchlist_remove",
+    "on_trade_executed",
+    "validate_ticker",
+    "InvalidTickerError",
 ]
diff --git a/backend/app/market/cache.py b/backend/app/market/cache.py
index 4d0215778..67ad714ce 100644
--- a/backend/app/market/cache.py
+++ b/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:
diff --git a/backend/app/market/massive_client.py b/backend/app/market/massive_client.py
index 00bc7b2aa..b986f16f1 100644
--- a/backend/app/market/massive_client.py
+++ b/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:
diff --git a/backend/app/market/models.py b/backend/app/market/models.py
index de81b1dbc..4651d04f6 100644
--- a/backend/app/market/models.py
+++ b/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,
         }
diff --git a/backend/app/market/reconcile.py b/backend/app/market/reconcile.py
new file mode 100644
index 000000000..0993ced1d
--- /dev/null
+++ b/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
diff --git a/backend/app/market/stream.py b/backend/app/market/stream.py
index 7fd974b7c..c1c60882a 100644
--- a/backend/app/market/stream.py
+++ b/backend/app/market/stream.py
@@ -5,6 +5,7 @@
 import asyncio
 import json
 import logging
+import time
 from collections.abc import AsyncGenerator
 
 from fastapi import APIRouter, Request
@@ -16,6 +17,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:
diff --git a/backend/app/market/validation.py b/backend/app/market/validation.py
new file mode 100644
index 000000000..ebcb3a296
--- /dev/null
+++ b/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
diff --git a/backend/tests/market/test_cache.py b/backend/tests/market/test_cache.py
index b5ab3d55d..33e07d47b 100644
--- a/backend/tests/market/test_cache.py
+++ b/backend/tests/market/test_cache.py
@@ -101,3 +101,58 @@ def test_price_rounding(self):
         cache = PriceCache()
         update = cache.update("AAPL", 190.12345)
         assert update.price == 190.12
+
+
+class TestPriceCacheAnchor:
+    """Unit tests for the day-change anchor tracked alongside each ticker's price."""
+
+    def test_first_update_sets_anchor_to_price(self):
+        """With no explicit anchor, the first observed price becomes the anchor."""
+        cache = PriceCache()
+        update = cache.update("AAPL", 190.00)
+        assert update.anchor == 190.00
+
+    def test_explicit_anchor_used_on_first_update(self):
+        cache = PriceCache()
+        update = cache.update("AAPL", 190.00, anchor=185.50)
+        assert update.anchor == 185.50
+
+    def test_anchor_is_sticky_across_updates(self):
+        """A later call's anchor argument is ignored once one has been captured."""
+        cache = PriceCache()
+        cache.update("AAPL", 190.00, anchor=185.50)
+        update = cache.update("AAPL", 192.00, anchor=999.00)
+        assert update.anchor == 185.50
+
+    def test_anchor_sticky_without_explicit_anchor_on_later_calls(self):
+        """Later calls without an anchor argument still keep the first-captured anchor."""
+        cache = PriceCache()
+        cache.update("AAPL", 190.00, anchor=185.50)
+        update = cache.update("AAPL", 192.00)
+        assert update.anchor == 185.50
+
+    def test_day_change_percent(self):
+        cache = PriceCache()
+        cache.update("AAPL", 190.00, anchor=100.00)
+        update = cache.update("AAPL", 200.00)
+        assert update.day_change == 100.00
+        assert update.day_change_percent == 100.0
+
+    def test_remove_drops_anchor(self):
+        """Removing a ticker clears its anchor so re-tracking re-anchors fresh."""
+        cache = PriceCache()
+        cache.update("AAPL", 190.00, anchor=185.50)
+        cache.remove("AAPL")
+        update = cache.update("AAPL", 300.00)
+        assert update.anchor == 300.00
+
+    def test_get_anchor(self):
+        cache = PriceCache()
+        cache.update("AAPL", 190.00, anchor=185.50)
+        assert cache.get_anchor("AAPL") == 185.50
+        assert cache.get_anchor("NOPE") is None
+
+    def test_anchor_rounded_to_two_decimals(self):
+        cache = PriceCache()
+        update = cache.update("AAPL", 190.00, anchor=185.5049)
+        assert update.anchor == 185.50
diff --git a/backend/tests/market/test_massive.py b/backend/tests/market/test_massive.py
index cdd7dbd24..e607a7665 100644
--- a/backend/tests/market/test_massive.py
+++ b/backend/tests/market/test_massive.py
@@ -8,13 +8,17 @@
 from app.market.massive_client import MassiveDataSource
 
 
-def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock:
+def _make_snapshot(
+    ticker: str, price: float, timestamp_ms: int, previous_close: float | None = None
+) -> MagicMock:
     """Create a mock Massive snapshot object."""
     snap = MagicMock()
     snap.ticker = ticker
     snap.last_trade = MagicMock()
     snap.last_trade.price = price
     snap.last_trade.timestamp = timestamp_ms
+    snap.day = MagicMock()
+    snap.day.previous_close = previous_close
     return snap
 
 
@@ -199,3 +203,48 @@ async def test_start_immediate_poll(self):
         assert cache.get_price("AAPL") == 190.50
 
         await source.stop()
+
+    async def test_poll_captures_previous_close_as_anchor(self):
+        """Test that the snapshot's previous close is captured as the day-change anchor."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()
+
+        snap = _make_snapshot("AAPL", 190.50, 1707580800000, previous_close=185.00)
+
+        with patch.object(source, "_fetch_snapshots", return_value=[snap]):
+            await source._poll_once()
+
+        assert cache.get_anchor("AAPL") == 185.00
+
+    async def test_missing_previous_close_falls_back_to_first_observed(self):
+        """Test that a missing previous close falls back to the first observed price."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()
+
+        snap = _make_snapshot("AAPL", 190.50, 1707580800000, previous_close=None)
+
+        with patch.object(source, "_fetch_snapshots", return_value=[snap]):
+            await source._poll_once()
+
+        assert cache.get_anchor("AAPL") == 190.50
+
+    async def test_anchor_stays_sticky_across_polls(self):
+        """Test that a later poll's previous_close does not overwrite the captured anchor."""
+        cache = PriceCache()
+        source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=60.0)
+        source._tickers = ["AAPL"]
+        source._client = MagicMock()
+
+        first = _make_snapshot("AAPL", 190.50, 1707580800000, previous_close=185.00)
+        with patch.object(source, "_fetch_snapshots", return_value=[first]):
+            await source._poll_once()
+
+        second = _make_snapshot("AAPL", 191.00, 1707580860000, previous_close=999.00)
+        with patch.object(source, "_fetch_snapshots", return_value=[second]):
+            await source._poll_once()
+
+        assert cache.get_anchor("AAPL") == 185.00
diff --git a/backend/tests/market/test_models.py b/backend/tests/market/test_models.py
index 21600dfd6..e890f39b5 100644
--- a/backend/tests/market/test_models.py
+++ b/backend/tests/market/test_models.py
@@ -10,68 +10,135 @@ 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, anchor=190.00, timestamp=1234567890.0
+        )
         assert update.ticker == "AAPL"
         assert update.price == 190.50
         assert update.previous_price == 190.00
+        assert update.anchor == 190.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, anchor=190.00, timestamp=1234567890.0
+        )
         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, anchor=190.00, timestamp=1234567890.0
+        )
         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, anchor=100.00, timestamp=1234567890.0
+        )
         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, anchor=200.00, timestamp=1234567890.0
+        )
         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, anchor=100.00, timestamp=1234567890.0
+        )
         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, anchor=190.00, timestamp=1234567890.0
+        )
         assert update.direction == "up"
 
     def test_direction_down(self):
         """Test direction calculation (down)."""
-        update = PriceUpdate(ticker="AAPL", price=189.00, previous_price=190.00, timestamp=1234567890.0)
+        update = PriceUpdate(
+            ticker="AAPL", price=189.00, previous_price=190.00, anchor=190.00, timestamp=1234567890.0
+        )
         assert update.direction == "down"
 
     def test_direction_flat(self):
         """Test direction calculation (flat)."""
-        update = PriceUpdate(ticker="AAPL", price=190.00, previous_price=190.00, timestamp=1234567890.0)
+        update = PriceUpdate(
+            ticker="AAPL", price=190.00, previous_price=190.00, anchor=190.00, timestamp=1234567890.0
+        )
         assert update.direction == "flat"
 
     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, anchor=190.00, timestamp=1234567890.0
+        )
         result = update.to_dict()
 
         assert result["ticker"] == "AAPL"
         assert result["price"] == 190.50
         assert result["previous_price"] == 190.00
+        assert result["anchor"] == 190.00
         assert result["timestamp"] == 1234567890.0
         assert result["change"] == 0.50
         assert result["change_percent"] == 0.2632  # (0.50 / 190.00) * 100
         assert result["direction"] == "up"
+        assert result["day_change"] == 0.50
+        assert result["day_change_percent"] == 0.2632
 
     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, anchor=190.00, timestamp=1234567890.0
+        )
 
         with pytest.raises(AttributeError):
             update.price = 200.00  # Should raise error
+
+
+class TestPriceUpdateDayChange:
+    """Unit tests for the day-change (anchor-relative) properties."""
+
+    def test_day_change_calculation(self):
+        """day_change measures the move vs. the anchor, not the previous tick."""
+        update = PriceUpdate(
+            ticker="AAPL", price=200.0, previous_price=198.0, anchor=190.0, timestamp=1234567890.0
+        )
+        assert update.day_change == 10.0
+
+    def test_day_change_percent_calculation(self):
+        update = PriceUpdate(
+            ticker="AAPL", price=200.0, previous_price=198.0, anchor=190.0, timestamp=1234567890.0
+        )
+        assert update.day_change_percent == pytest.approx(5.2632, rel=1e-3)
+
+    def test_day_change_percent_zero_anchor_is_safe(self):
+        """A zero anchor must not raise ZeroDivisionError."""
+        update = PriceUpdate(
+            ticker="X", price=10.0, previous_price=10.0, anchor=0.0, timestamp=1234567890.0
+        )
+        assert update.day_change_percent == 0.0
+
+    def test_day_change_independent_of_tick_change(self):
+        """Tick-to-tick change and day-change can point in different directions."""
+        update = PriceUpdate(
+            ticker="AAPL", price=189.0, previous_price=190.0, anchor=180.0, timestamp=1234567890.0
+        )
+        assert update.direction == "down"  # down vs. previous tick
+        assert update.day_change_percent > 0  # but up vs. the day's anchor
+
+    def test_to_dict_includes_anchor_fields(self):
+        update = PriceUpdate(
+            ticker="AAPL", price=200.0, previous_price=198.0, anchor=190.0, timestamp=1234567890.0
+        )
+        d = update.to_dict()
+        assert d["anchor"] == 190.0
+        assert d["day_change"] == 10.0
+        assert d["day_change_percent"] == pytest.approx(5.2632, rel=1e-3)
diff --git a/backend/tests/market/test_reconcile.py b/backend/tests/market/test_reconcile.py
new file mode 100644
index 000000000..048b49d91
--- /dev/null
+++ b/backend/tests/market/test_reconcile.py
@@ -0,0 +1,157 @@
+"""Tests for the watchlist <-> open-position tracked-ticker reconciliation helpers."""
+
+from dataclasses import dataclass
+
+import pytest
+
+from app.market.reconcile import (
+    get_tracked_tickers,
+    on_trade_executed,
+    on_watchlist_add,
+    on_watchlist_remove,
+)
+
+
+@dataclass
+class FakePosition:
+    quantity: float
+
+
+class FakeDB:
+    """In-memory stand-in for the database, exposing only what reconcile.py needs."""
+
+    def __init__(self, watchlist=None, positions=None):
+        self._watchlist: set[str] = set(watchlist or [])
+        self._positions: dict[str, FakePosition] = dict(positions or {})
+
+    def set_position(self, ticker: str, quantity: float) -> None:
+        self._positions[ticker] = FakePosition(quantity=quantity)
+
+    def set_on_watchlist(self, ticker: str, on: bool) -> None:
+        if on:
+            self._watchlist.add(ticker)
+        else:
+            self._watchlist.discard(ticker)
+
+    async def get_watchlist_tickers(self) -> list[str]:
+        return sorted(self._watchlist)
+
+    async def get_open_position_tickers(self) -> list[str]:
+        return sorted(t for t, p in self._positions.items() if p.quantity > 0)
+
+    async def get_position(self, ticker: str) -> FakePosition | None:
+        return self._positions.get(ticker)
+
+    async def is_on_watchlist(self, ticker: str) -> bool:
+        return ticker in self._watchlist
+
+
+class FakeSource:
+    """Records add_ticker/remove_ticker calls without touching a real cache."""
+
+    def __init__(self):
+        self.added: list[str] = []
+        self.removed: list[str] = []
+
+    async def add_ticker(self, ticker: str) -> None:
+        self.added.append(ticker)
+
+    async def remove_ticker(self, ticker: str) -> None:
+        self.removed.append(ticker)
+
+    @property
+    def add_called(self) -> bool:
+        return bool(self.added)
+
+    @property
+    def remove_called(self) -> bool:
+        return bool(self.removed)
+
+
+@pytest.mark.asyncio
+class TestGetTrackedTickers:
+    async def test_union_of_watchlist_and_positions(self):
+        db = FakeDB(watchlist=["AAPL", "GOOGL"], positions={"TSLA": FakePosition(5)})
+        tracked = await get_tracked_tickers(db)
+        assert tracked == ["AAPL", "GOOGL", "TSLA"]
+
+    async def test_deduplicates_overlap(self):
+        db = FakeDB(watchlist=["AAPL"], positions={"AAPL": FakePosition(5)})
+        tracked = await get_tracked_tickers(db)
+        assert tracked == ["AAPL"]
+
+    async def test_excludes_zero_quantity_positions(self):
+        db = FakeDB(watchlist=[], positions={"AAPL": FakePosition(0)})
+        tracked = await get_tracked_tickers(db)
+        assert tracked == []
+
+    async def test_empty_everything(self):
+        db = FakeDB()
+        assert await get_tracked_tickers(db) == []
+
+
+@pytest.mark.asyncio
+class TestOnWatchlistAdd:
+    async def test_adds_to_source(self):
+        source = FakeSource()
+        await on_watchlist_add(source, "AAPL")
+        assert source.added == ["AAPL"]
+
+
+@pytest.mark.asyncio
+class TestOnWatchlistRemove:
+    async def test_remove_watchlist_keeps_open_position(self):
+        db = FakeDB()
+        db.set_position("AAPL", quantity=10)
+        source = FakeSource()
+        await on_watchlist_remove(source, db, "AAPL")
+        assert not source.remove_called
+
+    async def test_remove_watchlist_no_position_removes(self):
+        db = FakeDB()
+        db.set_position("AAPL", quantity=0)
+        source = FakeSource()
+        await on_watchlist_remove(source, db, "AAPL")
+        assert source.remove_called
+
+    async def test_remove_watchlist_unknown_ticker_removes(self):
+        db = FakeDB()
+        source = FakeSource()
+        await on_watchlist_remove(source, db, "NOPE")
+        assert source.remove_called
+
+
+@pytest.mark.asyncio
+class TestOnTradeExecuted:
+    async def test_trade_opens_new_ticker_not_on_watchlist(self):
+        db = FakeDB()
+        db.set_position("PYPL", quantity=5)
+        db.set_on_watchlist("PYPL", False)
+        source = FakeSource()
+        await on_trade_executed(source, db, "PYPL")
+        assert source.added == ["PYPL"]
+
+    async def test_sell_to_zero_off_watchlist_removes(self):
+        db = FakeDB()
+        db.set_position("PYPL", quantity=0)
+        db.set_on_watchlist("PYPL", False)
+        source = FakeSource()
+        await on_trade_executed(source, db, "PYPL")
+        assert source.removed == ["PYPL"]
+
+    async def test_sell_to_zero_still_on_watchlist_keeps_tracked(self):
+        db = FakeDB()
+        db.set_position("AAPL", quantity=0)
+        db.set_on_watchlist("AAPL", True)
+        source = FakeSource()
+        await on_trade_executed(source, db, "AAPL")
+        assert not source.remove_called
+        assert not source.add_called
+
+    async def test_buy_more_of_existing_open_position(self):
+        db = FakeDB()
+        db.set_position("AAPL", quantity=15)
+        db.set_on_watchlist("AAPL", True)
+        source = FakeSource()
+        await on_trade_executed(source, db, "AAPL")
+        assert source.added == ["AAPL"]
diff --git a/backend/tests/market/test_stream.py b/backend/tests/market/test_stream.py
new file mode 100644
index 000000000..38f9cb76c
--- /dev/null
+++ b/backend/tests/market/test_stream.py
@@ -0,0 +1,134 @@
+"""Tests for the SSE streaming generator."""
+
+import json
+
+import pytest
+
+from app.market.cache import PriceCache
+from app.market.stream import _generate_events
+
+
+class FakeRequest:
+    """Minimal stand-in for fastapi.Request, enough for _generate_events()."""
+
+    class _Client:
+        host = "127.0.0.1"
+
+    def __init__(self, disconnect_after: int | None = None):
+        self.client = self._Client()
+        self._calls = 0
+        self._disconnect_after = disconnect_after
+
+    async def is_disconnected(self) -> bool:
+        self._calls += 1
+        if self._disconnect_after is None:
+            return False
+        return self._calls > self._disconnect_after
+
+
+async def _collect(generator, max_events: int) -> list[str]:
+    """Pull up to `max_events` from an async generator, then close it."""
+    events = []
+    async for event in generator:
+        events.append(event)
+        if len(events) >= max_events:
+            break
+    await generator.aclose()
+    return events
+
+
+@pytest.mark.asyncio
+class TestGenerateEvents:
+    """Unit tests for _generate_events()."""
+
+    async def test_first_event_is_retry_directive(self):
+        cache = PriceCache()
+        request = FakeRequest()
+        events = await _collect(
+            _generate_events(cache, request, interval=0.01, keepalive_interval=1.0), 1
+        )
+        assert events[0] == "retry: 1000\n\n"
+
+    async def test_data_event_sent_when_cache_has_prices(self):
+        cache = PriceCache()
+        cache.update("AAPL", 190.0)
+        request = FakeRequest()
+        events = await _collect(
+            _generate_events(cache, request, interval=0.01, keepalive_interval=1.0), 2
+        )
+        data_events = [e for e in events if e.startswith("data:")]
+        assert len(data_events) == 1
+        payload = json.loads(data_events[0][len("data: ") :].strip())
+        assert payload["AAPL"]["price"] == 190.0
+
+    async def test_no_data_event_when_cache_empty(self):
+        cache = PriceCache()
+        request = FakeRequest(disconnect_after=5)
+        events = [
+            e
+            async for e in _generate_events(cache, request, interval=0.001, keepalive_interval=1.0)
+        ]
+        assert not any(e.startswith("data:") for e in events)
+
+    async def test_keepalive_sent_when_cache_idle(self):
+        cache = PriceCache()
+        cache.update("AAPL", 190.0)  # one real event, then silence
+
+        request = FakeRequest(disconnect_after=20)
+        events = [
+            e
+            async for e in _generate_events(
+                cache, request, interval=0.01, keepalive_interval=0.02
+            )
+        ]
+        assert any(e.startswith(": keepalive") for e in events)
+
+    async def test_no_duplicate_data_events_without_version_change(self):
+        cache = PriceCache()
+        cache.update("AAPL", 190.0)
+        request = FakeRequest(disconnect_after=10)
+        events = [
+            e
+            async for e in _generate_events(cache, request, interval=0.001, keepalive_interval=5.0)
+        ]
+        data_events = [e for e in events if e.startswith("data:")]
+        assert len(data_events) == 1
+
+    async def test_stream_stops_on_disconnect(self):
+        cache = PriceCache()
+        request = FakeRequest(disconnect_after=2)
+        events = [
+            e
+            async for e in _generate_events(cache, request, interval=0.001, keepalive_interval=5.0)
+        ]
+        # Generator must terminate on its own (the test would hang otherwise).
+        assert isinstance(events, list)
+
+    async def test_new_price_after_idle_period_sent_as_data(self):
+        """A price change after a quiet period is sent as a fresh data event.
+
+        Iterates the generator manually (never calling aclose()) so it stays
+        alive across the cache mutation in the middle of the test.
+        """
+        cache = PriceCache()
+        cache.update("AAPL", 190.0)
+        request = FakeRequest(disconnect_after=15)
+
+        gen = _generate_events(cache, request, interval=0.005, keepalive_interval=0.01)
+        events = []
+        async for event in gen:
+            events.append(event)
+            if len(events) >= 2:  # retry directive + the initial data event
+                break
+
+        cache.update("AAPL", 191.0)
+
+        async for event in gen:
+            events.append(event)
+            if event.startswith("data:") and "191.0" in event:
+                break
+
+        data_events = [e for e in events if e.startswith("data:")]
+        assert len(data_events) >= 2
+        last_payload = json.loads(data_events[-1][len("data: ") :].strip())
+        assert last_payload["AAPL"]["price"] == 191.0
diff --git a/backend/tests/market/test_validation.py b/backend/tests/market/test_validation.py
new file mode 100644
index 000000000..34fa22876
--- /dev/null
+++ b/backend/tests/market/test_validation.py
@@ -0,0 +1,39 @@
+"""Tests for ticker symbol validation."""
+
+import pytest
+
+from app.market.validation import InvalidTickerError, validate_ticker
+
+
+class TestValidateTicker:
+    """Unit tests for validate_ticker()."""
+
+    @pytest.mark.parametrize(
+        "raw,expected",
+        [
+            ("aapl", "AAPL"),
+            (" TSLA ", "TSLA"),
+            ("V", "V"),
+            ("v", "V"),
+            ("GOOGL", "GOOGL"),
+            ("goog", "GOOG"),
+        ],
+    )
+    def test_valid_tickers_normalize(self, raw, expected):
+        assert validate_ticker(raw) == expected
+
+    @pytest.mark.parametrize(
+        "raw",
+        ["", "   ", "TOOLONG", "AB3", "AB-C", "aapl$", "123", "A B", "TOO-LONG-NAME"],
+    )
+    def test_invalid_tickers_raise(self, raw):
+        with pytest.raises(InvalidTickerError):
+            validate_ticker(raw)
+
+    def test_error_message_includes_original_input(self):
+        with pytest.raises(InvalidTickerError, match="NOT-A-TICKER"):
+            validate_ticker("NOT-A-TICKER")
+
+    def test_invalid_ticker_error_is_value_error(self):
+        """InvalidTickerError should be catchable as a ValueError by generic handlers."""
+        assert issubclass(InvalidTickerError, ValueError)
diff --git a/planning/MARKET_DATA_SUMMARY.md b/planning/MARKET_DATA_SUMMARY.md
index ae518283a..dfb004fc3 100644
--- a/planning/MARKET_DATA_SUMMARY.md
+++ b/planning/MARKET_DATA_SUMMARY.md
@@ -1,10 +1,13 @@
 # Market Data Backend — Summary
 
-**Status:** Complete, tested, reviewed, all issues resolved.
+**Status:** Complete, tested, reviewed, all issues resolved. Updated to also cover the four
+follow-up items `planning/MARKET_DATA_DESIGN.md` and `planning/REVIEW.md` flagged as
+not-yet-built: the day-change **anchor**, ticker-format **validation**, SSE **keepalive**, and
+the watchlist ∪ open-positions **reconciliation** helper. See "Follow-Up Additions" below.
 
 ## What Was Built
 
-A complete market data subsystem in `backend/app/market/` (8 modules, ~500 lines) providing live price simulation and real market data via a unified interface.
+A complete market data subsystem in `backend/app/market/` (10 modules, ~700 lines) providing live price simulation and real market data via a unified interface.
 
 ### Architecture
 
@@ -32,7 +35,9 @@ MarketDataSource (ABC)
 | `simulator.py` | `GBMSimulator` (Geometric Brownian Motion with Cholesky-correlated moves) + `SimulatorDataSource` |
 | `massive_client.py` | `MassiveDataSource` — REST polling client for Polygon.io via the `massive` package |
 | `factory.py` | `create_market_data_source()` — selects simulator or Massive based on `MASSIVE_API_KEY` env var |
-| `stream.py` | `create_stream_router()` — FastAPI SSE endpoint factory using version-based change detection |
+| `stream.py` | `create_stream_router()` — FastAPI SSE endpoint factory using version-based change detection, plus a ~15s `: keepalive` comment when the cache is idle |
+| `validation.py` | `validate_ticker()` / `InvalidTickerError` — normalizes and enforces the 1-5 uppercase letter ticker format at every write path |
+| `reconcile.py` | `get_tracked_tickers()` / `on_watchlist_add()` / `on_watchlist_remove()` / `on_trade_executed()` — keeps the market source's tracked set in sync with watchlist ∪ open positions; DB-agnostic (duck-typed), for the platform layer to call |
 
 ### Key Design Decisions
 
@@ -44,18 +49,21 @@ MarketDataSource (ABC)
 
 ## Test Suite
 
-**73 tests, all passing.** 6 test modules in `backend/tests/market/`.
-
-| Module | Tests | Coverage |
-|--------|-------|----------|
-| test_models.py | 11 | models.py: 100% |
-| test_cache.py | 13 | cache.py: 100% |
-| test_simulator.py | 17 | simulator.py: 98% |
-| test_simulator_source.py | 10 | (integration tests) |
-| test_factory.py | 7 | factory.py: 100% |
-| test_massive.py | 13 | massive_client.py: 56% (expected — API methods mocked) |
-
-Overall coverage: 84%.
+9 test modules in `backend/tests/market/` (originally 73 tests across 6 modules; extended with
+anchor/validation/keepalive/reconcile coverage below — run `uv run --extra dev pytest -v` to get
+the current count and coverage in your environment).
+
+| Module | Covers |
+|--------|--------|
+| test_models.py | `PriceUpdate`, including day-change (`anchor`, `day_change`, `day_change_percent`) |
+| test_cache.py | `PriceCache`, including anchor capture/stickiness/eviction (`TestPriceCacheAnchor`) |
+| test_simulator.py | `GBMSimulator` math, correlation, unknown-ticker synthesis |
+| test_simulator_source.py | `SimulatorDataSource` integration |
+| test_factory.py | `create_market_data_source()` selection logic |
+| test_massive.py | `MassiveDataSource`, including previous-close → anchor plumbing |
+| test_validation.py | `validate_ticker()` / `InvalidTickerError` — new |
+| test_stream.py | `_generate_events()` SSE loop, including the keepalive timer — new |
+| test_reconcile.py | tracked-ticker reconciliation helpers against a fake DB/source — new |
 
 ## Code Review & Fixes Applied
 
@@ -69,6 +77,36 @@ A comprehensive code review identified 7 issues. All were resolved:
 6. **Unused test imports removed** — `pytest`, `math`, `asyncio` cleaned from 4 test files
 7. **Massive test mocks fixed** — `source._client` set in tests, patches target correct names
 
+## Follow-Up Additions (day-change anchor, validation, keepalive, reconciliation)
+
+`planning/MARKET_DATA_DESIGN.md` and `planning/REVIEW.md` identified four pieces PLAN.md §6 and
+the Design Decisions Log call for that were not in the original build. All four are now
+implemented, per the design doc's specification:
+
+1. **Day-change anchor** — `PriceUpdate` gained a required `anchor` field plus `day_change` /
+   `day_change_percent` properties (distinct from the tick-to-tick `change` / `change_percent`).
+   `PriceCache` captures the anchor on a ticker's first write (previous close if Massive supplies
+   one via `snap.day.previous_close`, otherwise the first observed price) and keeps it sticky
+   across later updates; `remove()` drops it so re-tracking re-anchors fresh. `to_dict()` — and so
+   the SSE payload — now includes `anchor`, `day_change`, `day_change_percent`.
+2. **Ticker validation** — new `validation.py` with `validate_ticker()` / `InvalidTickerError`,
+   enforcing the `^[A-Z]{1,5}$` format. Scoped to the API boundary per the design doc: the
+   simulator's unknown-ticker synthesis (`SEED_PRICES.get(ticker, random.uniform(50, 300))`)
+   stays permissive internally and was already correct.
+3. **SSE keepalive** — `stream.py`'s `_generate_events()` now tracks the wall-clock time since
+   the last byte sent and emits a `: keepalive\n\n` comment after `KEEPALIVE_INTERVAL` (15s,
+   configurable) of no price changes, so the frontend's connection-status indicator can tell an
+   idle stream from a dropped one.
+4. **Watchlist ∪ open-positions reconciliation** — new `reconcile.py` with `get_tracked_tickers()`,
+   `on_watchlist_add()`, `on_watchlist_remove()`, `on_trade_executed()`. These are intentionally
+   DB-agnostic (a `TrackedTickerStore` `Protocol` with `get_watchlist_tickers` /
+   `get_open_position_tickers` / `get_position` / `is_on_watchlist`) since no database module
+   exists yet — the backend platform agent wires in the real DB implementation and calls these
+   helpers from the watchlist/trade routes and at startup, per §13 of `MARKET_DATA_DESIGN.md`.
+
+All four are additive (new optional kwargs, new modules) — no existing call site outside the
+market package needed to change.
+
 ## Demo
 
 A Rich terminal demo is available at `backend/market_data_demo.py`:
@@ -102,3 +140,27 @@ await source.remove_ticker("GOOGL")
 # Shutdown
 await source.stop()
 ```
+
+```python
+from app.market import InvalidTickerError, validate_ticker
+
+# At every write path that accepts a ticker (watchlist add, trade, LLM actions)
+try:
+    ticker = validate_ticker(raw_ticker)
+except InvalidTickerError as e:
+    ...  # 400 for manual API calls; folded into the per-action chat result for LLM actions
+```
+
+```python
+from app.market import get_tracked_tickers, on_trade_executed, on_watchlist_add, on_watchlist_remove
+
+# `db` is anything exposing get_watchlist_tickers / get_open_position_tickers /
+# get_position / is_on_watchlist (see reconcile.py) — supplied by the platform's DB module.
+initial_tickers = await get_tracked_tickers(db)
+await source.start(initial_tickers)
+
+# After the DB transaction commits in each write path:
+await on_watchlist_add(source, ticker)
+await on_watchlist_remove(source, db, ticker)
+await on_trade_executed(source, db, ticker)
+```