Claude/market data fixes improvements oc05s7

.claude/settings.json

@@ -5,3 +5,4 @@
     "playwright@claude-plugins-official": true
   }
 }
+

.github/workflows/claude-code-review.yml

@@ -31,14 +31,27 @@ jobs:
         with:
           fetch-depth: 1
 
+      # uv is NOT preinstalled on the GitHub runner image. Without this step
+      # every `uv` command fails with "command not found", regardless of which
+      # tools are allowed below.
+      - name: Install uv
+        uses: astral-sh/setup-uv@v10.0.1
+        with:
+          enable-cache: true
+          working-directory: backend
+
       - name: Run Claude Code Review
         id: claude-review
         uses: anthropics/claude-code-action@v1
         with:
           claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }}
           plugin_marketplaces: 'https://github.com/anthropics/claude-code.git'
           plugins: 'code-review@claude-code-plugins'
-          prompt: '/code-review:code-review ${{ github.repository }}/pull/${{ github.event.pull_request.number }}'
+          prompt: '/code-review:code-review --comment ${{ github.repository }}/pull/${{ github.event.pull_request.number }}'
+          # Bash(cd:*),Bash(uv:*) let the review actually run the test suite
+          # instead of reasoning about the code statically.
+          claude_args: |
+            --allowedTools "mcp__github_inline_comment__create_inline_comment,Bash(cd:*),Bash(uv:*)"
           # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md
           # or https://code.claude.com/docs/en/cli-reference for available options
 

.github/workflows/claude.yml

@@ -19,7 +19,7 @@ jobs:
       (github.event_name == 'issues' && (contains(github.event.issue.body, '@claude') || contains(github.event.issue.title, '@claude')))
     runs-on: ubuntu-latest
     permissions:
-      contents: read
+      contents: write # Lets Claude commit and push fixes, not just read
       pull-requests: read
       issues: read
       id-token: write
@@ -30,6 +30,15 @@ jobs:
         with:
           fetch-depth: 1
 
+      # uv is NOT preinstalled on the GitHub runner image. Without this step
+      # every `uv` command fails with "command not found", regardless of which
+      # tools are allowed below.
+      - name: Install uv
+        uses: astral-sh/setup-uv@v10.0.1
+        with:
+          enable-cache: true
+          working-directory: backend
+
       - name: Run Claude Code
         id: claude
         uses: anthropics/claude-code-action@v1
@@ -43,8 +52,9 @@ jobs:
           # Optional: Give a custom prompt to Claude. If this is not specified, Claude will perform the instructions specified in the comment that tagged it.
           # prompt: 'Update the pull request description to include a summary of changes.'
 
-          # Optional: Add claude_args to customize behavior and configuration
-          # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md
-          # or https://code.claude.com/docs/en/cli-reference for available options
-          # claude_args: '--allowed-tools Bash(gh pr:*)'
-
+          # Lets Claude actually run the test suite. Without these, Bash is
+          # unavailable in a non-interactive run and there is no human to
+          # approve it, so Claude can only review statically.
+          # `cd` is needed because the uv project lives in backend/.
+          claude_args: |
+            --allowedTools "Bash(cd:*),Bash(uv:*)"

.gitignore

@@ -205,3 +205,8 @@ cython_debug/
 marimo/_static/
 marimo/_lsp/
 __marimo__/
