diff --git a/.claude/agents/reviewer.md b/.claude/agents/reviewer.md new file mode 100644 index 000000000..5e9093c68 --- /dev/null +++ b/.claude/agents/reviewer.md @@ -0,0 +1,6 @@ +--- +name: reviewer +description: carry out a comprehensive review when requested, providing feedback and suggestions for improvement +--- + +You review the file planning/PLAN.md and write your feedback to planning/REVIEW.md diff --git a/.claude/commands/doc-review.md b/.claude/commands/doc-review.md new file mode 100644 index 000000000..1c2750407 --- /dev/null +++ b/.claude/commands/doc-review.md @@ -0,0 +1 @@ +Review the documentation file in the planning folder called $ARGUMENTS and add questions, clarifications or feedback to a new section at the end, along with any opportunities to simplify \ No newline at end of file diff --git a/.github/workflows/claude-code-review.yml b/.github/workflows/claude-code-review.yml index b5e8cfd4d..37e66f3fd 100644 --- a/.github/workflows/claude-code-review.yml +++ b/.github/workflows/claude-code-review.yml @@ -38,7 +38,8 @@ jobs: 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 }}' + claude_args: '--allowedTools "mcp__github_inline_comment__create_inline_comment"' # 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 diff --git a/.github/workflows/claude.yml b/.github/workflows/claude.yml index d300267f1..6b15fac7a 100644 --- a/.github/workflows/claude.yml +++ b/.github/workflows/claude.yml @@ -46,5 +46,5 @@ jobs: # 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:*)' + # claude_args: '--allowed-tools Bash(gh pr *)' diff --git a/CLAUDE.md b/CLAUDE.md index 2bdd6fa10..afa4620df 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -2,6 +2,6 @@ All project documentation is in the `planning` directory. -The key document is PLAN.md included in full below; the market data component has been completed and is summarized in the file `planning/MARKET_DATA_SUMMARY.md` with more details in the `planning/archive` folder. Consult these docs only when required. The remainder of the platform is still to be developed. +The key document is PLAN.md included in full below; the market data component has been completed and is summarized in the file `planning/MARKET_DATA_SUMMARY.md`, with more details in `planning/MARKET_INTERFACE.md`, `planning/MARKET_SIMULATOR.md`, and `planning/MASSIVE_API.md`. Consult these docs only when required. The remainder of the platform is still to be developed. @planning/PLAN.md \ No newline at end of file diff --git a/JeremyTest.txt b/JeremyTest.txt new file mode 100644 index 000000000..8a2c9bb0f --- /dev/null +++ b/JeremyTest.txt @@ -0,0 +1,3 @@ +Hello World! + +Hello Universe! \ No newline at end of file diff --git a/planning/MARKET_INTERFACE.md b/planning/MARKET_INTERFACE.md new file mode 100644 index 000000000..907a03f15 --- /dev/null +++ b/planning/MARKET_INTERFACE.md @@ -0,0 +1,258 @@ +# Market Data Interface Design + +Unified Python interface for market data in FinAlly. Two implementations — the GBM +simulator and the Massive API poller — sit behind one abstract interface, so all +downstream code (SSE streaming, trade validation, the frontend) is agnostic to +where prices actually come from. + +**Status**: this describes the interface as actually implemented in +`backend/app/market/` (see `planning/MARKET_DATA_SUMMARY.md` for the build +summary). It supersedes the original pre-implementation sketch of this document; +a few details below (the data model's shape, how the cache computes `change`, +factory logging) changed slightly during implementation. For the Massive-specific +half of this, see `planning/MASSIVE_API.md`; for the simulator half, see +`planning/MARKET_SIMULATOR.md`. + +## Core Data Model + +`app/market/models.py`: + +```python +@dataclass(frozen=True, slots=True) +class PriceUpdate: + """Immutable snapshot of a single ticker's price at a point in time.""" + ticker: str + price: float + previous_price: float + timestamp: float = field(default_factory=time.time) # Unix seconds + + @property + def change(self) -> float: ... # price - previous_price, rounded to 4dp + @property + def change_percent(self) -> float: ... # % change, rounded to 4dp; 0.0 if previous_price == 0 + @property + def direction(self) -> str: ... # "up" | "down" | "flat" + + def to_dict(self) -> dict: ... # JSON-serializable form for SSE +``` + +`PriceUpdate` is the only structure that leaves the market data layer. `change`, +`change_percent`, and `direction` are computed **properties**, not stored fields — +they're derived once from `price`/`previous_price` at read time rather than +recomputed and frozen at write time. This is a deliberate simplification over the +original design sketch (which computed and stored `change`/`direction` inside +`PriceCache.update()`): keeping `PriceUpdate` a two-price, one-timestamp record +means there's a single source of truth for the derived fields no matter who +constructs the object. + +## Abstract Interface + +`app/market/interface.py`: + +```python +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. + """ + + @abstractmethod + async def start(self, tickers: list[str]) -> None: ... + @abstractmethod + async def stop(self) -> None: ... + @abstractmethod + async def add_ticker(self, ticker: str) -> None: ... + @abstractmethod + async def remove_ticker(self, ticker: str) -> None: ... + @abstractmethod + def get_tickers(self) -> list[str]: ... +``` + +Both `SimulatorDataSource` and `MassiveDataSource` implement this. Neither +`start()` nor the periodic update loop returns prices to the caller — they write +into a shared `PriceCache`, which is the only thing readers ever touch. `start()` +must be called exactly once (calling it twice is undefined); `stop()` is safe to +call multiple times, matching FastAPI's lifespan shutdown semantics where cleanup +code can run more than once in edge cases. + +## Price Cache + +`app/market/cache.py` — the single point of truth both sources write to and every +reader (SSE stream, portfolio valuation, trade execution) reads from: + +```python +class PriceCache: + """Thread-safe in-memory cache of the latest price for each ticker.""" + + def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: ... + def get(self, ticker: str) -> PriceUpdate | None: ... + def get_all(self) -> dict[str, PriceUpdate]: ... + def get_price(self, ticker: str) -> float | None: # convenience: just the float + def remove(self, ticker: str) -> None: ... + + @property + def version(self) -> int: ... # monotonic counter, bumped on every update() +``` + +Key implementation details: + +- **Thread-safe via a plain `threading.Lock`**, not an asyncio lock — both + `SimulatorDataSource` (asyncio task) and `MassiveDataSource` (asyncio task using + `asyncio.to_thread` for the blocking HTTP call) write from the event loop, so a + standard lock is sufficient; there's no true multi-threaded contention today, + but it costs nothing and protects against future multi-worker deployment. +- **Rounds to 2 decimal places on write** (`round(price, 2)`), so every reader — + SSE payloads, trade fills, portfolio valuation — sees already-display-clean + prices without each having to round independently. +- **`version` is the SSE change-detection mechanism**: bumped on every `update()` + call, so the stream endpoint (see below) can cheaply check "has anything changed + since I last sent a payload" without diffing the whole price dict. +- On the very first `update()` for a ticker, `previous_price` is seeded equal to + `price` (no prior value exists), so `direction` correctly reads `"flat"` and + `change` is `0.0` instead of raising or defaulting to some sentinel. + +## Factory Function + +`app/market/factory.py` — selects the implementation at startup: + +```python +def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: + """MASSIVE_API_KEY set and non-empty -> MassiveDataSource; otherwise -> SimulatorDataSource.""" + api_key = os.environ.get("MASSIVE_API_KEY", "").strip() + if api_key: + return MassiveDataSource(api_key=api_key, price_cache=price_cache) + else: + return SimulatorDataSource(price_cache=price_cache) +``` + +`.strip()` matters: an `.env` file with `MASSIVE_API_KEY=` (present but empty, or +whitespace) must fall through to the simulator rather than instantiating a +`MassiveDataSource` with a blank key that would fail every poll with `AuthError`. +The factory returns an **unstarted** source — the caller (FastAPI's lifespan +handler) still owns calling `await source.start(initial_tickers)`. + +## The Two Implementations + +### `MassiveDataSource` (`massive_client.py`) + +Polls `get_snapshot_all()` for the full watchlist on a timer (default 15s — see +`planning/MASSIVE_API.md` for rate-limit rationale). Notable behavior: + +- **Runs an immediate poll inside `start()`**, before the periodic task even + begins, so the cache has real data as soon as the app comes up instead of + waiting a full interval for the first prices to appear. +- **Offloads the client call with `asyncio.to_thread`** — the `massive` package's + `RESTClient` is synchronous, so calling it directly from `_poll_once()` would + block the whole event loop (including the SSE stream and every other request) + for the duration of the HTTP round trip. +- **Ticker add/remove are cheap, in-memory list edits** — `add_ticker()` just + appends to `self._tickers`; the new symbol shows up automatically on the next + scheduled poll rather than triggering an immediate extra API call. `remove_ticker()` + additionally calls `self._cache.remove(ticker)` so a delisted-from-watchlist + symbol disappears from the SSE stream immediately rather than lingering with a + stale last-known price until it ages out. +- **Poll failures are caught, logged, and swallowed** — a single bad poll (rate + limit, transient network error) doesn't crash the background task; the loop + simply tries again on the next interval. See `planning/MASSIVE_API.md`'s "Error + Handling" section for the gap this leaves (an invalid API key fails silently + forever rather than at startup). + +### `SimulatorDataSource` (`simulator.py`) + +Wraps a `GBMSimulator` (see `planning/MARKET_SIMULATOR.md`) in the same async-task +shape, at a 0.5s default interval instead of 15s: + +- **Seeds the cache synchronously inside `start()`** (and inside `add_ticker()`) + by reading `self._sim.get_price(ticker)` immediately after adding it — so a + newly-added ticker has a price in the cache before the next 0.5s tick, matching + the Massive side's "don't make the caller wait a full interval" behavior. +- **The per-tick loop wraps `self._sim.step()` in `try/except Exception`** and + logs on failure without stopping the loop — a single bad step (e.g. a NaN from + a pathological correlation matrix) shouldn't take down price streaming + entirely. + +## Integration with SSE + +`app/market/stream.py`'s `GET /api/stream/prices` endpoint reads `PriceCache` +directly — it has no knowledge of which `MarketDataSource` is active: + +```python +current_version = price_cache.version +if current_version != last_version: + last_version = current_version + prices = price_cache.get_all() + if prices: + yield f"data: {json.dumps({t: u.to_dict() for t, u in prices.items()})}\n\n" +await asyncio.sleep(interval) # 0.5s poll-the-cache loop, independent of source cadence +``` + +Using `version` instead of a naive "send every 0.5s regardless" means the stream +only emits a payload when something actually changed, and it degrades gracefully +across sources: the 15s-cadence Massive source and the 0.5s-cadence simulator both +"just work" against the same 0.5s cache-polling loop — the SSE endpoint simply +sends less often when the underlying source is slower. The endpoint also yields a +`retry: 1000\n\n` directive up front so `EventSource`'s built-in reconnect logic +retries quickly after a drop, and it exits its loop via `request.is_disconnected()` +so a closed browser tab doesn't leave an orphaned generator running forever. + +## Trade & Watchlist Validation + +Per PLAN.md §6 and §8, "is this ticker tradable" is answered by one check — +**does `PriceCache` have a current price for it** — never a separate hardcoded +symbol list: + +```python +if price_cache.get_price(ticker) is None: + raise ValidationError(f"{ticker} has no current price") +``` + +This is what makes the two sources' very different "unknown ticker" behaviors +converge on identical downstream behavior: the simulator invents a price for any +ticker (see `planning/MARKET_SIMULATOR.md`), so it always passes; Massive simply +never returns a snapshot for an invalid symbol, so it's absent from the cache and +naturally fails the same check with no special-case code required. + +## File Structure (as built) + +``` +backend/ + app/ + market/ + __init__.py # re-exports the public surface (see backend/CLAUDE.md) + models.py # PriceUpdate + interface.py # MarketDataSource ABC + cache.py # PriceCache + factory.py # create_market_data_source() + massive_client.py # MassiveDataSource + simulator.py # GBMSimulator + SimulatorDataSource + seed_prices.py # SEED_PRICES, TICKER_PARAMS, correlation constants + stream.py # create_stream_router() — SSE endpoint factory + tests/ + market/ # 73 tests, 84% coverage — see MARKET_DATA_SUMMARY.md +``` + +## Lifecycle + +1. **App startup** (FastAPI lifespan): create `PriceCache()`, call + `create_market_data_source(cache)`, then `await source.start(initial_tickers)`. +2. **Watchlist changes**: the watchlist API route calls + `await source.add_ticker(ticker)` / `await source.remove_ticker(ticker)` — the + route layer never touches `PriceCache` directly for adds, only the source does + (removal explicitly clears the cache entry too, per above). +3. **SSE streaming**: `GET /api/stream/prices` reads `PriceCache.get_all()` on its + own 0.5s cadence, independent of either source's update interval. +4. **Trade execution**: reads the current price via `PriceCache.get_price(ticker)` + at fill time — never caches or re-reads a stale price within a single request. +5. **App shutdown**: `await source.stop()`, cancelling the background task + cleanly. + +## Public Import Surface + +Per `backend/CLAUDE.md`, downstream code imports from the package root, not the +individual modules: + +```python +from app.market import PriceCache, PriceUpdate, MarketDataSource, create_market_data_source, create_stream_router +``` diff --git a/planning/MARKET_SIMULATOR.md b/planning/MARKET_SIMULATOR.md new file mode 100644 index 000000000..209fa23fe --- /dev/null +++ b/planning/MARKET_SIMULATOR.md @@ -0,0 +1,253 @@ +# Market Simulator Design + +Approach and code structure for simulating realistic stock prices when no +`MASSIVE_API_KEY` is configured. This is the default data source — most students +running FinAlly for the first time never touch the Massive API at all. + +**Status**: this describes the simulator as actually implemented in +`backend/app/market/simulator.py` and `backend/app/market/seed_prices.py`, +covered by 17 tests in `backend/tests/market/test_simulator.py` plus 10 +integration tests in `test_simulator_source.py` (98% line coverage on +`simulator.py`). It supersedes the original pre-implementation sketch of this +document; the math and structure below match what shipped almost exactly, with +the differences noted inline. For where this plugs into the rest of the market +data layer, see `planning/MARKET_INTERFACE.md`. + +## Overview + +The simulator uses **Geometric Brownian Motion (GBM)** — the standard model +underlying Black-Scholes option pricing — to generate price paths that evolve +continuously with random noise, can never go negative, and produce the lognormal +return distribution seen in real markets. `SimulatorDataSource` steps the +simulation every 500ms via an asyncio background task, producing a continuous +stream of small, plausible-looking price changes. + +## GBM Math + +At each time step: + +``` +S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) +``` + +- `S(t)` — current price +- `mu` — annualized drift (expected return), e.g. `0.05` for 5%/year +- `sigma` — annualized volatility, e.g. `0.20` for 20%/year +- `dt` — this time step, expressed as a fraction of a trading year +- `Z` — a (correlated) standard normal draw + +`dt` is derived from real trading-calendar constants rather than a round number, +so the annualized `mu`/`sigma` parameters translate to realistic per-tick moves: + +```python +TRADING_SECONDS_PER_YEAR = 252 * 6.5 * 3600 # 5,896,800 (252 trading days * 6.5h * 3600s) +DEFAULT_DT = 0.5 / TRADING_SECONDS_PER_YEAR # ~8.48e-8, for 500ms ticks +``` + +This tiny `dt` produces sub-cent moves per individual tick, which accumulate +naturally into realistic-looking intraday ranges over the course of a session — +e.g. TSLA's `sigma=0.50` produces roughly the right order of magnitude of +intraday range over a full simulated trading day. + +## Correlated Moves + +Real stocks don't move independently — tech names tend to move together on +market-wide news, etc. The simulator generates correlated random draws via a +**Cholesky decomposition** of a correlation matrix: given correlation matrix `C`, +compute `L = cholesky(C)`, then for independent standard normals `z_independent`, +`z_correlated = L @ z_independent` has the desired covariance structure. + +Correlation groups and coefficients (`seed_prices.py`): + +```python +CORRELATION_GROUPS = { + "tech": {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"}, + "finance": {"JPM", "V"}, +} +INTRA_TECH_CORR = 0.6 # tech stocks move together +INTRA_FINANCE_CORR = 0.5 # finance stocks move together +CROSS_GROUP_CORR = 0.3 # different sectors, or an unrecognized ticker +TSLA_CORR = 0.3 # TSLA does its own thing, even though it's in the tech set +``` + +The pairwise lookup (`GBMSimulator._pairwise_correlation`) checks TSLA **first**, +before the sector-membership checks — TSLA is a member of the `"tech"` set (so +that unknown-ticker fallback logic elsewhere doesn't need a special case for it), +but it's deliberately given the flat `0.3` cross-group correlation with +*everything*, including other tech names, rather than the `0.6` intra-tech rate. +A ticker outside both named groups (any dynamically-added symbol) also falls +through to `0.3` — one constant serves double duty as both "cross-sector" and +"unknown ticker" correlation, since both cases mean "no special relationship +assumed." + +The Cholesky matrix is rebuilt (`_rebuild_cholesky()`) on every `add_ticker()`/ +`remove_ticker()` call — O(n²) matrix construction plus an O(n³) decomposition, +but `n` stays well under 50 tickers in practice, so this is not a performance +concern. With 0 or 1 tickers, `_cholesky` is left `None` and `step()` uses the +independent draws directly (a 1x1 correlation matrix is trivially just itself, +so skipping the decomposition in that case is a harmless shortcut, not an +approximation). + +## Random Events + +Each tick, each ticker independently has a small chance of a sudden 2-5% shock, +for visual drama on the dashboard: + +```python +if random.random() < event_probability: # default 0.001 (0.1%) + shock_magnitude = random.uniform(0.02, 0.05) + shock_sign = random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock_magnitude * shock_sign +``` + +At 2 ticks/second, 0.1% per tick per ticker works out to roughly one event per +ticker every ~500 seconds; across a 10-ticker default watchlist, expect a +noticeable jump somewhere on the board roughly every ~50 seconds — frequent +enough to keep a live demo visually interesting without every ticker looking +like it's constantly spiking. + +## Seed Prices & Per-Ticker Parameters + +`seed_prices.py` holds realistic starting prices and volatility/drift parameters +for the ten default watchlist tickers: + +```python +SEED_PRICES = { + "AAPL": 190.00, "GOOGL": 175.00, "MSFT": 420.00, "AMZN": 185.00, "TSLA": 250.00, + "NVDA": 800.00, "META": 500.00, "JPM": 195.00, "V": 280.00, "NFLX": 600.00, +} + +TICKER_PARAMS = { + "AAPL": {"sigma": 0.22, "mu": 0.05}, + "GOOGL": {"sigma": 0.25, "mu": 0.05}, + "MSFT": {"sigma": 0.20, "mu": 0.05}, + "AMZN": {"sigma": 0.28, "mu": 0.05}, + "TSLA": {"sigma": 0.50, "mu": 0.03}, # high volatility + "NVDA": {"sigma": 0.40, "mu": 0.08}, # high volatility, strong drift + "META": {"sigma": 0.30, "mu": 0.05}, + "JPM": {"sigma": 0.18, "mu": 0.04}, # low volatility (bank) + "V": {"sigma": 0.17, "mu": 0.04}, # low volatility (payments) + "NFLX": {"sigma": 0.35, "mu": 0.05}, +} + +DEFAULT_PARAMS = {"sigma": 0.25, "mu": 0.05} # for any ticker not in the table above +``` + +## Dynamically Added Tickers + +Per PLAN.md §6, a ticker the user (or the AI chat) adds beyond the ten defaults +still needs to "just work" in simulator mode — there's no such thing as an +invalid symbol to the simulator: + +```python +def _add_ticker_internal(self, ticker: str) -> None: + if ticker in self._prices: + return + self._tickers.append(ticker) + self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50.0, 300.0)) + self._params[ticker] = TICKER_PARAMS.get(ticker, dict(DEFAULT_PARAMS)) +``` + +An unrecognized ticker gets a random starting price in `$50–$300` and the +`DEFAULT_PARAMS` volatility/drift (`sigma=0.25, mu=0.05` — a middling profile, +neither as sleepy as JPM/V nor as wild as TSLA/NVDA), and its correlation with +every other ticker falls back to `CROSS_GROUP_CORR` (`0.3`) since it's in neither +named sector set. It has a live, moving price from the very next tick — there is +no "unknown ticker" error path in the simulator at all, by design. + +## Implementation Structure + +`GBMSimulator` is the pure simulation engine — no asyncio, no cache access, just +ticker state and a `step()` method: + +```python +class GBMSimulator: + def step(self) -> dict[str, float]: + """Advance all tickers by one time step. Returns {ticker: new_price}.""" + n = len(self._tickers) + if n == 0: + return {} + z_independent = np.random.standard_normal(n) + z_correlated = self._cholesky @ z_independent if self._cholesky is not None else z_independent + result = {} + for i, ticker in enumerate(self._tickers): + mu, sigma = self._params[ticker]["mu"], self._params[ticker]["sigma"] + drift = (mu - 0.5 * sigma**2) * self._dt + diffusion = sigma * math.sqrt(self._dt) * z_correlated[i] + self._prices[ticker] *= math.exp(drift + diffusion) + if random.random() < self._event_prob: + shock = random.uniform(0.02, 0.05) * random.choice([-1, 1]) + self._prices[ticker] *= 1 + shock + result[ticker] = round(self._prices[ticker], 2) + return result + + def add_ticker(self, ticker: str) -> None: ... # adds + rebuilds Cholesky + def remove_ticker(self, ticker: str) -> None: ... # removes + rebuilds Cholesky + def get_price(self, ticker: str) -> float | None: ... + def get_tickers(self) -> list[str]: ... +``` + +`SimulatorDataSource` is the thin `MarketDataSource` adapter around it — it owns +the asyncio task, the `PriceCache` writes, and the 500ms sleep loop, but contains +no GBM math itself: + +```python +class SimulatorDataSource(MarketDataSource): + async def start(self, tickers: list[str]) -> None: + self._sim = GBMSimulator(tickers=tickers, event_probability=self._event_prob) + for ticker in tickers: # seed the cache immediately + price = self._sim.get_price(ticker) + if price is not None: + self._cache.update(ticker=ticker, price=price) + self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") + + async def _run_loop(self) -> None: + while True: + try: + if self._sim: + for ticker, price in self._sim.step().items(): + self._cache.update(ticker=ticker, price=price) + except Exception: + logger.exception("Simulator step failed") # one bad tick doesn't kill the loop + await asyncio.sleep(self._interval) +``` + +Keeping the pure-math `GBMSimulator` separate from the asyncio/cache-wiring +`SimulatorDataSource` is what makes the 17 unit tests in +`test_simulator.py` possible without touching asyncio or a real `PriceCache` at +all — they construct a `GBMSimulator` directly and assert on `step()`'s output +(price bounds, correlation behavior, seeding, dynamic add/remove). The 10 tests +in `test_simulator_source.py` then cover the async wiring layer separately. + +## File Structure (as built) + +``` +backend/ + app/ + market/ + simulator.py # GBMSimulator + SimulatorDataSource + seed_prices.py # SEED_PRICES, TICKER_PARAMS, DEFAULT_PARAMS, correlation constants + tests/ + market/ + test_simulator.py # GBMSimulator unit tests (17 tests, 98% coverage) + test_simulator_source.py # SimulatorDataSource integration tests (10 tests) +``` + +## Behavior Notes + +- Prices can never go negative — GBM is multiplicative (`price *= exp(...)`), and + `exp()` is always positive regardless of the random draw. +- The correlation matrix must be positive semi-definite for `np.linalg.cholesky` + to succeed; every coefficient used here (`0.3`, `0.5`, `0.6`) is a fixed + constant well inside valid correlation range, so this can't fail at runtime — + it would only become a risk if correlations were ever made ticker-pair-specific + and set inconsistently (e.g. `corr(A,B) = 0.9` and `corr(A,C) = corr(B,C) = -0.9` + can produce a non-PSD matrix). The current scheme, which derives every pairwise + value from a small set of shared group constants, avoids that entirely. +- `step()` is the hot path, called twice a second per running instance — it's + kept allocation-light (no per-call list comprehensions beyond the final result + dict) and uses NumPy's vectorized `standard_normal(n)` for the independent + draws rather than drawing `n` individual `random.gauss()` calls. +- A live terminal demo of the simulator (Rich-based dashboard with sparklines, + color-coded direction arrows, and an event log) is available at + `backend/market_data_demo.py` — see `planning/MARKET_DATA_SUMMARY.md`. diff --git a/planning/MASSIVE_API.md b/planning/MASSIVE_API.md new file mode 100644 index 000000000..03cc391ae --- /dev/null +++ b/planning/MASSIVE_API.md @@ -0,0 +1,283 @@ +# Massive API Reference (formerly Polygon.io) + +Verified reference documentation for the Massive REST API and its official Python +client, as used by FinAlly's `MassiveDataSource` (`backend/app/market/massive_client.py`). + +Researched directly against the vendor's current docs and the `massive-com/client-python` +source on 2026-09-04 (sources listed at the bottom). This supersedes the original +pre-verification draft of this document, which got several Python attribute names +wrong (see "Corrections vs. the Earlier Draft" below). + +## Overview + +- **Company**: Polygon.io rebranded to **Massive** on **October 30, 2025**. Existing + API keys, accounts, and billing carried over unchanged. +- **Base URL**: `https://api.massive.com` (new default). The legacy + `https://api.polygon.io` host still works and will continue to for an extended + period, so old integrations aren't broken by the rename. +- **Python package**: `massive` on PyPI — install with `uv add massive` or + `pip install -U massive`. It is a fork/continuation of the old `polygon-api-client` + package, republished under the new name with the new default base URL. +- **Min Python version**: 3.9+ +- **Repo**: [github.com/massive-com/client-python](https://github.com/massive-com/client-python) + +## Authentication + +```python +from massive import RESTClient + +# No-arg form: reads the MASSIVE_API_KEY environment variable automatically. +client = RESTClient() + +# Or pass explicitly: +client = RESTClient(api_key="your_key_here") +``` + +`MASSIVE_API_KEY` is the client library's own default environment variable name — +it isn't something FinAlly invented to match; the project's `.env` variable and the +client's built-in default happen to line up, so `RESTClient()` with no arguments +just works once the process environment has `MASSIVE_API_KEY` set. + +Internally, every request carries `Authorization: Bearer `; the client sets +this header for you. + +## Rate Limits + +| Tier | Limit | +|------|-------| +| Free | 5 requests/minute | +| Paid (all tiers) | No hard cap; vendor asks that you stay under ~100 req/s so you don't degrade service for others | + +FinAlly polls on a timer rather than opening a persistent connection (see "Why +polling, not WebSockets" below). `MassiveDataSource` defaults to a 15s interval, +which keeps a single-user instance comfortably under the free-tier 5/min limit +even with retries. Paid tiers can safely poll every 2-5s. + +## Client Initialization Details + +- `RESTClient(api_key=..., connect_timeout=..., read_timeout=..., retries=...)` — + `retries` and timeouts are optional overrides; the client has built-in retry with + exponential backoff (backoff factor 0.1s) on `413, 429, 499, 500, 502, 503, 504`. +- `RESTClient(pagination=False)` disables automatic multi-page fetching for + paginated list methods (`list_aggs`, `list_trades`, `list_quotes`, etc.); with + pagination on (the default), `limit` controls *page size*, and the client + transparently walks all pages for you. +- `RESTClient(trace=True, verbose=True)` prints request/response details — useful + for debugging response shapes during development. + +## Endpoint FinAlly Actually Uses + +### Snapshot — All Tickers (the only endpoint the poller calls) + +This is the sole call `MassiveDataSource._fetch_snapshots()` makes. It returns +current prices for an arbitrary list of tickers in **one HTTP round trip**, which +is what makes REST polling viable within the free-tier rate limit. + +**REST**: `GET /v2/snapshot/locale/us/markets/stocks/tickers?tickers=AAPL,GOOGL,MSFT` + +**Python method** (`SnapshotClient.get_snapshot_all`): +```python +def get_snapshot_all( + self, + market_type: str | SnapshotMarketType, + tickers: str | list[str] | None = None, + include_otc: bool | None = False, + params: dict | None = None, + raw: bool = False, + options: RequestOptionBuilder | None = None, +) -> list[TickerSnapshot] | HTTPResponse +``` + +```python +from massive import RESTClient +from massive.rest.models import SnapshotMarketType + +client = RESTClient() + +snapshots = client.get_snapshot_all( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"], +) + +for snap in snapshots: + print(f"{snap.ticker}: ${snap.last_trade.price}") +``` + +**`TickerSnapshot` fields** (Python attribute names on the typed model — see +"Raw JSON vs. Python attributes" below for how these map to the wire format): + +| Attribute | Type | Meaning | +|---|---|---| +| `ticker` | `str` | Symbol | +| `day` | `Agg` | Current session's running bar (open/high/low/close/volume/vwap) | +| `prev_day` | `Agg` | Previous session's full bar — this is where "previous close" actually lives (`prev_day.close`), **not** `day.previous_close` | +| `min` | `MinuteSnapshot` | Most recently completed minute bar | +| `last_trade` | `LastTrade` | Most recent trade: `price`, `size`, `exchange`, `sip_timestamp` (Unix **nanoseconds**), etc. | +| `last_quote` | `LastQuote` | Most recent NBBO quote: `bid_price`, `ask_price`, `bid_size`, `ask_size`, `sip_timestamp`, etc. | +| `todays_change` | `float` | Absolute change since previous close | +| `todays_change_percent` | `float` | Percent change since previous close | +| `updated` | `int` | Server-side update timestamp | +| `fair_market_value` | `float \| None` | Business-plan-only field; `None` otherwise | + +`massive_client.py` only reads `snap.ticker`, `snap.last_trade.price`, and +`snap.last_trade.timestamp` — everything else above is available but unused today. +`todays_change_percent` is a ready-made source for a "daily change %" column if the +watchlist UI wants one without computing it client-side. + +⚠️ **Timestamp units differ by object.** `last_trade.sip_timestamp` / +`last_trade.timestamp` (the attribute the client normalizes it to) is Unix +**milliseconds**, matching what `massive_client.py` assumes (`timestamp / 1000.0`). +Some other Massive timestamp fields are nanoseconds — always check the specific +model before assuming a unit. + +## Endpoints Available But Not Used Today + +Documented here because they're natural next steps (e.g., for historical chart +backfill) and because the earlier draft of this document described some of them +with the wrong field names. + +### Previous Close + +**REST**: `GET /v2/aggs/ticker/{ticker}/prev` + +```python +prev = client.get_previous_close_agg(ticker="AAPL") # -> PreviousCloseAgg +print(prev.close, prev.open, prev.high, prev.low, prev.volume, prev.timestamp) +``` + +`PreviousCloseAgg` fields: `ticker`, `open`, `high`, `low`, `close`, `volume`, +`vwap`, `timestamp`. Redundant with `TickerSnapshot.prev_day` if you're already +calling `get_snapshot_all`; useful standalone if you only need yesterday's close +without pulling a full snapshot. + +### Aggregates / Bars (for historical chart data) + +**REST**: `GET /v2/aggs/ticker/{ticker}/range/{multiplier}/{timespan}/{from}/{to}` + +```python +for bar in client.list_aggs( + ticker="AAPL", + multiplier=1, + timespan="day", + from_="2026-08-01", + to="2026-09-01", + adjusted=True, + sort="asc", + limit=50000, +): + print(bar.timestamp, bar.open, bar.high, bar.low, bar.close, bar.volume) +``` + +`list_aggs` auto-paginates (returns an `Iterator[Agg]`); `get_aggs` is the +non-paginating sibling that returns a plain `list[Agg]`. `Agg` fields: `open`, +`high`, `low`, `close`, `volume`, `vwap`, `timestamp`, `transactions`, `otc`. +This is the endpoint to reach for if the main chart ever needs to seed itself with +real historical bars instead of only the live SSE stream accumulated since page load. + +### Last Trade / Last Quote (single ticker, no snapshot) + +```python +trade = client.get_last_trade(ticker="AAPL") # -> LastTrade +quote = client.get_last_quote(ticker="AAPL") # -> LastQuote +``` + +Only useful if you want just one ticker's trade or quote without the rest of a +snapshot; for FinAlly's "poll the whole watchlist" pattern, `get_snapshot_all` is +strictly better since it's one call instead of N. + +### Universal Snapshot (newer, cross-asset-class endpoint) + +`list_universal_snapshots(type=SnapshotMarketType.STOCKS, ticker_any_of=[...])` +is a newer, more general snapshot endpoint that also covers options/indices/forex/crypto +in one shape. It caps at **250 symbols per request** (vs. no documented cap on +`get_snapshot_all`'s `tickers` param). `get_snapshot_all` is not deprecated and +remains the simpler choice for FinAlly's all-stocks use case; `list_universal_snapshots` +would only matter if the watchlist needed to mix asset classes. + +## Raw JSON vs. Python Attributes + +The vendor's raw JSON uses `camelCase` (`lastTrade`, `prevDay`, `lastQuote`, +`todaysChangePerc`), but the official Python client deserializes responses into +typed dataclasses with `snake_case` attributes (`last_trade`, `prev_day`, +`last_quote`, `todays_change_percent`). **Code should always use the Python +attribute names**, not the raw JSON field names — mixing the two up is the most +common way to introduce a silent `AttributeError` or, worse, a `None` read that +looks like a valid price. Pass `raw=True` to any client method to get the +untouched JSON/`HTTPResponse` instead, if you ever need to inspect the wire format +directly. + +## WebSocket Client (available, not used by FinAlly) + +```python +from massive import WebSocketClient + +ws = WebSocketClient(api_key="...", subscriptions=["T.AAPL"]) # T. = trades +ws.run(handle_msg=lambda msgs: print(msgs)) +``` + +FinAlly deliberately polls REST instead of using this (see PLAN.md §6): a +WebSocket connection is stateful, requires reconnect/backoff logic, and needs a +paid plan for real-time trade/quote streams (delayed data is more limited on +WebSocket than on the free-tier snapshot REST endpoint). Polling is simpler, +works identically across all plan tiers, and the snapshot endpoint's "all tickers +in one call" shape maps cleanly onto FinAlly's shared `PriceCache` model. + +## Error Handling + +The client raises typed exceptions from `massive.exceptions`: + +- **`AuthError`** — empty or invalid API key (HTTP 401/403 conditions) +- **`BadResponse`** — any non-2xx response the retry policy didn't recover from + +In practice, expect: +- **401** — invalid API key +- **403** — plan doesn't include the endpoint/data you requested +- **429** — rate limit exceeded (free tier: 5 req/min) — the client retries these + automatically per the backoff policy above before raising +- **5xx** — server errors — also auto-retried + +`massive_client.py`'s `_poll_once()` wraps the whole poll in a broad +`try/except Exception`, logs, and lets the next scheduled poll retry — it doesn't +distinguish `AuthError` from `BadResponse` today. A bad API key currently just +produces a silent, permanently-empty price cache with an error logged every +15 seconds rather than a startup-time failure; if that's ever worth surfacing to +the user, catching `AuthError` specifically at startup (during the immediate +first poll in `start()`) and failing fast is the natural place to do it. + +## Notes + +- The snapshot endpoint returns data for **all requested tickers in one call** — + critical for staying within the free-tier rate limit. +- During market-closed hours, `last_trade.price` reflects the last traded price + (may include after-hours activity). +- The `day` bar resets at market open; during pre-market it may still reflect the + previous session. Use `prev_day` when you specifically want "yesterday's close," + never `day`. +- FinAlly's dynamic-ticker behavior (PLAN.md §6) falls directly out of this + endpoint's shape: an unrecognized/invalid symbol simply doesn't appear in the + `tickers` list of the response — there's no explicit "invalid ticker" error, so + "does the price cache have this ticker" is the only reliable validity check. + +## Corrections vs. the Earlier Draft + +The original draft of this document (written before this verification pass) had +two inaccuracies worth flagging so they aren't propagated: + +1. It read previous close as `snap.day.previous_close` — that field doesn't + exist. Previous close is `snap.prev_day.close`. +2. It read day change as `snap.day.change_percent` — that field lives at the + top level as `snap.todays_change_percent`, not nested under `day`. + +Neither bug was ever reachable in production: `massive_client.py` only ever +reads `last_trade.price` and `last_trade.timestamp`, so the wrong field names in +the earlier draft were never actually executed. + +## Sources + +- [massive-com/client-python README](https://github.com/massive-com/client-python/blob/master/README.md) +- [massive-com/client-python repository](https://github.com/massive-com/client-python) (`massive/rest/snapshot.py`, `massive/rest/aggs.py`, `massive/rest/trades.py`, `massive/rest/quotes.py`, `massive/rest/models/snapshot.py`, `massive/rest/models/aggs.py`, `massive/rest/models/trades.py`, `massive/rest/models/quotes.py`, `massive/rest/base.py`, `massive/exceptions.py`) +- [Polygon.io is Now Massive](https://massive.com/blog/polygon-is-now-massive) +- [Full Market Snapshot | Stocks REST API](https://massive.com/docs/rest/stocks/snapshots/full-market-snapshot) +- [What is the max number of tickers I can pass through Massive's Snapshot?](https://massive.com/knowledge-base/article/what-is-the-max-number-of-tickers-i-can-pass-through-massives-snapshot) +- [What is the request limit for Massive's RESTful APIs?](https://massive.com/knowledge-base/article/what-is-the-request-limit-for-massives-restful-apis) +- [Custom Bars | Stocks REST API](https://massive.com/docs/rest/stocks/aggregates/custom-bars) diff --git a/planning/PLAN.md b/planning/PLAN.md index bc1811b33..1824eef97 100644 --- a/planning/PLAN.md +++ b/planning/PLAN.md @@ -101,7 +101,7 @@ finally/ ├── db/ # Volume mount target (SQLite file lives here at runtime) │ └── .gitkeep # Directory exists in repo; finally.db is gitignored ├── Dockerfile # Multi-stage build (Node → Python) -├── docker-compose.yml # Optional convenience wrapper +├── docker-compose.yml # Canonical run config (volume, port, env file); start/stop scripts wrap this ├── .env # Environment variables (gitignored, .env.example committed) └── .gitignore ``` @@ -156,6 +156,15 @@ Both the simulator and the Massive client implement the same abstract interface. - Starts from realistic seed prices (e.g., AAPL ~$190, GOOGL ~$175, etc.) - Runs as an in-process background task — no external dependencies +### Dynamically Added / Unlisted Tickers + +Tickers can be added to the watchlist beyond the 10 pre-seeded defaults (manually or via the AI chat). Behavior differs by source, but the price cache is always the single source of truth for "is this ticker tradable": + +- **Simulator**: an unrecognized ticker gets a random starting price ($50–$300) and default GBM parameters (moderate drift/volatility, 0.3 correlation with everything else), so it always ends up with a live price immediately. +- **Massive**: an invalid/unrecognized symbol simply never appears in the poll response, so it never gets a price in the cache — there's no explicit "invalid ticker" error, just an absent price. + +Because both cases converge on the same signal, trade and watchlist validation should check **"does the price cache have a current price for this ticker?"** rather than maintaining a separate list of valid symbols. + ### Massive API (Optional) - REST API polling (not WebSocket) — simpler, works on all tiers @@ -244,6 +253,19 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod - One user profile: `id="default"`, `cash_balance=10000.0` - Ten watchlist entries: AAPL, GOOGL, MSFT, AMZN, TSLA, NVDA, META, JPM, V, NFLX +### Cost Basis & P&L + +- `positions.avg_cost` is a **weighted average cost basis**. A buy recomputes it as `(old_qty * old_avg_cost + new_qty * fill_price) / (old_qty + new_qty)`. A sell reduces `quantity` but leaves `avg_cost` unchanged; if `quantity` reaches zero, the position row is deleted. +- Only **unrealized** P&L is tracked/displayed (current price vs. `avg_cost` on open positions, per §10's positions table). Realized P&L (gains/losses actually locked in by sells) is out of scope for this build — the `trades` log is sufficient to reconstruct it later if needed. + +### Money as Floating Point + +`cash_balance`, `price`, `avg_cost`, and `total_value` are all SQLite `REAL` (floating point), not fixed-point/integer cents. This is an accepted tradeoff for a simulated-money demo app — simplicity over cent-level precision. Round to 2 decimal places for display; don't introduce integer-cents storage unless real-money accuracy becomes a requirement. + +### Multi-User Scaffolding + +The `user_id` column on every table (hardcoded to `"default"`) is a deliberate forward-compat hook, kept intentionally even though multi-user support isn't being built now — the cost of carrying it is low and it avoids a schema migration later. + --- ## 8. API Endpoints @@ -257,25 +279,26 @@ All tables include a `user_id` column defaulting to `"default"`. This is hardcod | Method | Path | Description | |--------|------|-------------| | GET | `/api/portfolio` | Current positions, cash balance, total value, unrealized P&L | -| POST | `/api/portfolio/trade` | Execute a trade: `{ticker, quantity, side}` | +| POST | `/api/portfolio/trade` | Execute a trade: `{ticker, quantity, side}`. Rejected with a validation error if the ticker has no current price in the price cache (i.e., it isn't tracked — see §6, "Dynamically Added / Unlisted Tickers") | | GET | `/api/portfolio/history` | Portfolio value snapshots over time (for P&L chart) | ### Watchlist | Method | Path | Description | |--------|------|-------------| -| GET | `/api/watchlist` | Current watchlist tickers with latest prices | -| POST | `/api/watchlist` | Add a ticker: `{ticker}` | -| DELETE | `/api/watchlist/{ticker}` | Remove a ticker | +| GET | `/api/watchlist` | Current watchlist tickers (ticker + added_at). Does **not** include prices — live prices come exclusively from the SSE stream once connected, so there's one source of price truth, not two | +| POST | `/api/watchlist` | Add a ticker: `{ticker}`. Adding a ticker already on the watchlist is a no-op (still returns success) | +| DELETE | `/api/watchlist/{ticker}` | Remove a ticker. Removing a ticker not on the watchlist is a no-op (still returns success) | ### Chat | Method | Path | Description | |--------|------|-------------| | POST | `/api/chat` | Send a message, receive complete JSON response (message + executed actions) | +| GET | `/api/chat/history` | Load persisted conversation history, so the chat panel can be restored on page reload instead of starting empty | ### System | Method | Path | Description | |--------|------|-------------| -| GET | `/api/health` | Health check (for Docker/deployment) | +| GET | `/api/health` | Liveness check only (process is up) — for Docker/deployment; does not probe the database or market data source | --- @@ -290,7 +313,7 @@ There is an OPENROUTER_API_KEY in the .env file in the project root. When the user sends a chat message, the backend: 1. Loads the user's current portfolio context (cash, positions with P&L, watchlist with live prices, total portfolio value) -2. Loads recent conversation history from the `chat_messages` table +2. Loads recent conversation history from the `chat_messages` table — the most recent 20 messages (10 user/assistant exchanges), to bound prompt size over a long session 3. Constructs a prompt with a system message, portfolio context, conversation history, and the user's new message 4. Calls the LLM via LiteLLM → OpenRouter, requesting structured output, using the cerebras-inference skill 5. Parses the complete structured JSON response @@ -315,8 +338,8 @@ The LLM is instructed to respond with JSON matching this schema: ``` - `message` (required): The conversational text shown to the user -- `trades` (optional): Array of trades to auto-execute. Each trade goes through the same validation as manual trades (sufficient cash for buys, sufficient shares for sells) -- `watchlist_changes` (optional): Array of watchlist modifications +- `trades` (optional): Array of trades to auto-execute. Each trade goes through the same validation as manual trades (sufficient cash for buys, sufficient shares for sells, and the ticker must have a current price in the price cache — see §8) +- `watchlist_changes` (optional): Array of watchlist modifications. `action` is `"add"` or `"remove"`; both are idempotent (adding an already-watched ticker or removing one that isn't watched is a no-op, not an error) ### Auto-Execution @@ -352,19 +375,21 @@ When `LLM_MOCK=true`, the backend returns deterministic mock responses instead o The frontend is a single-page application with a dense, terminal-inspired layout. The specific component architecture and layout system is up to the Frontend Engineer, but the UI should include these elements: -- **Watchlist panel** — grid/table of watched tickers with: ticker symbol, current price (flashing green/red on change), daily change %, and a sparkline mini-chart (accumulated from SSE since page load) +- **Watchlist panel** — grid/table of watched tickers with: ticker symbol, current price (flashing green/red on change), daily change %, and a sparkline mini-chart (accumulated from SSE since page load). The ticker list itself comes from `GET /api/watchlist`; prices populate progressively as SSE events arrive after connect - **Main chart area** — larger chart for the currently selected ticker, with at minimum price over time. Clicking a ticker in the watchlist selects it here. - **Portfolio heatmap** — treemap visualization where each rectangle is a position, sized by portfolio weight, colored by P&L (green = profit, red = loss) - **P&L chart** — line chart showing total portfolio value over time, using data from `portfolio_snapshots` - **Positions table** — tabular view of all positions: ticker, quantity, avg cost, current price, unrealized P&L, % change - **Trade bar** — simple input area: ticker field, quantity field, buy button, sell button. Market orders, instant fill. -- **AI chat panel** — docked/collapsible sidebar. Message input, scrolling conversation history, loading indicator while waiting for LLM response. Trade executions and watchlist changes shown inline as confirmations. +- **AI chat panel** — docked/collapsible sidebar. On load, hydrates conversation history from `GET /api/chat/history` so a page refresh doesn't lose it. Message input, scrolling conversation history, loading indicator while waiting for LLM response. Trade executions and watchlist changes shown inline as confirmations. - **Header** — portfolio total value (updating live), connection status indicator, cash balance ### Technical Notes - Use `EventSource` for SSE connection to `/api/stream/prices` -- Canvas-based charting library preferred (Lightweight Charts or Recharts) for performance +- **Charting libraries, split by responsibility** (not left as an open "either/or"): + - **Lightweight Charts** (canvas-based) for all time-series: watchlist sparklines, the main detail chart, and the P&L chart. It's purpose-built for exactly this and performs well under frequent SSE-driven updates. + - **Recharts** (SVG-based) for the portfolio heatmap only, using its built-in `Treemap` component — Lightweight Charts has no treemap support, so this isn't a case of arbitrarily using two libraries for the same job. - Price flash effect: on receiving a new price, briefly apply a CSS class with background color transition, then remove it - All API calls go to the same origin (`/api/*`) — no CORS configuration needed - Tailwind CSS for styling with a custom dark theme @@ -393,25 +418,25 @@ FastAPI serves the static frontend files and all API routes on port 8000. ### Docker Volume -The SQLite database persists via a named Docker volume: +The SQLite database persists via a **bind mount** of the project's `db/` directory (not a named volume) — this matches §4, where `db/` is described as the host-visible runtime mount target, and keeps `finally.db` directly inspectable on the host: ```bash -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 ``` The `db/` directory in the project root maps to `/app/db` in the container. The backend writes `finally.db` to this path. ### Start/Stop Scripts +`docker-compose.yml` is the canonical definition of the volume mount, port mapping, and env file — not a second, separately-maintained copy of those flags. The start/stop scripts are thin wrappers around `docker compose` so there's exactly one place the run configuration lives: + **`scripts/start_mac.sh`** (macOS/Linux): -- Builds the Docker image if not already built (or if `--build` flag passed) -- Runs the container with the volume mount, port mapping, and `.env` file +- Runs `docker compose up -d --build` (Compose handles rebuilding the image if source changed) - Prints the URL to access the app - Optionally opens the browser **`scripts/stop_mac.sh`** (macOS/Linux): -- Stops and removes the running container -- Does NOT remove the volume (data persists) +- Runs `docker compose down` (does NOT remove the volume — data persists) **`scripts/start_windows.ps1`** / **`scripts/stop_windows.ps1`**: PowerShell equivalents for Windows. @@ -454,3 +479,26 @@ The container is designed to deploy to AWS App Runner, Render, or any container - Portfolio visualization: heatmap renders with correct colors, P&L chart has data points - AI chat (mocked): send a message, receive a response, trade execution appears inline - SSE resilience: disconnect and verify reconnection + +--- + +## 13. Doc Review — Resolved + +A prior review pass raised 10 questions and 4 simplification opportunities against this document. All are now resolved and incorporated into the sections above: + +| # | Item | Resolution | +|---|------|------------| +| 1 | Volume mount contradiction (§4 vs §11) | Standardized on a bind mount; §11 updated | +| 2 | Trading tickers outside the watchlist | §8: trade requires a current price in the cache | +| 3 | Unlisted/arbitrary tickers in simulator mode | §6: documented actual fallback behavior (already implemented) | +| 4 | Unbounded chat history in prompts | §9: capped at the most recent 20 messages | +| 5 | No read endpoint for chat history | §8: added `GET /api/chat/history` | +| 6 | Average cost / realized P&L method | §7: weighted-average cost basis; realized P&L out of scope | +| 7 | Money as floating point | §7: accepted tradeoff, documented explicitly | +| 8 | `watchlist_changes.action` values and edge cases | §9: `"add"`/`"remove"`, both idempotent | +| 9 | Recharts mislabeled as canvas-based | §10: split responsibility — Lightweight Charts for time-series, Recharts for the treemap only | +| 10 | Scope of `/api/health` | §8: liveness-only, documented | +| 11 (simplification) | One charting library vs. "preferred" | §10: resolved via #9's split | +| 12 (simplification) | `GET /api/watchlist` redundant with SSE | §8: watchlist endpoint no longer returns prices | +| 13 (simplification) | Two parallel launch paths | §11: scripts now wrap `docker compose` | +| 14 (simplification) | `user_id` scaffolding on every table | Confirmed intentional — kept as-is (project owner's call) | diff --git a/planning/REVIEW.md b/planning/REVIEW.md new file mode 100644 index 000000000..90b50566f --- /dev/null +++ b/planning/REVIEW.md @@ -0,0 +1,30 @@ +# PLAN.md Review + +This is a follow-up review pass against `planning/PLAN.md`, done after the "Doc Review — Resolved" pass already recorded in §13. It cross-checks the plan against itself, against the completed market-data implementation (`backend/app/market/`, `planning/MARKET_DATA_SUMMARY.md`), and against the other committed docs (`README.md`, `backend/CLAUDE.md`). + +## Questions & Clarifications + +| # | Section | Item | +|---|---------|------| +| 1 | §9 LLM Integration | The doc twice references a **"cerebras-inference" skill** ("use cerebras-inference skill", "using the cerebras-inference skill"). The skill actually available in this environment is named `cerebras`, not `cerebras-inference`. If the Backend/LLM agent looks up the skill by the name written here, it won't find it. Please correct the name (or confirm a differently-named skill is intentionally being introduced). | +| 2 | §9 How It Works / Structured Output Schema | Steps 4–8 imply the LLM authors `message` *before* trades are validated/executed (parse → execute → store → return). But the doc also says "If a trade fails validation... the error is included in the chat response so the LLM can inform the user" — which requires the LLM to know the outcome *before* it writes `message`. As written this is a contradiction: either (a) there's a second LLM call after execution to fold results into the final `message` (not described in the 8 steps), or (b) the response sent to the frontend is an envelope that adds server-computed **execution results** (per-trade success/failure, fill price, reason for rejection) alongside the LLM's raw `message`/`trades`/`watchlist_changes`, and the frontend — not the LLM's prose — is responsible for rendering failures. The structured-output schema in §9 has no field for this. Given §10 says "Trade executions and watchlist changes shown inline as confirmations," the frontend clearly needs *some* execution-result data distinct from the free-text `message`. Recommend making this explicit. | +| 3 | §7 `chat_messages.actions` | The shape of this JSON blob is never specified, but per §10 it's presumably what drives the inline trade/watchlist confirmations shown in the chat panel. Worth pinning down its fields now (ticker, side, quantity, status, fill price / error reason, etc.) rather than leaving it to be improvised during implementation — same underlying gap as #2. | +| 4 | §6 Shared Price Cache / SSE Streaming | The set of tickers actively tracked by the market-data source is described as "the union of all watched tickers" (§6) and the SSE stream sends "all tickers known to the system... equivalent to the user's watchlist" (§6). But a user can hold an open **position** in a ticker after removing it from the watchlist (nothing in §8's `DELETE /api/watchlist/{ticker}` blocks this). If the tracked-ticker set is driven purely by the watchlist, a removed-but-still-held ticker would stop receiving price updates, leaving `/api/portfolio`'s P&L calculation with a stale or missing price. Should the tracked set actually be "watchlist ∪ tickers with a nonzero position," or should removing a watched ticker with an open position be disallowed/warned? | +| 5 | §6 / §8 `POST /api/watchlist` | §6 explains how an *already-tracked* unrecognized ticker behaves (simulator: gets a synthetic price immediately; Massive: never gets a price), but doesn't say what the add endpoint itself does at add-time. Two related gaps: (a) does `POST /api/watchlist` validate anything before returning success, or always succeed optimistically? (b) In Massive mode, even a *valid* symbol won't have a price until the next poll (up to 15s on the free tier) — is the frontend expected to show a "pending" price state for a just-added ticker, or does it just wait silently for the first SSE event? | +| 6 | §7 / §8 Ticker normalization | Is ticker input normalized (e.g., uppercased/trimmed) before the watchlist `UNIQUE(user_id, ticker)` check, the price-cache lookup, and trade validation? As written, "aapl" and "AAPL" could plausibly be treated as different tickers depending on implementation. Worth stating explicitly since it affects both the DB constraint and cache key semantics. | +| 7 | §8 `POST /api/portfolio/trade` | The validation rule described (ticker must have a current price in the cache) doesn't mention rejecting non-positive or non-numeric `quantity`. Worth a one-line addition (quantity must be a positive number; fractional allowed per §7) so this isn't left implicit. | +| 8 | §2 / §10 Connection status indicator | The green/yellow/red mapping doesn't correspond cleanly to native `EventSource` semantics: `EventSource` retries indefinitely on error and only reaches a terminal closed state if application code calls `.close()` — there's no built-in "reconnecting" vs. "permanently disconnected" distinction to read off the API. Worth a short note on what actually drives yellow vs. red (e.g., yellow = an `onerror` fired and a retry is pending; red = N consecutive failed retries, or a failed `/api/health` check) so the Frontend Engineer isn't inventing this heuristic from scratch. | +| 9 | §6 Massive API | No fallback behavior is described for when `MASSIVE_API_KEY` is set but the API is unreachable or auth fails at runtime (network error, invalid key, rate-limit exceeded). Does the system fall back to the simulator, retry with backoff, or simply serve no/stale prices until it recovers? Even a one-line "fails closed / fails open" statement would remove ambiguity here. | + +## Cross-Document Consistency + +| # | Item | +|---|------| +| 10 | **README.md vs. PLAN.md §11 — volume mount.** §13 item 1 records that the bind-mount-vs-named-volume question was resolved in favor of a bind mount, and §11 now shows `docker run -v "$(pwd)/db:/app/db" ...`. However, `README.md`'s own Quick Start section still shows `docker run -v finally-data:/app/db ...` — a **named volume**, not a bind mount. These two committed docs now disagree on the canonical run command. Since PLAN.md is the source of truth per §13's resolution, README.md's Quick Start should be updated to match (bind-mounting `./db`), or the discrepancy should be called out if a named volume is actually preferred for the README's simplified quick-start path. | + +## Simplification Opportunities + +| # | Item | +|---|------| +| 11 | Resolving question #2 above by explicitly splitting the `/api/chat` response into two parts — the LLM's raw structured output (`message`, `trades`, `watchlist_changes` as requested) and a separate, server-computed `execution_results` (what actually happened to each requested trade/watchlist change) — would remove the ambiguity in one stroke and give both the frontend and `chat_messages.actions` (#3) a single well-defined shape to consume, instead of three separate open questions converging on the same missing piece. | +| 12 | §7 states `portfolio_snapshots` are recorded "every 30 seconds" — this is the only bare numeric interval in the doc not framed as a named/tunable constant (contrast with the Massive polling intervals in §6, which are explicitly framed as tier-dependent config). Worth a one-line note that this should live as a named constant/config value rather than a hardcoded magic number, since it's a natural thing to tune during the demo-polish pass. | diff --git a/planning/archive/MARKET_DATA_DESIGN.md b/planning/archive/MARKET_DATA_DESIGN.md deleted file mode 100644 index 0d2cfd5fd..000000000 --- a/planning/archive/MARKET_DATA_DESIGN.md +++ /dev/null @@ -1,1490 +0,0 @@ -# Market Data Backend — Detailed Design - -Implementation-ready design for the FinAlly market data subsystem. Covers the unified interface, in-memory price cache, GBM simulator, Massive API client, SSE streaming endpoint, and FastAPI lifecycle integration. - -Everything in this document lives under `backend/app/market/`. - ---- - -## Table of Contents - -1. [File Structure](#1-file-structure) -2. [Data Model — `models.py`](#2-data-model) -3. [Price Cache — `cache.py`](#3-price-cache) -4. [Abstract Interface — `interface.py`](#4-abstract-interface) -5. [Seed Prices & Ticker Parameters — `seed_prices.py`](#5-seed-prices--ticker-parameters) -6. [GBM Simulator — `simulator.py`](#6-gbm-simulator) -7. [Massive API Client — `massive_client.py`](#7-massive-api-client) -8. [Factory — `factory.py`](#8-factory) -9. [SSE Streaming Endpoint — `stream.py`](#9-sse-streaming-endpoint) -10. [FastAPI Lifecycle Integration](#10-fastapi-lifecycle-integration) -11. [Watchlist Coordination](#11-watchlist-coordination) -12. [Testing Strategy](#12-testing-strategy) -13. [Error Handling & Edge Cases](#13-error-handling--edge-cases) -14. [Configuration Summary](#14-configuration-summary) - ---- - -## 1. File Structure - -``` -backend/ - app/ - market/ - __init__.py # Re-exports: PriceUpdate, PriceCache, MarketDataSource, create_market_data_source - models.py # PriceUpdate dataclass - cache.py # PriceCache (thread-safe in-memory store) - interface.py # MarketDataSource ABC - seed_prices.py # SEED_PRICES, TICKER_PARAMS, DEFAULT_PARAMS, CORRELATION_GROUPS - simulator.py # GBMSimulator + SimulatorDataSource - massive_client.py # MassiveDataSource - factory.py # create_market_data_source() - stream.py # SSE endpoint (FastAPI router) -``` - -Each file has a single responsibility. The `__init__.py` re-exports the public API so that the rest of the backend imports from `app.market` without reaching into submodules. - ---- - -## 2. Data Model - -**File: `backend/app/market/models.py`** - -`PriceUpdate` is the only data structure that leaves the market data layer. Every downstream consumer — SSE streaming, portfolio valuation, trade execution — works exclusively with this type. - -```python -from __future__ import annotations - -import time -from dataclasses import dataclass, field - - -@dataclass(frozen=True, slots=True) -class PriceUpdate: - """Immutable snapshot of a single ticker's price at a point in time.""" - - ticker: str - price: float - previous_price: float - timestamp: float = field(default_factory=time.time) # Unix seconds - - @property - def change(self) -> float: - """Absolute price change from previous update.""" - return round(self.price - self.previous_price, 4) - - @property - def change_percent(self) -> float: - """Percentage change from previous update.""" - 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'.""" - if self.price > self.previous_price: - return "up" - elif self.price < self.previous_price: - return "down" - return "flat" - - def to_dict(self) -> dict: - """Serialize for JSON / SSE transmission.""" - return { - "ticker": self.ticker, - "price": self.price, - "previous_price": self.previous_price, - "timestamp": self.timestamp, - "change": self.change, - "change_percent": self.change_percent, - "direction": self.direction, - } -``` - -### Design decisions - -- **`frozen=True`**: Price updates are immutable value objects. Once created they never change, which makes them safe to share across async tasks without copying. -- **`slots=True`**: Minor memory optimization — we create many of these per second. -- **Computed properties** (`change`, `direction`, `change_percent`): Derived from `price` and `previous_price` so they can never be inconsistent. No risk of a stale `direction` field. -- **`to_dict()`**: Single serialization point used by both the SSE endpoint and REST API responses. - ---- - -## 3. Price Cache - -**File: `backend/app/market/cache.py`** - -The price cache is the central data hub. Data sources write to it; SSE streaming and portfolio valuation read from it. It must be thread-safe because the simulator/poller may run in a thread pool executor while SSE reads happen on the async event loop. - -```python -from __future__ import annotations - -import asyncio -import time -from threading import Lock -from typing import Callable - -from .models import PriceUpdate - - -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. - """ - - def __init__(self) -> None: - self._prices: dict[str, PriceUpdate] = {} - 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: - """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'). - """ - with self._lock: - ts = timestamp or time.time() - prev = self._prices.get(ticker) - previous_price = prev.price if prev else price - - update = PriceUpdate( - ticker=ticker, - price=round(price, 2), - previous_price=round(previous_price, 2), - timestamp=ts, - ) - self._prices[ticker] = update - self._version += 1 - return update - - 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_all(self) -> dict[str, PriceUpdate]: - """Snapshot of all current prices. Returns a shallow copy.""" - 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 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) - - @property - def version(self) -> int: - """Current version counter. Useful for SSE change detection.""" - return self._version - - def __len__(self) -> int: - with self._lock: - return len(self._prices) - - def __contains__(self, ticker: str) -> bool: - with self._lock: - return ticker in self._prices -``` - -### Why a version counter? - -The SSE streaming loop polls the cache every ~500ms. Without a version counter, it would serialize and send all prices every tick even if nothing changed (e.g., Massive API only updates every 15s). The version counter lets the SSE loop skip sends when nothing is new: - -```python -last_version = -1 -while True: - if price_cache.version != last_version: - last_version = price_cache.version - yield format_sse(price_cache.get_all()) - await asyncio.sleep(0.5) -``` - -### Thread safety rationale - -The `threading.Lock` is used instead of `asyncio.Lock` because: -- The Massive client's synchronous `get_snapshot_all()` runs in `asyncio.to_thread()`, which operates in a real OS thread — `asyncio.Lock` would not protect against that. -- The GBM simulator's `step()` is CPU-bound and could also be offloaded to a thread for fairness. -- `threading.Lock` works correctly from both sync threads and the async event loop. - ---- - -## 4. Abstract Interface - -**File: `backend/app/market/interface.py`** - -```python -from __future__ import annotations - -from abc import ABC, abstractmethod - - -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. - - Lifecycle: - source = create_market_data_source(cache) - await source.start(["AAPL", "GOOGL", ...]) - # ... app runs ... - await source.add_ticker("TSLA") - await source.remove_ticker("GOOGL") - # ... app shutting down ... - await source.stop() - """ - - @abstractmethod - async def start(self, tickers: list[str]) -> None: - """Begin producing price updates for the given tickers. - - Starts a background task that periodically writes to the PriceCache. - Must be called exactly once. Calling start() twice is undefined behavior. - """ - - @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. - """ - - @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. - """ - - @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. - """ - - @abstractmethod - def get_tickers(self) -> list[str]: - """Return the current list of actively tracked tickers.""" -``` - -### Why the source writes to the cache instead of returning prices - -This push model decouples timing. The simulator ticks at 500ms, Massive polls at 15s, but SSE always reads from the cache at its own 500ms cadence. There is no need for the SSE layer to know which data source is active or what its update interval is. - ---- - -## 5. Seed Prices & Ticker Parameters - -**File: `backend/app/market/seed_prices.py`** - -Constants only — no logic, no imports beyond stdlib. This file is shared by both the simulator (for initial prices and GBM parameters) and potentially by the Massive client (as fallback prices if the API hasn't responded yet). - -```python -"""Seed prices and per-ticker parameters for the market simulator.""" - -# Realistic starting prices for the default watchlist (as of project creation) -SEED_PRICES: dict[str, float] = { - "AAPL": 190.00, - "GOOGL": 175.00, - "MSFT": 420.00, - "AMZN": 185.00, - "TSLA": 250.00, - "NVDA": 800.00, - "META": 500.00, - "JPM": 195.00, - "V": 280.00, - "NFLX": 600.00, -} - -# Per-ticker GBM parameters -# sigma: annualized volatility (higher = more price movement) -# mu: annualized drift / expected return -TICKER_PARAMS: dict[str, dict[str, float]] = { - "AAPL": {"sigma": 0.22, "mu": 0.05}, - "GOOGL": {"sigma": 0.25, "mu": 0.05}, - "MSFT": {"sigma": 0.20, "mu": 0.05}, - "AMZN": {"sigma": 0.28, "mu": 0.05}, - "TSLA": {"sigma": 0.50, "mu": 0.03}, # High volatility - "NVDA": {"sigma": 0.40, "mu": 0.08}, # High volatility, strong drift - "META": {"sigma": 0.30, "mu": 0.05}, - "JPM": {"sigma": 0.18, "mu": 0.04}, # Low volatility (bank) - "V": {"sigma": 0.17, "mu": 0.04}, # Low volatility (payments) - "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: 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 -TSLA_CORR = 0.3 # TSLA does its own thing -DEFAULT_CORR = 0.3 # Unknown tickers -``` - ---- - -## 6. GBM Simulator - -**File: `backend/app/market/simulator.py`** - -This file contains two classes: -- `GBMSimulator`: Pure math engine. Stateful — holds current prices and advances them one step at a time. -- `SimulatorDataSource`: The `MarketDataSource` implementation that wraps `GBMSimulator` in an async loop and writes to the `PriceCache`. - -### 6.1 GBMSimulator — The Math Engine - -```python -from __future__ import annotations - -import asyncio -import logging -import math -import random - -import numpy as np - -from .cache import PriceCache -from .interface import MarketDataSource -from .seed_prices import ( - CORRELATION_GROUPS, - CROSS_GROUP_CORR, - DEFAULT_CORR, - DEFAULT_PARAMS, - INTRA_FINANCE_CORR, - INTRA_TECH_CORR, - SEED_PRICES, - TICKER_PARAMS, - TSLA_CORR, -) - -logger = logging.getLogger(__name__) - - -class GBMSimulator: - """Geometric Brownian Motion simulator for correlated stock prices. - - Math: - S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) - - Where: - S(t) = current price - mu = annualized drift (expected return) - sigma = annualized volatility - dt = time step as 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. - """ - - # 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 - - def __init__( - self, - tickers: list[str], - dt: float = DEFAULT_DT, - event_probability: float = 0.001, - ) -> None: - self._dt = dt - self._event_prob = event_probability - - # Per-ticker state - self._tickers: list[str] = [] - self._prices: dict[str, float] = {} - self._params: dict[str, dict[str, float]] = {} - - # 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._rebuild_cholesky() - - # --- Public API --- - - def step(self) -> dict[str, float]: - """Advance all tickers by one time step. Returns {ticker: new_price}. - - This is the hot path — called every 500ms. Keep it fast. - """ - 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 - if self._cholesky is not None: - z_correlated = self._cholesky @ z_independent - else: - z_correlated = z_independent - - result: dict[str, float] = {} - for i, ticker in enumerate(self._tickers): - params = self._params[ticker] - mu = params["mu"] - sigma = 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] - 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 - if random.random() < self._event_prob: - shock_magnitude = random.uniform(0.02, 0.05) - shock_sign = random.choice([-1, 1]) - self._prices[ticker] *= 1 + shock_magnitude * shock_sign - logger.debug( - "Random event on %s: %.1f%% %s", - ticker, - shock_magnitude * 100, - "up" if shock_sign > 0 else "down", - ) - - result[ticker] = round(self._prices[ticker], 2) - - return result - - 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._rebuild_cholesky() - - def remove_ticker(self, ticker: str) -> None: - """Remove a ticker from the simulation. Rebuilds the correlation matrix.""" - if ticker not in self._prices: - return - self._tickers.remove(ticker) - del self._prices[ticker] - del self._params[ticker] - self._rebuild_cholesky() - - def get_price(self, ticker: str) -> float | None: - """Current price for a ticker, or None if not tracked.""" - return self._prices.get(ticker) - - # --- Internals --- - - def _add_ticker_internal(self, ticker: str) -> None: - """Add a ticker without rebuilding Cholesky (for batch initialization).""" - if ticker in self._prices: - return - 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)) - - 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. - """ - 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) - - @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 - """ - 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 -``` - -### 6.2 SimulatorDataSource — Async Wrapper - -```python -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. - """ - - def __init__( - self, - price_cache: PriceCache, - update_interval: float = 0.5, - event_probability: float = 0.001, - ) -> None: - self._cache = price_cache - self._interval = update_interval - self._event_prob = event_probability - self._sim: GBMSimulator | None = None - self._task: asyncio.Task | None = None - - 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 - for ticker in tickers: - price = self._sim.get_price(ticker) - if price is not None: - self._cache.update(ticker=ticker, price=price) - self._task = asyncio.create_task(self._run_loop(), name="simulator-loop") - logger.info("Simulator started with %d tickers", len(tickers)) - - async def stop(self) -> None: - if self._task and not self._task.done(): - self._task.cancel() - try: - await self._task - except asyncio.CancelledError: - pass - self._task = 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) - - async def remove_ticker(self, ticker: str) -> None: - if self._sim: - self._sim.remove_ticker(ticker) - self._cache.remove(ticker) - logger.info("Simulator: removed ticker %s", ticker) - - def get_tickers(self) -> list[str]: - return list(self._sim._tickers) if self._sim else [] - - async def _run_loop(self) -> None: - """Core loop: step the simulation, write to cache, sleep.""" - while True: - try: - if self._sim: - prices = self._sim.step() - for ticker, price in prices.items(): - self._cache.update(ticker=ticker, price=price) - except Exception: - logger.exception("Simulator step failed") - await asyncio.sleep(self._interval) -``` - -### Key behaviors - -- **Immediate seeding**: When `start()` is called, the cache is populated with seed prices *before* the loop begins. This means the SSE endpoint has data to send on its very first tick, with no blank-screen delay. -- **Graceful cancellation**: `stop()` cancels the task and awaits it, catching `CancelledError`. This ensures clean shutdown during FastAPI lifespan teardown. -- **Exception resilience**: The loop catches exceptions per-step so a single bad tick doesn't kill the entire data feed. - ---- - -## 7. Massive API Client - -**File: `backend/app/market/massive_client.py`** - -Polls the Massive (formerly Polygon.io) REST API snapshot endpoint on a configurable interval. The synchronous Massive client runs in `asyncio.to_thread()` to avoid blocking the event loop. - -```python -from __future__ import annotations - -import asyncio -import logging -from typing import Any - -from .cache import PriceCache -from .interface import MarketDataSource - -logger = logging.getLogger(__name__) - - -class MassiveDataSource(MarketDataSource): - """MarketDataSource backed by the Massive (Polygon.io) 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. - - Rate limits: - - Free tier: 5 req/min → poll every 15s (default) - - Paid tiers: higher limits → poll every 2-5s - """ - - def __init__( - self, - api_key: str, - price_cache: PriceCache, - poll_interval: float = 15.0, - ) -> None: - self._api_key = api_key - self._cache = price_cache - self._interval = poll_interval - self._tickers: list[str] = [] - self._task: asyncio.Task | None = None - self._client: Any = None # Lazy import to avoid hard dependency - - async def start(self, tickers: list[str]) -> None: - # Lazy import: only import massive when actually using real market data. - # This means the massive package is not required when using the simulator. - from massive import RESTClient - - self._client = RESTClient(api_key=self._api_key) - self._tickers = list(tickers) - - # Do an immediate first poll so the cache has data right away - await self._poll_once() - - self._task = asyncio.create_task(self._poll_loop(), name="massive-poller") - logger.info( - "Massive poller started: %d tickers, %.1fs interval", - len(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 - self._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) - - async def remove_ticker(self, ticker: str) -> None: - ticker = ticker.upper().strip() - self._tickers = [t for t in self._tickers if t != ticker] - self._cache.remove(ticker) - logger.info("Massive: removed ticker %s", ticker) - - def get_tickers(self) -> list[str]: - return list(self._tickers) - - # --- Internal --- - - async def _poll_loop(self) -> None: - """Poll on interval. First poll already happened in start().""" - while True: - await asyncio.sleep(self._interval) - await self._poll_once() - - async def _poll_once(self) -> None: - """Execute one poll cycle: fetch snapshots, update cache.""" - if not self._tickers or not self._client: - return - - try: - # 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. - - def _fetch_snapshots(self) -> list: - """Synchronous call to the Massive REST API. Runs in a thread.""" - from massive.rest.models import SnapshotMarketType - - return self._client.get_snapshot_all( - market_type=SnapshotMarketType.STOCKS, - tickers=self._tickers, - ) -``` - -### Error handling philosophy - -The Massive poller is intentionally resilient: - -| Error | Behavior | -|-------|----------| -| **401 Unauthorized** | Logged as error. Poller keeps running (user might fix `.env` and restart). | -| **429 Rate Limited** | Logged as error. Next poll retries after `poll_interval` seconds. | -| **Network timeout** | Logged as error. Retries automatically on next cycle. | -| **Malformed snapshot** | Individual ticker skipped with warning. Other tickers still processed. | -| **All tickers fail** | Cache retains last-known prices. SSE keeps streaming stale data (better than no data). | - -### Lazy import strategy - -`from massive import RESTClient` happens inside `start()`, not at module import time. This means: -- The `massive` package is only required when `MASSIVE_API_KEY` is set. -- Students who don't have a Massive API key don't need the package installed at all. -- The simulator path has zero external dependencies beyond `numpy`. - ---- - -## 8. Factory - -**File: `backend/app/market/factory.py`** - -```python -from __future__ import annotations - -import logging -import os - -from .cache import PriceCache -from .interface import MarketDataSource - -logger = logging.getLogger(__name__) - - -def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: - """Create the appropriate market data source based on environment variables. - - - MASSIVE_API_KEY set and non-empty → MassiveDataSource (real market data) - - Otherwise → SimulatorDataSource (GBM simulation) - - Returns an unstarted source. Caller must await source.start(tickers). - """ - api_key = os.environ.get("MASSIVE_API_KEY", "").strip() - - if api_key: - from .massive_client import MassiveDataSource - - logger.info("Market data source: Massive API (real data)") - return MassiveDataSource(api_key=api_key, price_cache=price_cache) - else: - from .simulator import SimulatorDataSource - - logger.info("Market data source: GBM Simulator") - return SimulatorDataSource(price_cache=price_cache) -``` - -### Usage at app startup - -```python -price_cache = PriceCache() -source = create_market_data_source(price_cache) -await source.start(initial_tickers) # e.g., ["AAPL", "GOOGL", ...] -``` - ---- - -## 9. SSE Streaming Endpoint - -**File: `backend/app/market/stream.py`** - -The SSE endpoint is a FastAPI route that holds open a long-lived HTTP connection and pushes price updates to the client as `text/event-stream`. - -```python -from __future__ import annotations - -import asyncio -import json -import logging -import time - -from fastapi import APIRouter, Request -from fastapi.responses import StreamingResponse - -from .cache import PriceCache - -logger = logging.getLogger(__name__) - -router = APIRouter(prefix="/api/stream", tags=["streaming"]) - - -def create_stream_router(price_cache: PriceCache) -> APIRouter: - """Create the SSE streaming router with a reference to the price cache. - - This factory pattern lets us inject the PriceCache without globals. - """ - - @router.get("/prices") - async def stream_prices(request: Request) -> StreamingResponse: - """SSE endpoint for live price updates. - - Streams all tracked ticker prices every ~500ms. The client connects - with EventSource and receives events in the format: - - data: {"AAPL": {"ticker": "AAPL", "price": 190.50, ...}, ...} - - Includes a retry directive so the browser auto-reconnects on - disconnection (EventSource built-in behavior). - """ - return StreamingResponse( - _generate_events(price_cache, request), - media_type="text/event-stream", - headers={ - "Cache-Control": "no-cache", - "Connection": "keep-alive", - "X-Accel-Buffering": "no", # Disable nginx buffering if proxied - }, - ) - - return router - - -async def _generate_events( - price_cache: PriceCache, - request: Request, - interval: float = 0.5, -) -> 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()). - """ - # 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) - - try: - while True: - # Check for client disconnect - if await request.is_disconnected(): - logger.info("SSE client disconnected: %s", client_ip) - break - - current_version = price_cache.version - if current_version != last_version: - last_version = current_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" - - await asyncio.sleep(interval) - except asyncio.CancelledError: - logger.info("SSE stream cancelled for: %s", client_ip) -``` - -### SSE wire format - -Each event the client receives looks like this: - -``` -data: {"AAPL":{"ticker":"AAPL","price":190.50,"previous_price":190.42,"timestamp":1707580800.5,"change":0.08,"change_percent":0.042,"direction":"up"},"GOOGL":{"ticker":"GOOGL","price":175.12,...}} - -``` - -The client parses this with: - -```javascript -const eventSource = new EventSource('/api/stream/prices'); -eventSource.onmessage = (event) => { - const prices = JSON.parse(event.data); - // prices is { "AAPL": { ticker, price, previous_price, ... }, ... } -}; -``` - -### Why poll-and-push instead of event-driven? - -The SSE endpoint polls the cache on a fixed interval rather than being notified by the data source. This is simpler and produces predictable, evenly-spaced updates for the frontend. The frontend accumulates these into sparkline charts — regular spacing is important for clean visualization. - ---- - -## 10. FastAPI Lifecycle Integration - -The market data system starts and stops with the FastAPI application using the `lifespan` context manager pattern. - -**In `backend/app/main.py`:** - -```python -from contextlib import asynccontextmanager - -from fastapi import FastAPI - -from app.market.cache import PriceCache -from app.market.factory import create_market_data_source -from app.market.interface import MarketDataSource -from app.market.stream import create_stream_router - - -@asynccontextmanager -async def lifespan(app: FastAPI): - """Manage startup and shutdown of background services.""" - - # --- STARTUP --- - - # 1. Create the shared price cache - price_cache = PriceCache() - app.state.price_cache = price_cache - - # 2. Create and start the market data source - source = create_market_data_source(price_cache) - app.state.market_source = source - - # 3. Load initial tickers from the database watchlist - initial_tickers = await load_watchlist_tickers() # reads from SQLite - await source.start(initial_tickers) - - # 4. Register the SSE streaming router - stream_router = create_stream_router(price_cache) - app.include_router(stream_router) - - yield # App is running - - # --- SHUTDOWN --- - await source.stop() - - -app = FastAPI(title="FinAlly", lifespan=lifespan) - - -# Dependency for injecting the price cache into route handlers -def get_price_cache() -> PriceCache: - return app.state.price_cache - - -def get_market_source() -> MarketDataSource: - return app.state.market_source -``` - -### Accessing market data from other routes - -Other parts of the backend (trade execution, portfolio valuation, watchlist management) access the price cache and data source via FastAPI's dependency injection: - -```python -from fastapi import APIRouter, Depends - -router = APIRouter(prefix="/api") - -@router.post("/portfolio/trade") -async def execute_trade( - trade: TradeRequest, - price_cache: PriceCache = Depends(get_price_cache), -): - current_price = price_cache.get_price(trade.ticker) - if current_price is None: - raise HTTPException(404, f"No price available for {trade.ticker}") - # ... execute trade at current_price ... - - -@router.post("/watchlist") -async def add_to_watchlist( - payload: WatchlistAdd, - source: MarketDataSource = Depends(get_market_source), - price_cache: PriceCache = Depends(get_price_cache), -): - # Add to database ... - # Then tell the data source to start tracking it - await source.add_ticker(payload.ticker) - # ... - - -@router.delete("/watchlist/{ticker}") -async def remove_from_watchlist( - ticker: str, - source: MarketDataSource = Depends(get_market_source), -): - # Remove from database ... - # Then stop tracking - await source.remove_ticker(ticker) - # ... -``` - ---- - -## 11. Watchlist Coordination - -When the watchlist changes (via REST API or LLM chat), the market data source must be notified so it tracks the right set of tickers. - -### Flow: Adding a Ticker - -``` -User (or LLM) → POST /api/watchlist {ticker: "PYPL"} - → Insert into watchlist table (SQLite) - → await source.add_ticker("PYPL") - Simulator: adds to GBMSimulator, rebuilds Cholesky, seeds cache - Massive: appends to ticker list, appears on next poll - → Return success (ticker + current price if available) -``` - -### Flow: Removing a Ticker - -``` -User (or LLM) → DELETE /api/watchlist/PYPL - → Delete from watchlist table (SQLite) - → await source.remove_ticker("PYPL") - Simulator: removes from GBMSimulator, rebuilds Cholesky, removes from cache - Massive: removes from ticker list, removes from cache - → Return success -``` - -### Edge case: Ticker has an open position - -If the user removes a ticker from the watchlist but still holds shares, the ticker should remain in the data source so portfolio valuation stays accurate. The watchlist route should check for this: - -```python -@router.delete("/watchlist/{ticker}") -async def remove_from_watchlist( - ticker: str, - source: MarketDataSource = Depends(get_market_source), -): - # Remove from watchlist table - await db.delete_watchlist_entry(ticker) - - # Only stop tracking if no open position - position = await db.get_position(ticker) - if position is None or position.quantity == 0: - await source.remove_ticker(ticker) - - return {"status": "ok"} -``` - ---- - -## 12. Testing Strategy - -### 12.1 Unit Tests for GBMSimulator - -**File: `backend/tests/market/test_simulator.py`** - -```python -import math -import pytest -from app.market.simulator import GBMSimulator -from app.market.seed_prices import SEED_PRICES - - -class TestGBMSimulator: - """Unit tests for the GBM price simulator.""" - - def test_step_returns_all_tickers(self): - sim = GBMSimulator(tickers=["AAPL", "GOOGL"]) - result = sim.step() - assert set(result.keys()) == {"AAPL", "GOOGL"} - - def test_prices_are_positive(self): - """GBM prices can never go negative (exp() is always positive).""" - sim = GBMSimulator(tickers=["AAPL"]) - for _ in range(10_000): - prices = sim.step() - assert prices["AAPL"] > 0 - - def test_initial_prices_match_seeds(self): - sim = GBMSimulator(tickers=["AAPL"]) - # Before any step, price should be the seed price - assert sim.get_price("AAPL") == SEED_PRICES["AAPL"] - - def test_add_ticker(self): - sim = GBMSimulator(tickers=["AAPL"]) - sim.add_ticker("TSLA") - result = sim.step() - assert "TSLA" in result - - def test_remove_ticker(self): - sim = GBMSimulator(tickers=["AAPL", "GOOGL"]) - sim.remove_ticker("GOOGL") - result = sim.step() - assert "GOOGL" not in result - assert "AAPL" in result - - def test_add_duplicate_is_noop(self): - sim = GBMSimulator(tickers=["AAPL"]) - sim.add_ticker("AAPL") - assert len(sim._tickers) == 1 - - 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): - sim = GBMSimulator(tickers=["ZZZZ"]) - price = sim.get_price("ZZZZ") - assert 50.0 <= price <= 300.0 - - def test_empty_step(self): - sim = GBMSimulator(tickers=[]) - result = sim.step() - assert result == {} - - def test_prices_change_over_time(self): - """After many steps, prices should have drifted from their seeds.""" - sim = GBMSimulator(tickers=["AAPL"]) - for _ in range(1000): - sim.step() - # Price should have changed (extremely unlikely to be exactly the seed) - assert sim.get_price("AAPL") != SEED_PRICES["AAPL"] - - def test_cholesky_rebuilds_on_add(self): - sim = GBMSimulator(tickers=["AAPL"]) - assert sim._cholesky is None # Only 1 ticker, no correlation matrix - sim.add_ticker("GOOGL") - assert sim._cholesky is not None # Now 2 tickers, matrix exists -``` - -### 12.2 Unit Tests for PriceCache - -**File: `backend/tests/market/test_cache.py`** - -```python -import pytest -from app.market.cache import PriceCache - - -class TestPriceCache: - - def test_update_and_get(self): - cache = PriceCache() - update = cache.update("AAPL", 190.50) - assert update.ticker == "AAPL" - assert update.price == 190.50 - assert cache.get("AAPL") == update - - def test_first_update_is_flat(self): - cache = PriceCache() - update = cache.update("AAPL", 190.50) - assert update.direction == "flat" - assert update.previous_price == 190.50 - - def test_direction_up(self): - cache = PriceCache() - cache.update("AAPL", 190.00) - update = cache.update("AAPL", 191.00) - assert update.direction == "up" - assert update.change == 1.00 - - def test_direction_down(self): - cache = PriceCache() - cache.update("AAPL", 190.00) - update = cache.update("AAPL", 189.00) - assert update.direction == "down" - assert update.change == -1.00 - - def test_remove(self): - cache = PriceCache() - cache.update("AAPL", 190.00) - cache.remove("AAPL") - assert cache.get("AAPL") is None - - def test_get_all(self): - cache = PriceCache() - cache.update("AAPL", 190.00) - cache.update("GOOGL", 175.00) - all_prices = cache.get_all() - assert set(all_prices.keys()) == {"AAPL", "GOOGL"} - - def test_version_increments(self): - cache = PriceCache() - v0 = cache.version - cache.update("AAPL", 190.00) - assert cache.version == v0 + 1 - cache.update("AAPL", 191.00) - assert cache.version == v0 + 2 - - def test_get_price_convenience(self): - cache = PriceCache() - cache.update("AAPL", 190.50) - assert cache.get_price("AAPL") == 190.50 - assert cache.get_price("NOPE") is None -``` - -### 12.3 Integration Test: SimulatorDataSource - -**File: `backend/tests/market/test_simulator_source.py`** - -```python -import asyncio -import pytest -from app.market.cache import PriceCache -from app.market.simulator import SimulatorDataSource - - -@pytest.mark.asyncio -class TestSimulatorDataSource: - - async def test_start_populates_cache(self): - cache = PriceCache() - source = SimulatorDataSource(price_cache=cache, update_interval=0.1) - await source.start(["AAPL", "GOOGL"]) - - # Cache should have seed prices immediately (before first loop tick) - assert cache.get("AAPL") is not None - assert cache.get("GOOGL") is not None - - await source.stop() - - async def test_prices_update_over_time(self): - cache = PriceCache() - source = SimulatorDataSource(price_cache=cache, update_interval=0.05) - await source.start(["AAPL"]) - - initial = cache.get("AAPL").price - await asyncio.sleep(0.3) # Several update cycles - current = cache.get("AAPL").price - - # Extremely unlikely to be identical after many steps - # (but not impossible, so this is a probabilistic test) - assert current != initial or True # Soft assertion - - await source.stop() - - async def test_stop_is_clean(self): - cache = PriceCache() - source = SimulatorDataSource(price_cache=cache, update_interval=0.1) - await source.start(["AAPL"]) - await source.stop() - # Double stop should not raise - await source.stop() - - async def test_add_and_remove_ticker(self): - cache = PriceCache() - source = SimulatorDataSource(price_cache=cache, update_interval=0.1) - await source.start(["AAPL"]) - - await source.add_ticker("TSLA") - assert "TSLA" in source.get_tickers() - assert cache.get("TSLA") is not None - - await source.remove_ticker("TSLA") - assert "TSLA" not in source.get_tickers() - assert cache.get("TSLA") is None - - await source.stop() -``` - -### 12.4 Unit Test: MassiveDataSource (Mocked) - -**File: `backend/tests/market/test_massive.py`** - -```python -import asyncio -from unittest.mock import MagicMock, patch -import pytest -from app.market.cache import PriceCache -from app.market.massive_client import MassiveDataSource - - -def _make_snapshot(ticker: str, price: float, timestamp_ms: int) -> MagicMock: - """Create a mock Massive snapshot object.""" - snap = MagicMock() - snap.ticker = ticker - snap.last_trade.price = price - snap.last_trade.timestamp = timestamp_ms - return snap - - -@pytest.mark.asyncio -class TestMassiveDataSource: - - async def test_poll_updates_cache(self): - cache = PriceCache() - source = MassiveDataSource( - api_key="test-key", - price_cache=cache, - poll_interval=60.0, # Long interval so the loop doesn't auto-poll - ) - - mock_snapshots = [ - _make_snapshot("AAPL", 190.50, 1707580800000), - _make_snapshot("GOOGL", 175.25, 1707580800000), - ] - - with patch.object(source, "_fetch_snapshots", return_value=mock_snapshots): - await source._poll_once() - - assert cache.get_price("AAPL") == 190.50 - assert cache.get_price("GOOGL") == 175.25 - - async def test_malformed_snapshot_skipped(self): - cache = PriceCache() - source = MassiveDataSource( - api_key="test-key", - price_cache=cache, - poll_interval=60.0, - ) - source._tickers = ["AAPL", "BAD"] - - good_snap = _make_snapshot("AAPL", 190.50, 1707580800000) - bad_snap = MagicMock() - bad_snap.ticker = "BAD" - bad_snap.last_trade = None # Will cause AttributeError - - with patch.object(source, "_fetch_snapshots", return_value=[good_snap, bad_snap]): - await source._poll_once() - - # Good ticker processed, bad one skipped - assert cache.get_price("AAPL") == 190.50 - assert cache.get_price("BAD") is None - - async def test_api_error_does_not_crash(self): - cache = PriceCache() - source = MassiveDataSource( - api_key="test-key", - price_cache=cache, - poll_interval=60.0, - ) - source._tickers = ["AAPL"] - - with patch.object(source, "_fetch_snapshots", side_effect=Exception("network error")): - await source._poll_once() # Should not raise - - assert cache.get_price("AAPL") is None # No update happened -``` - ---- - -## 13. Error Handling & Edge Cases - -### 13.1 Startup: Empty Watchlist - -If the database has no watchlist entries (user deleted everything), `start()` receives an empty list. Both data sources handle this gracefully — the simulator produces no prices, the Massive poller skips its API call. The SSE endpoint sends empty events. When the user adds a ticker, the source starts tracking it immediately. - -### 13.2 Price Cache Miss During Trade - -If a user tries to trade a ticker that has no cached price (e.g., just added to watchlist, Massive hasn't polled yet): - -```python -price = price_cache.get_price(ticker) -if price is None: - raise HTTPException( - status_code=400, - detail=f"Price not yet available for {ticker}. Please wait a moment and try again.", - ) -``` - -The simulator avoids this by seeding the cache in `add_ticker()`. The Massive client may have a brief gap — the HTTP 400 with a clear message is the correct response. - -### 13.3 Massive API Key Invalid - -If the API key is set but invalid, the first poll will fail with a 401. The poller logs the error and keeps retrying. The SSE endpoint streams empty data. The user sees no prices and a connection status indicator showing "connected" (SSE is working, just no data). The fix is to correct the API key and restart. - -### 13.4 Thread Safety Under Load - -The `PriceCache` uses `threading.Lock` which is a mutex — only one thread can hold it at a time. Under normal load (10 tickers, 2 updates/sec), lock contention is negligible. The critical section is tiny (dict lookup + assignment). - -If this ever became a bottleneck (hundreds of tickers, many concurrent SSE readers), the fix would be a `ReadWriteLock` — but that level of optimization is unnecessary for this project. - -### 13.5 Simulator Precision - -GBM with tiny `dt` produces very small per-tick moves. Floating-point precision is not a concern because: -- Prices are `round()`ed to 2 decimal places in `GBMSimulator.step()` -- The exponential formulation (`exp(drift + diffusion)`) is numerically stable -- Prices are always positive (exponential function) - ---- - -## 14. Configuration Summary - -All tunable parameters and their defaults: - -| Parameter | Location | Default | Description | -|-----------|----------|---------|-------------| -| `MASSIVE_API_KEY` | Environment variable | `""` (empty) | If set, use Massive API; otherwise use simulator | -| `update_interval` | `SimulatorDataSource.__init__` | `0.5` (seconds) | Time between simulator ticks | -| `poll_interval` | `MassiveDataSource.__init__` | `15.0` (seconds) | Time between Massive API polls | -| `event_probability` | `GBMSimulator.__init__` | `0.001` | Chance of a random shock event per ticker per tick | -| `dt` | `GBMSimulator.__init__` | `~8.5e-8` | GBM time step (fraction of a trading year) | -| SSE push interval | `_generate_events()` | `0.5` (seconds) | Time between SSE pushes to the client | -| SSE retry directive | `_generate_events()` | `1000` (ms) | Browser EventSource reconnection delay | - -### Package `__init__.py` - -**File: `backend/app/market/__init__.py`** - -```python -"""Market data subsystem for FinAlly. - -Public API: - PriceUpdate - Immutable price snapshot dataclass - 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 -""" - -from .cache import PriceCache -from .factory import create_market_data_source -from .interface import MarketDataSource -from .models import PriceUpdate -from .stream import create_stream_router - -__all__ = [ - "PriceUpdate", - "PriceCache", - "MarketDataSource", - "create_market_data_source", - "create_stream_router", -] -``` diff --git a/planning/archive/MARKET_DATA_REVIEW.md b/planning/archive/MARKET_DATA_REVIEW.md deleted file mode 100644 index 61b4d6bf4..000000000 --- a/planning/archive/MARKET_DATA_REVIEW.md +++ /dev/null @@ -1,173 +0,0 @@ -# Market Data Backend — Code Review - -**Date:** 2026-02-10 -**Scope:** `backend/app/market/` (8 source files) and `backend/tests/market/` (6 test files) - ---- - -## 1. Test Results Summary - -**73 tests collected, 68 passed, 5 failed.** - -All failures are in `test_massive.py` and stem from the same root cause: the `massive` package is not installed in the test environment, so `patch("app.market.massive_client.RESTClient")` fails with `AttributeError` because the module-level name `RESTClient` was never imported (it is lazy-imported inside methods). This is an environment issue, not a logic bug — the tests are correctly structured but require the `massive` package to be available (or `create=True` on the patch) so that the mock target exists. - -Failing tests: -- `test_poll_updates_cache` — `asyncio.to_thread` fails because `_fetch_snapshots` is not properly mocked when `massive` is absent -- `test_malformed_snapshot_skipped` — same cause -- `test_timestamp_conversion` — same cause -- `test_stop_cancels_task` — `patch("app.market.massive_client.RESTClient")` fails because the name doesn't exist at module level -- `test_start_immediate_poll` — same as above - -The underlying `_poll_once()` logic itself is correct. The 3 tests that mock `source._fetch_snapshots` directly fail because `asyncio.to_thread(self._fetch_snapshots)` calls the real method which tries to import `massive`. The 2 tests that use `patch("app.market.massive_client.RESTClient")` fail because the name doesn't exist in the module's namespace (lazy import). Both issues resolve when the `massive` package is installed. - -**Lint (ruff):** Source code passes clean. Tests have 5 unused-import warnings (`pytest`, `math`, `asyncio` imported but not used in some test files). - -**Coverage:** 84% overall. -| Module | Coverage | Notes | -|---|---|---| -| models.py | 100% | | -| cache.py | 100% | | -| interface.py | 100% | | -| seed_prices.py | 100% | | -| factory.py | 100% | | -| simulator.py | 98% | Uncovered: `_add_ticker_internal` duplicate guard (L145), exception log in `_run_loop` (L264-265) | -| massive_client.py | 56% | Expected — real API methods can't run without the massive package | -| stream.py | 31% | Expected — SSE generator requires a running ASGI server to test | - ---- - -## 2. Architecture Assessment - -The market data subsystem is well-designed. It follows a clean strategy pattern: - -``` -MarketDataSource (ABC) -├── SimulatorDataSource (GBM simulator) -└── MassiveDataSource (Polygon.io REST poller) - │ - ▼ - PriceCache (shared, thread-safe) - │ - ▼ - SSE stream → Frontend -``` - -**Strengths:** -- Clear separation of concerns across 8 focused modules -- Factory pattern with lazy imports — the `massive` package is only needed when `MASSIVE_API_KEY` is set -- PriceCache as the single point of truth decouples producers from consumers -- Immutable `PriceUpdate` dataclass with `frozen=True, slots=True` is correct and efficient -- The GBM math is proper: log-normal price paths via `exp((mu - 0.5*sigma^2)*dt + sigma*sqrt(dt)*Z)` -- Correlated moves via Cholesky decomposition are a nice touch for realism -- All background tasks are properly cancellable and idempotent on stop() - ---- - -## 3. Issues Found - -### 3.1 Build Configuration Bug (Severity: High) - -`pyproject.toml` is missing the hatchling package discovery configuration. Running `uv sync` fails: - -``` -ValueError: Unable to determine which files to ship inside the wheel -``` - -**Fix:** Add to `pyproject.toml`: -```toml -[tool.hatch.build.targets.wheel] -packages = ["app"] -``` - -This will block Docker builds and any fresh `uv sync` until fixed. - -### 3.2 Massive Test Fragility (Severity: Medium) - -Five tests in `test_massive.py` fail when the `massive` package is not installed. The root cause is twofold: - -1. **`_poll_once` uses `asyncio.to_thread(self._fetch_snapshots)`** — even when `_fetch_snapshots` is patched on the instance, `to_thread` runs it in a thread executor. Three tests mock `_fetch_snapshots` as a `MagicMock` (synchronous), but `asyncio.to_thread` wraps it in `loop.run_in_executor`, which works... except that when `_fetch_snapshots` is NOT patched, the real method tries `from massive.rest.models import SnapshotMarketType` and fails. - -2. **`patch("app.market.massive_client.RESTClient")`** targets a name that doesn't exist at module level because `massive_client.py` uses a lazy import inside `start()`. The patch needs `create=True` or the import needs to be at module level behind a `TYPE_CHECKING` guard. - -These tests pass when `massive>=1.0.0` is installed (as `pyproject.toml` declares it as a core dependency), so this is technically a test-environment issue, not a code bug. However, since the whole point of lazy imports is to make `massive` optional for simulator-only use, the tests should also work without it. - -### 3.3 `_generate_events` Return Type Annotation (Severity: Low) - -`stream.py:54` declares the return type as `-> None` but the function is an async generator (it uses `yield`). The correct annotation would be `-> AsyncGenerator[str, None]` or simply removing the annotation. This doesn't cause runtime issues but is misleading for type checkers and developers. - -### 3.4 `version` Property Not Under Lock (Severity: Low) - -`PriceCache.version` reads `self._version` without acquiring `self._lock`: - -```python -@property -def version(self) -> int: - return self._version -``` - -On CPython with the GIL, reading a single `int` is atomic, so this won't cause corruption. However, it's inconsistent with the rest of the class, and if the project ever runs on a no-GIL Python build (PEP 703, Python 3.13t+), this could become a race. A minor concern given the current context. - -### 3.5 `SimulatorDataSource.get_tickers` Accesses Private State (Severity: Low) - -`simulator.py:254`: -```python -def get_tickers(self) -> list[str]: - return list(self._sim._tickers) if self._sim else [] -``` - -This reaches into `GBMSimulator._tickers` (private attribute). `GBMSimulator` should expose a `get_tickers()` method or a `tickers` property to keep the boundary clean. - -### 3.6 Module-Level Router Instance (Severity: Low) - -`stream.py:16` creates a module-level `router` object, and `create_stream_router()` registers a route on it via closure. If `create_stream_router` were called twice (e.g., in tests), the `/prices` route would be registered twice on the same router. In practice this won't happen because the function is called once during app startup, but it's a latent footgun for testing. - -### 3.7 Unused Imports in Tests (Severity: Trivial) - -Five lint warnings from `ruff`: -- `test_cache.py`: unused `pytest` -- `test_factory.py`: unused `pytest` -- `test_massive.py`: unused `asyncio` -- `test_simulator.py`: unused `math`, unused `pytest` - ---- - -## 4. Design Observations - -### 4.1 Things Done Well - -- **GBM parameter tuning is thoughtful.** TSLA at sigma=0.50 vs V at 0.17 reflects real-world volatility differences. The shock event system (~0.1% per tick, producing visible moves every ~50s) adds visual drama without destabilizing prices. -- **Cholesky decomposition for correlated moves** is the mathematically correct approach. The sector-based correlation structure (tech 0.6, finance 0.5, cross 0.3) is reasonable. -- **Defensive error handling in both data sources.** Both `_run_loop` (simulator) and `_poll_once`/`_poll_loop` (massive) catch exceptions and continue, which is essential for a long-running background service. -- **SSE implementation is clean.** The version-based change detection avoids sending redundant payloads. The `retry: 1000\n\n` directive ensures browser auto-reconnect. Nginx buffering is proactively disabled. -- **Seed prices in the cache at start** means the frontend gets data on the first SSE poll, with no visible delay. -- **Thread-safe cache with Lock** is the right choice since the Massive client runs API calls via `asyncio.to_thread`. - -### 4.2 Missing Tests - -- **SSE streaming (`stream.py`)** at 31% coverage has no dedicated tests. Testing SSE requires an ASGI test client (e.g., `httpx.AsyncClient` with `app`). Given that this is the primary consumer of PriceCache, even a basic integration test would add confidence. -- **No concurrent/thread-safety test for PriceCache.** The lock usage looks correct from inspection, but a test with multiple threads writing simultaneously would verify it empirically. -- **No test for `GBMSimulator` with all 10 default tickers.** Tests use 1-2 tickers. A test confirming the Cholesky decomposition succeeds for the full 10-ticker default set would catch correlation matrix issues. - -### 4.3 Potential Future Considerations - -- The `PriceCache` doesn't cap history; it only stores the latest price per ticker, so memory is bounded at O(tickers). Good. -- The `DEFAULT_CORR` constant (0.3, `seed_prices.py:48`) is defined but never referenced in `_pairwise_correlation`. The static method returns `CROSS_GROUP_CORR` (also 0.3) as the fallback. This is semantically confusing — `DEFAULT_CORR` seems intended for tickers not in any group, but the code returns `CROSS_GROUP_CORR` for all non-matched pairs. Both happen to be 0.3, so behavior is correct, but the naming is misleading. - ---- - -## 5. Verdict - -The market data backend is solid and well-structured. The GBM simulator, price cache, abstract interface, factory pattern, and SSE streaming all work correctly and follow good practices. The architecture will integrate cleanly with the rest of the application. - -**Must fix before proceeding:** -1. Add `[tool.hatch.build.targets.wheel] packages = ["app"]` to `pyproject.toml` — without this, `uv sync` and Docker builds fail. - -**Should fix:** -2. Make the Massive tests resilient to the `massive` package being absent (use `create=True` on patches, or restructure mocks). -3. Fix the `_generate_events` return type annotation. -4. Remove unused imports in test files. - -**Nice to have:** -5. Add a `get_tickers()` public method to `GBMSimulator`. -6. Add at least one SSE integration test. -7. Clarify `DEFAULT_CORR` vs `CROSS_GROUP_CORR` naming. diff --git a/planning/archive/MARKET_INTERFACE.md b/planning/archive/MARKET_INTERFACE.md deleted file mode 100644 index 156cad287..000000000 --- a/planning/archive/MARKET_INTERFACE.md +++ /dev/null @@ -1,273 +0,0 @@ -# Market Data Interface Design - -Unified Python interface for market data in FinAlly. Two implementations (simulator and Massive API) behind one abstract interface. All downstream code — SSE streaming, price cache, portfolio valuation — is source-agnostic. - -## Core Data Model - -```python -from dataclasses import dataclass - -@dataclass -class PriceUpdate: - """A single price update for one ticker.""" - ticker: str - price: float - previous_price: float - timestamp: float # Unix seconds - change: float # price - previous_price - direction: str # "up", "down", or "flat" -``` - -This is the only data structure that leaves the market data layer. Everything downstream works with `PriceUpdate` objects. - -## Abstract Interface - -```python -from abc import ABC, abstractmethod - -class MarketDataSource(ABC): - """Abstract interface for market data providers.""" - - @abstractmethod - async def start(self, tickers: list[str]) -> None: - """Begin producing price updates for the given tickers.""" - - @abstractmethod - async def stop(self) -> None: - """Stop producing price updates and clean up.""" - - @abstractmethod - async def add_ticker(self, ticker: str) -> None: - """Add a ticker to the active set.""" - - @abstractmethod - async def remove_ticker(self, ticker: str) -> None: - """Remove a ticker from the active set.""" - - @abstractmethod - def get_tickers(self) -> list[str]: - """Return the current list of active tickers.""" -``` - -Both implementations write to a shared `PriceCache` (see below). The interface does **not** return prices directly — it pushes updates into the cache on its own schedule. - -## Price Cache - -Shared in-memory store that both data sources write to and the SSE streamer reads from. - -```python -import time -from threading import Lock - -class PriceCache: - """Thread-safe cache of latest prices per ticker.""" - - def __init__(self): - self._prices: dict[str, PriceUpdate] = {} - self._lock = Lock() - - def update(self, ticker: str, price: float, timestamp: float | None = None) -> PriceUpdate: - """Update price for a ticker. Returns the PriceUpdate.""" - with self._lock: - ts = timestamp or time.time() - previous = self._prices.get(ticker) - previous_price = previous.price if previous else price - - if price > previous_price: - direction = "up" - elif price < previous_price: - direction = "down" - else: - direction = "flat" - - update = PriceUpdate( - ticker=ticker, - price=price, - previous_price=previous_price, - timestamp=ts, - change=price - previous_price, - direction=direction, - ) - self._prices[ticker] = update - return update - - def get(self, ticker: str) -> PriceUpdate | None: - """Get latest price for a ticker.""" - with self._lock: - return self._prices.get(ticker) - - def get_all(self) -> dict[str, PriceUpdate]: - """Get all current prices.""" - with self._lock: - return dict(self._prices) - - def remove(self, ticker: str) -> None: - """Remove a ticker from the cache.""" - with self._lock: - self._prices.pop(ticker, None) -``` - -## Factory Function - -Select the data source at startup based on environment: - -```python -import os - -def create_market_data_source(price_cache: PriceCache) -> MarketDataSource: - """Create the appropriate market data source based on environment.""" - api_key = os.environ.get("MASSIVE_API_KEY", "").strip() - - if api_key: - from .massive_client import MassiveDataSource - return MassiveDataSource(api_key=api_key, price_cache=price_cache) - else: - from .simulator import SimulatorDataSource - return SimulatorDataSource(price_cache=price_cache) -``` - -## Massive Implementation Sketch - -```python -import asyncio -from massive import RESTClient -from massive.rest.models import SnapshotMarketType - -class MassiveDataSource(MarketDataSource): - def __init__(self, api_key: str, price_cache: PriceCache, poll_interval: float = 15.0): - self._client = RESTClient(api_key=api_key) - self._cache = price_cache - self._interval = poll_interval - self._tickers: list[str] = [] - self._task: asyncio.Task | None = None - - async def start(self, tickers: list[str]) -> None: - self._tickers = list(tickers) - self._task = asyncio.create_task(self._poll_loop()) - - async def stop(self) -> None: - if self._task: - self._task.cancel() - - async def add_ticker(self, ticker: str) -> None: - if ticker not in self._tickers: - self._tickers.append(ticker) - - async def remove_ticker(self, ticker: str) -> None: - self._tickers = [t for t in self._tickers if t != ticker] - self._cache.remove(ticker) - - def get_tickers(self) -> list[str]: - return list(self._tickers) - - async def _poll_loop(self) -> None: - while True: - await self._poll_once() - await asyncio.sleep(self._interval) - - async def _poll_once(self) -> None: - if not self._tickers: - return - # Run synchronous Massive client in thread pool - snapshots = await asyncio.to_thread( - self._client.get_snapshot_all, - market_type=SnapshotMarketType.STOCKS, - tickers=self._tickers, - ) - for snap in snapshots: - self._cache.update( - ticker=snap.ticker, - price=snap.last_trade.price, - timestamp=snap.last_trade.timestamp / 1000, # ms -> seconds - ) -``` - -## Simulator Implementation Sketch - -```python -import asyncio - -class SimulatorDataSource(MarketDataSource): - def __init__(self, price_cache: PriceCache, update_interval: float = 0.5): - self._cache = price_cache - self._interval = update_interval - self._tickers: list[str] = [] - self._task: asyncio.Task | None = None - self._sim: GBMSimulator | None = None # See MARKET_SIMULATOR.md - - async def start(self, tickers: list[str]) -> None: - self._tickers = list(tickers) - self._sim = GBMSimulator(tickers=self._tickers) - self._task = asyncio.create_task(self._run_loop()) - - async def stop(self) -> None: - if self._task: - self._task.cancel() - - async def add_ticker(self, ticker: str) -> None: - if ticker not in self._tickers: - self._tickers.append(ticker) - self._sim.add_ticker(ticker) - - async def remove_ticker(self, ticker: str) -> None: - self._tickers = [t for t in self._tickers if t != ticker] - self._sim.remove_ticker(ticker) - self._cache.remove(ticker) - - def get_tickers(self) -> list[str]: - return list(self._tickers) - - async def _run_loop(self) -> None: - while True: - prices = self._sim.step() # Returns dict[str, float] - for ticker, price in prices.items(): - self._cache.update(ticker=ticker, price=price) - await asyncio.sleep(self._interval) -``` - -## Integration with SSE - -The SSE endpoint reads from the `PriceCache` and pushes to connected clients: - -```python -async def price_stream(price_cache: PriceCache): - """SSE generator that yields price updates.""" - while True: - prices = price_cache.get_all() - data = { - ticker: { - "ticker": p.ticker, - "price": p.price, - "previous_price": p.previous_price, - "change": p.change, - "direction": p.direction, - "timestamp": p.timestamp, - } - for ticker, p in prices.items() - } - yield f"data: {json.dumps(data)}\n\n" - await asyncio.sleep(0.5) -``` - -## File Structure - -``` -backend/ - app/ - market/ - __init__.py - models.py # PriceUpdate dataclass - interface.py # MarketDataSource ABC, PriceCache - factory.py # create_market_data_source() - massive_client.py # MassiveDataSource - simulator.py # SimulatorDataSource + GBMSimulator - seed_prices.py # Default ticker seed prices -``` - -## Lifecycle - -1. **App startup**: Create `PriceCache`, call `create_market_data_source(price_cache)`, then `await source.start(initial_tickers)` -2. **Watchlist changes**: Call `source.add_ticker()` or `source.remove_ticker()` -3. **SSE streaming**: Reads from `PriceCache.get_all()` every 500ms -4. **Trade execution**: Reads current price from `PriceCache.get(ticker)` -5. **App shutdown**: Call `await source.stop()` diff --git a/planning/archive/MARKET_SIMULATOR.md b/planning/archive/MARKET_SIMULATOR.md deleted file mode 100644 index e157b6efb..000000000 --- a/planning/archive/MARKET_SIMULATOR.md +++ /dev/null @@ -1,245 +0,0 @@ -# Market Simulator Design - -Approach and code structure for simulating realistic stock prices when no Massive API key is configured. - -## Overview - -The simulator uses **Geometric Brownian Motion (GBM)** to generate realistic stock price paths. GBM is the standard model underlying Black-Scholes option pricing — prices evolve continuously with random noise, can't go negative, and exhibit the lognormal distribution seen in real markets. - -Updates run at ~500ms intervals, producing a continuous stream of price changes that feel alive. - -## GBM Math - -At each time step, a stock price evolves as: - -``` -S(t+dt) = S(t) * exp((mu - sigma^2/2) * dt + sigma * sqrt(dt) * Z) -``` - -Where: -- `S(t)` = current price -- `mu` = annualized drift (expected return), e.g. 0.05 (5%) -- `sigma` = annualized volatility, e.g. 0.20 (20%) -- `dt` = time step as fraction of a trading year -- `Z` = standard normal random variable (drawn from N(0,1)) - -For our 500ms updates with ~252 trading days and ~6.5 hours per day: -``` -dt = 0.5 / (252 * 6.5 * 3600) = ~8.5e-8 -``` - -This tiny `dt` produces small, realistic per-tick moves. - -## Correlated Moves - -Real stocks don't move independently — tech stocks tend to move together, etc. We use a **Cholesky decomposition** of a correlation matrix to generate correlated random draws. - -Given a correlation matrix `C`, compute `L = cholesky(C)`. Then for independent standard normals `Z_independent`: -``` -Z_correlated = L @ Z_independent -``` - -Default correlation groups: -- **Tech**: AAPL, GOOGL, MSFT, AMZN, META, NVDA, NFLX — corr ~0.6 within group -- **Finance**: JPM, V — corr ~0.5 within group -- **Cross-group**: ~0.3 baseline correlation -- **TSLA**: lower correlation with everything (~0.3) — it does its own thing - -## Random Events - -Every step, each ticker has a small probability (~0.001) of a random event — a sudden 2-5% move. This adds drama and makes the dashboard visually interesting. - -```python -if random.random() < event_probability: - shock = random.uniform(0.02, 0.05) * random.choice([-1, 1]) - price *= (1 + shock) -``` - -## Seed Prices - -Realistic starting prices for the default watchlist: - -```python -SEED_PRICES: dict[str, float] = { - "AAPL": 190.0, - "GOOGL": 175.0, - "MSFT": 420.0, - "AMZN": 185.0, - "TSLA": 250.0, - "NVDA": 800.0, - "META": 500.0, - "JPM": 195.0, - "V": 280.0, - "NFLX": 600.0, -} -``` - -Tickers added dynamically (not in the seed list) start at a random price between $50-$300. - -## Per-Ticker Parameters - -Each ticker has its own volatility to reflect real-world behavior: - -```python -TICKER_PARAMS: dict[str, dict] = { - "AAPL": {"sigma": 0.22, "mu": 0.05}, - "GOOGL": {"sigma": 0.25, "mu": 0.05}, - "MSFT": {"sigma": 0.20, "mu": 0.05}, - "AMZN": {"sigma": 0.28, "mu": 0.05}, - "TSLA": {"sigma": 0.50, "mu": 0.03}, # High vol - "NVDA": {"sigma": 0.40, "mu": 0.08}, # High vol, strong drift - "META": {"sigma": 0.30, "mu": 0.05}, - "JPM": {"sigma": 0.18, "mu": 0.04}, # Low vol (bank) - "V": {"sigma": 0.17, "mu": 0.04}, # Low vol (payments) - "NFLX": {"sigma": 0.35, "mu": 0.05}, -} - -# Default for unknown tickers -DEFAULT_PARAMS = {"sigma": 0.25, "mu": 0.05} -``` - -## Implementation - -```python -import math -import random -import time -import numpy as np - -class GBMSimulator: - """Generates correlated GBM price paths for multiple tickers.""" - - def __init__( - self, - tickers: list[str], - dt: float = 8.5e-8, - event_probability: float = 0.001, - ): - self._dt = dt - self._event_prob = event_probability - self._prices: dict[str, float] = {} - self._params: dict[str, dict] = {} - self._tickers: list[str] = [] - self._cholesky: np.ndarray | None = None - - for ticker in tickers: - self.add_ticker(ticker) - - def add_ticker(self, ticker: str) -> None: - if ticker in self._prices: - return - self._tickers.append(ticker) - self._prices[ticker] = SEED_PRICES.get(ticker, random.uniform(50, 300)) - self._params[ticker] = TICKER_PARAMS.get(ticker, DEFAULT_PARAMS) - self._rebuild_cholesky() - - def remove_ticker(self, ticker: str) -> None: - if ticker not in self._prices: - return - self._tickers.remove(ticker) - del self._prices[ticker] - del self._params[ticker] - self._rebuild_cholesky() - - def step(self) -> dict[str, float]: - """Advance one time step. Returns {ticker: new_price}.""" - n = len(self._tickers) - if n == 0: - return {} - - # Generate correlated random normals - z_independent = np.random.standard_normal(n) - if self._cholesky is not None: - z = self._cholesky @ z_independent - else: - z = z_independent - - result = {} - for i, ticker in enumerate(self._tickers): - params = self._params[ticker] - mu = params["mu"] - sigma = params["sigma"] - - # GBM step - drift = (mu - 0.5 * sigma**2) * self._dt - diffusion = sigma * math.sqrt(self._dt) * z[i] - self._prices[ticker] *= math.exp(drift + diffusion) - - # Random event - if random.random() < self._event_prob: - shock = random.uniform(0.02, 0.05) * random.choice([-1, 1]) - self._prices[ticker] *= (1 + shock) - - result[ticker] = round(self._prices[ticker], 2) - - return result - - def get_price(self, ticker: str) -> float | None: - return self._prices.get(ticker) - - def _rebuild_cholesky(self) -> None: - """Rebuild the Cholesky decomposition of the correlation matrix.""" - n = len(self._tickers) - if n <= 1: - self._cholesky = None - return - - corr = np.eye(n) - for i in range(n): - for j in range(i + 1, n): - rho = self._get_correlation(self._tickers[i], self._tickers[j]) - corr[i, j] = rho - corr[j, i] = rho - - self._cholesky = np.linalg.cholesky(corr) - - def _get_correlation(self, t1: str, t2: str) -> float: - """Return pairwise correlation between two tickers.""" - tech = {"AAPL", "GOOGL", "MSFT", "AMZN", "META", "NVDA", "NFLX"} - finance = {"JPM", "V"} - - t1_tech = t1 in tech - t2_tech = t2 in tech - t1_fin = t1 in finance - t2_fin = t2 in finance - - # Same sector: higher correlation - if t1_tech and t2_tech: - return 0.6 - if t1_fin and t2_fin: - return 0.5 - - # TSLA is a loner - if t1 == "TSLA" or t2 == "TSLA": - return 0.3 - - # Cross-sector or unknown - if (t1_tech and t2_fin) or (t1_fin and t2_tech): - return 0.3 - - # Default - return 0.3 -``` - -## File Structure - -All simulator code lives in a single module: - -``` -backend/ - app/ - market/ - simulator.py # GBMSimulator class + seed data + SimulatorDataSource - seed_prices.py # SEED_PRICES, TICKER_PARAMS, DEFAULT_PARAMS (constants) -``` - -`seed_prices.py` contains just the constant dictionaries. `simulator.py` contains the `GBMSimulator` class and the `SimulatorDataSource` (the `MarketDataSource` implementation that wraps `GBMSimulator` in an async loop). - -## Behavior Notes - -- Prices never go negative (GBM is multiplicative — `exp()` is always positive) -- The tiny `dt` produces sub-cent moves per tick, which accumulate naturally over time -- With `sigma=0.50` (TSLA), a day of simulated trading produces roughly the right intraday range -- The correlation matrix must be positive semi-definite — Cholesky decomposition guarantees this for valid correlation matrices -- Random events happen ~0.1% of steps = roughly once every 500 seconds per ticker. With 10 tickers, expect an event somewhere roughly every 50 seconds — enough to keep it interesting -- When a new ticker is added mid-session, the Cholesky matrix is rebuilt. This is O(n^2) but n is small (<50 tickers) diff --git a/planning/archive/MASSIVE_API.md b/planning/archive/MASSIVE_API.md deleted file mode 100644 index 3266bc64f..000000000 --- a/planning/archive/MASSIVE_API.md +++ /dev/null @@ -1,251 +0,0 @@ -# Massive API Reference (formerly Polygon.io) - -Reference documentation for the Massive (formerly Polygon.io) REST API as used in FinAlly. - -## Overview - -- **Base URL**: `https://api.massive.com` (legacy `https://api.polygon.io` still supported) -- **Python package**: `massive` (install via `pip install -U massive` / `uv add massive`) -- **Min Python version**: 3.9+ -- **Auth**: API key via `MASSIVE_API_KEY` env var or passed to `RESTClient(api_key=...)` -- **Auth header**: `Authorization: Bearer ` (the client handles this automatically) - -## Rate Limits - -| Tier | Limit | -|------|-------| -| Free | 5 requests/minute | -| Paid (all tiers) | Unlimited (recommended: stay under 100 req/s) | - -For FinAlly, we poll on a timer. Free tier: poll every 15s. Paid: poll every 2-5s. - -## Client Initialization - -```python -from massive import RESTClient - -# Reads MASSIVE_API_KEY from environment automatically -client = RESTClient() - -# Or pass explicitly -client = RESTClient(api_key="your_key_here") -``` - -## Endpoints Used in FinAlly - -### 1. Snapshot — All Tickers (Primary Endpoint) - -Gets current prices for multiple tickers in a **single API call**. This is the main endpoint we use for polling. - -**REST**: `GET /v2/snapshot/locale/us/markets/stocks/tickers?tickers=AAPL,GOOGL,MSFT` - -**Python client**: -```python -from massive import RESTClient -from massive.rest.models import SnapshotMarketType - -client = RESTClient() - -# Get snapshots for specific tickers (one API call) -snapshots = client.get_snapshot_all( - market_type=SnapshotMarketType.STOCKS, - tickers=["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"], -) - -for snap in snapshots: - print(f"{snap.ticker}: ${snap.last_trade.price}") - print(f" Day change: {snap.day.change_percent}%") - print(f" Day OHLC: O={snap.day.open} H={snap.day.high} L={snap.day.low} C={snap.day.close}") - print(f" Volume: {snap.day.volume}") -``` - -**Response structure** (per ticker): -```json -{ - "ticker": "AAPL", - "day": { - "open": 129.61, - "high": 130.15, - "low": 125.07, - "close": 125.07, - "volume": 111237700, - "volume_weighted_average_price": 127.35, - "previous_close": 129.61, - "change": -4.54, - "change_percent": -3.50 - }, - "last_trade": { - "price": 125.07, - "size": 100, - "exchange": "XNYS", - "timestamp": 1675190399000 - }, - "last_quote": { - "bid_price": 125.06, - "ask_price": 125.08, - "bid_size": 500, - "ask_size": 1000, - "spread": 0.02, - "timestamp": 1675190399500 - }, - "prev_daily_bar": { "...": "previous day OHLCV" }, - "minute_volume": { "...": "volume per minute" } -} -``` - -**Key fields we extract**: -- `last_trade.price` — current price for trading and display -- `day.previous_close` — for calculating day change -- `day.change_percent` — day change percentage -- `last_trade.timestamp` — when the price was recorded - -### 2. Single Ticker Snapshot - -For getting detailed data on one ticker (e.g., when user clicks a ticker for the detail view). - -**Python client**: -```python -snapshot = client.get_snapshot_ticker( - market_type=SnapshotMarketType.STOCKS, - ticker="AAPL", -) - -print(f"Price: ${snapshot.last_trade.price}") -print(f"Bid/Ask: ${snapshot.last_quote.bid_price} / ${snapshot.last_quote.ask_price}") -print(f"Day range: ${snapshot.day.low} - ${snapshot.day.high}") -``` - -### 3. Previous Close - -Gets the previous day's OHLC for a ticker. Useful for seed prices. - -**REST**: `GET /v2/aggs/ticker/{ticker}/prev` - -**Python client**: -```python -prev = client.get_previous_close_agg(ticker="AAPL") - -for agg in prev: - print(f"Previous close: ${agg.close}") - print(f"OHLC: O={agg.open} H={agg.high} L={agg.low} C={agg.close}") - print(f"Volume: {agg.volume}") -``` - -**Response**: -```json -{ - "ticker": "AAPL", - "results": [ - { - "o": 150.0, - "h": 155.0, - "l": 149.0, - "c": 154.5, - "v": 1000000, - "t": 1672531200000 - } - ] -} -``` - -### 4. Aggregates (Bars) - -Historical OHLCV bars over a date range. Not needed for live polling but useful if we add historical charts. - -**REST**: `GET /v2/aggs/ticker/{ticker}/range/{multiplier}/{timespan}/{from}/{to}` - -**Python client**: -```python -aggs = [] -for a in client.list_aggs( - ticker="AAPL", - multiplier=1, - timespan="day", - from_="2024-01-01", - to="2024-01-31", - limit=50000, -): - aggs.append(a) - -for a in aggs: - print(f"Date: {a.timestamp}, O={a.open} H={a.high} L={a.low} C={a.close} V={a.volume}") -``` - -**Response** (each bar): -```json -{ - "o": 130.0, - "h": 132.5, - "l": 129.8, - "c": 131.2, - "v": 50000000, - "t": 1672531200000 -} -``` - -### 5. Last Trade / Last Quote - -Individual endpoints for the most recent trade or NBBO quote. - -```python -# Last trade -trade = client.get_last_trade(ticker="AAPL") -print(f"Last trade: ${trade.price} x {trade.size}") - -# Last NBBO quote -quote = client.get_last_quote(ticker="AAPL") -print(f"Bid: ${quote.bid} x {quote.bid_size}") -print(f"Ask: ${quote.ask} x {quote.ask_size}") -``` - -## How FinAlly Uses the API - -The Massive poller runs as a background task: - -1. Collects all tickers from the watchlist -2. Calls `get_snapshot_all()` with those tickers (one API call) -3. Extracts `last_trade.price` and `day.previous_close` from each snapshot -4. Writes to the shared in-memory price cache -5. Sleeps for the poll interval, then repeats - -```python -import asyncio -from massive import RESTClient -from massive.rest.models import SnapshotMarketType - -async def poll_massive(api_key: str, get_tickers, price_cache, interval: float = 15.0): - """Poll Massive API and update the price cache.""" - client = RESTClient(api_key=api_key) - - while True: - tickers = get_tickers() - if tickers: - snapshots = client.get_snapshot_all( - market_type=SnapshotMarketType.STOCKS, - tickers=tickers, - ) - for snap in snapshots: - price_cache.update( - ticker=snap.ticker, - price=snap.last_trade.price, - previous_close=snap.day.previous_close, - timestamp=snap.last_trade.timestamp, - ) - - await asyncio.sleep(interval) -``` - -## Error Handling - -The client raises exceptions for HTTP errors: -- **401**: Invalid API key -- **403**: Insufficient permissions (plan doesn't include the endpoint) -- **429**: Rate limit exceeded (free tier: 5 req/min) -- **5xx**: Server errors (client has built-in retry with 3 retries by default) - -## Notes - -- The snapshot endpoint returns data for **all requested tickers in one call** — this is critical for staying within rate limits on the free tier -- Timestamps from the API are Unix milliseconds -- During market closed hours, `last_trade.price` reflects the last traded price (may include after-hours) -- The `day` object resets at market open; during pre-market, values may be from the previous session