+
+# FinAlly
+db/*.db
+db/*.db-wal
+db/*.db-shm

README.md

@@ -32,7 +32,7 @@ cp .env.example .env
 
 # Run with Docker
 docker build -t finally .
-docker run -v finally-data:/app/db -p 8000:8000 --env-file .env finally
+docker run -v "$(pwd)/db:/app/db" -p 8000:8000 --env-file .env finally
 
 # Open http://localhost:8000
 ```

backend/CLAUDE.md

@@ -24,7 +24,8 @@ from app.market import PriceCache, PriceUpdate, MarketDataSource, create_market_
   - `get(ticker) -> PriceUpdate | None`
   - `get_price(ticker) -> float | None`
   - `get_all() -> dict[str, PriceUpdate]`
-  - `remove(ticker)`
+  - `get_history(ticker, limit=HISTORY_MAXLEN) -> list[tuple[float, float]]` — rolling in-memory `(timestamp, price)` points, oldest-first, capped at `HISTORY_MAXLEN` (600, ~5 min at the simulator's 500ms cadence). Empty list for an untracked ticker.
+  - `remove(ticker)` — also clears that ticker's history
   - `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()`.
@@ -40,6 +41,26 @@ router = create_stream_router(price_cache)  # Returns FastAPI APIRouter
 # Endpoint: GET /api/stream/prices (text/event-stream)
 ```
 
+Pushes the entire price cache as one JSON object (map keyed by ticker) whenever
+`PriceCache.version` changes, polled every 500ms. A connecting client always gets
+a full snapshot immediately, including after a reconnect. When the version hasn't
+changed for 15s (`KEEPALIVE_SECONDS`), an SSE comment line (`: ping`) is sent so
+proxies and the frontend's connection indicator don't mistake a quiet market for
+a dead connection.
+
+### Price History
+
+```python
+from app.market import create_history_router
+
+router = create_history_router(price_cache)  # Returns FastAPI APIRouter
+# Endpoint: GET /api/prices/{ticker}/history?limit=600
+```
+
+Backs the main chart's initial backfill from `PriceCache`'s rolling in-memory
+history. Not persisted — a restart clears it, matching the simulator's own
+reset-to-seed behavior.
+
 ### Seed Data
 
 Default tickers: AAPL, GOOGL, MSFT, AMZN, TSLA, NVDA, META, JPM, V, NFLX. Seed prices and per-ticker volatility/drift params are in `app/market/seed_prices.py`.

backend/app/market/__init__.py

@@ -2,22 +2,25 @@
 
 Public API:
     PriceUpdate         - Immutable price snapshot dataclass
-    PriceCache          - Thread-safe in-memory price store
+    PriceCache          - Thread-safe in-memory price store (latest price + rolling history)
     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
+    create_stream_router - FastAPI router factory for the SSE endpoint
+    create_history_router - FastAPI router factory for the price history endpoint
 """
 
-from .cache import PriceCache
+from .cache import HISTORY_MAXLEN, PriceCache
 from .factory import create_market_data_source
 from .interface import MarketDataSource
 from .models import PriceUpdate
-from .stream import create_stream_router
+from .stream import create_history_router, create_stream_router
 
 __all__ = [
     "PriceUpdate",
     "PriceCache",
+    "HISTORY_MAXLEN",
     "MarketDataSource",
     "create_market_data_source",
     "create_stream_router",
+    "create_history_router",
 ]

backend/app/market/cache.py

@@ -3,20 +3,29 @@
 from __future__ import annotations
 
 import time
+from collections import deque
 from threading import Lock
 
 from .models import PriceUpdate
 
+# ~5 minutes of history at the 500ms simulator cadence. Under Massive's 15s
+# poll interval this fills much more slowly (one point per poll), which is
+# correct: the chart is sparse but accurate rather than padded with guesses.
+HISTORY_MAXLEN = 600
+
 
 class PriceCache:
     """Thread-safe in-memory cache of the latest price for each ticker.
 
     Writers: SimulatorDataSource or MassiveDataSource (one at a time).
-    Readers: SSE streaming endpoint, portfolio valuation, trade execution.
+    Readers: SSE streaming endpoint, portfolio valuation, trade execution,
+    the price history endpoint.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, history_maxlen: int = HISTORY_MAXLEN) -> None:
         self._prices: dict[str, PriceUpdate] = {}
+        self._history: dict[str, deque[tuple[float, float]]] = {}
+        self._history_maxlen = history_maxlen
         self._lock = Lock()
         self._version: int = 0  # Monotonically increasing; bumped on every update
 
@@ -25,9 +34,10 @@ def update(self, ticker: str, price: float, timestamp: float | None = None) -> P
 
         Automatically computes direction and change from the previous price.
         If this is the first update for the ticker, previous_price == price (direction='flat').
+        Also appends to the ticker's rolling history (see get_history()).
         """
         with self._lock:
-            ts = timestamp or time.time()
+            ts = timestamp if timestamp is not None else time.time()
             prev = self._prices.get(ticker)
             previous_price = prev.price if prev else price
 
@@ -38,6 +48,13 @@ def update(self, ticker: str, price: float, timestamp: float | None = None) -> P
                 timestamp=ts,
             )
             self._prices[ticker] = update
+
+            history = self._history.get(ticker)
+            if history is None:
+                history = deque(maxlen=self._history_maxlen)
+                self._history[ticker] = history
+            history.append((ts, update.price))
+
             self._version += 1
             return update
 
@@ -60,11 +77,25 @@ def remove(self, ticker: str) -> None:
         """Remove a ticker from the cache (e.g., when removed from watchlist)."""
         with self._lock:
             self._prices.pop(ticker, None)
+            self._history.pop(ticker, None)
+
+    def get_history(self, ticker: str, limit: int = HISTORY_MAXLEN) -> list[tuple[float, float]]:
+        """Oldest-first (timestamp, price) points for a ticker.
+
+        Returns an empty list for an untracked ticker rather than raising, so
+        callers (e.g. the chart) can draw nothing instead of erroring.
+        """
+        with self._lock:
+            points = self._history.get(ticker)
+            if not points:
+                return []
+            return list(points)[-limit:]
 
     @property
     def version(self) -> int:
         """Current version counter. Useful for SSE change detection."""
-        return self._version
+        with self._lock:
+            return self._version
 
     def __len__(self) -> int:
         with self._lock:

backend/app/market/massive_client.py

@@ -4,15 +4,20 @@
 
 import asyncio
 import logging
+import time
 
 from massive import RESTClient
+from massive.exceptions import AuthError, BadResponse
 from massive.rest.models import SnapshotMarketType
 
 from .cache import PriceCache
 from .interface import MarketDataSource
 
 logger = logging.getLogger(__name__)
 
+# Snapshot last_trade.sip_timestamp is Unix nanoseconds; PriceCache wants seconds.
+NANOS_PER_SECOND = 1_000_000_000
+
 
 class MassiveDataSource(MarketDataSource):
     """MarketDataSource backed by the Massive (Polygon.io) REST API.
@@ -37,6 +42,13 @@ def __init__(
         self._tickers: list[str] = []
         self._task: asyncio.Task | None = None
         self._client: RESTClient | None = None
+        # Flipped to False if the poll loop dies (e.g. a revoked API key
+        # raising AuthError after start() already succeeded). Nothing awaits
+        # self._task until stop(), so without this the failure would only
+        # ever surface as an unretrieved-exception log at GC time. A future
+        # GET /api/health can report `market_source` as degraded by reading
+        # this flag.
+        self._healthy = True
 
     async def start(self, tickers: list[str]) -> None:
         self._client = RESTClient(api_key=self._api_key)
@@ -45,7 +57,9 @@ async def start(self, tickers: list[str]) -> None:
         # Do an immediate first poll so the cache has data right away
         await self._poll_once()
 
+        self._healthy = True
         self._task = asyncio.create_task(self._poll_loop(), name="massive-poller")
+        self._task.add_done_callback(self._on_poll_task_done)
         logger.info(
             "Massive poller started: %d tickers, %.1fs interval",
             len(tickers),
@@ -78,8 +92,31 @@ async def remove_ticker(self, ticker: str) -> None:
     def get_tickers(self) -> list[str]:
         return list(self._tickers)
 
+    @property
+    def is_healthy(self) -> bool:
+        """False once the poll loop has died from an unhandled exception.
+
+        A deliberate stop() (which cancels the task) never flips this.
+        """
+        return self._healthy
+
     # --- Internal ---
 
+    def _on_poll_task_done(self, task: asyncio.Task) -> None:
+        """Surface a dead poll loop loudly instead of an easy-to-miss
+        "Task exception was never retrieved" log at garbage-collection time.
+        """
+        if task.cancelled():
+            return
+        exc = task.exception()
+        if exc is not None:
+            self._healthy = False
+            logger.critical(
+                "Massive poller task died unexpectedly, live prices are now frozen: %s",
+                exc,
+                exc_info=exc,
+            )
+
     async def _poll_loop(self) -> None:
         """Poll on interval. First poll already happened in start()."""
         while True:
@@ -95,30 +132,42 @@ 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 AuthError:
+            logger.error("Massive API key rejected — the source does not fall back automatically")
+            raise  # unrecoverable: do not retry on a loop
+        except BadResponse as e:
+            logger.warning("Massive returned an error response: %s", e)
+            return  # transient: retry next interval
+        except Exception:
+            logger.exception("Massive poll failed")
+            return
+
+        processed = self._apply_snapshots(snapshots)
+        logger.debug("Massive poll: updated %d/%d tickers", processed, len(self._tickers))
+
+    def _apply_snapshots(self, snapshots: list) -> int:
+        """Write snapshot data into the cache. Returns the number of tickers updated.
+
+        Extracted from _poll_once so it can be tested directly against real
+        `TickerSnapshot` objects (built via `TickerSnapshot.from_dict(...)`)
+        instead of mocks that would silently accept a misspelled attribute.
+        """
+        processed = 0
+        for snap in snapshots:
+            trade = snap.last_trade
+            if trade is None or trade.price is None:
+                # No trade yet today (pre-market, or an unrecognized symbol
+                # that still made it into the response) — leave it as "—".
+                continue
+            self._cache.update(
+                ticker=snap.ticker,
+                price=trade.price,
+                timestamp=(
+                    trade.sip_timestamp / NANOS_PER_SECOND if trade.sip_timestamp else time.time()
+                ),
+            )
+            processed += 1
+        return processed
 
     def _fetch_snapshots(self) -> list:
         """Synchronous call to the Massive REST API. Runs in a thread."""