Build Complete Market Data Backend

[luiv-engineer]

Description

Build a complete market data subsystem for FinAlly supporting both a GBM simulator (default) and Massive API (real market data) via a unified interface.

Status: Architecture complete. Production-ready implementation required.

---

Components to Build

1. Massive API Interface (backend/app/market/massive_client.py)

Implement a REST client for the Massive (Polygon.io) API that:

• Polls the snapshot endpoint for all watched tickers in a single API call
• Handles rate limits (free tier: 5 req/min → poll every 15s; paid tiers: faster)
• Runs synchronously in asyncio.to_thread() to avoid blocking the event loop
• Implements the MarketDataSource abstract interface
• Includes error handling for invalid API keys, rate limits, and malformed responses
• Gracefully recovers from failures without stopping the poller

Key methods:

• start(tickers) - Begin polling with initial tickers
• stop() - Stop polling and clean up
• add_ticker(ticker) - Add ticker to active set
• remove_ticker(ticker) - Remove ticker from active set
• get_tickers() - Return current list of tracked tickers

Configuration:

• API key via MASSIVE_API_KEY environment variable
• Poll interval: 15s for free tier (configurable for paid tiers)

2. Unified Market Interface (backend/app/market/)

Implement a complete abstraction layer with:

Data Model (models.py):

• PriceUpdate - Immutable frozen dataclass with ticker, price, previous_price, timestamp, change, direction
• Computed properties for change percentage and direction
• Serialization method for SSE transmission

Abstract Interface (interface.py):

• MarketDataSource ABC defining start/stop/add_ticker/remove_ticker/get_tickers
• Enables both simulator and Massive implementations to be source-agnostic

Price Cache (cache.py):

• Thread-safe in-memory store for latest prices per ticker
• Version counter for efficient SSE change detection
• Methods: update, get, get_all, get_price, remove

Factory (factory.py):

• create_market_data_source() - Selects simulator or Massive based on MASSIVE_API_KEY env var
• Lazy imports to avoid unnecessary dependencies

SSE Streaming (stream.py):

• create_stream_router() - FastAPI router factory for /api/stream/prices endpoint
• Version-based change detection to avoid redundant sends
• Auto-reconnection support via EventSource retry directive

Package Export (__init__.py):

• Clean public API: PriceUpdate, PriceCache, MarketDataSource, create_market_data_source, create_stream_router

3. Market Data Simulator (backend/app/market/simulator.py)

Implement a Geometric Brownian Motion (GBM) simulator with:

Seed Prices & Parameters (seed_prices.py):

• Realistic starting prices for 10 default tickers (AAPL, GOOGL, MSFT, AMZN, TSLA, NVDA, META, JPM, V, NFLX)
• Per-ticker GBM drift (mu) and volatility (sigma) parameters
• Sector-based correlation groups (tech, finance)
• Correlation coefficients for intra-group and cross-group moves

GBM Simulator Engine (simulator.py):

• Geometric Brownian Motion: S(t+dt) = S(t) * exp((mu - sigma²/2)*dt + sigma*sqrt(dt)*Z)
• Cholesky decomposition of correlation matrix for realistic correlated moves
• Random shock events (~0.1% per tick) for 2-5% sudden moves (visual drama)
• Time step: 500ms ticks; 252 trading days/year normalization
• Supports dynamic ticker add/remove with automatic correlation recalculation

SimulatorDataSource Implementation:

• Implements MarketDataSource interface
• Background asyncio loop stepping the simulator every 500ms
• Seeds cache with initial prices on startup
• Exception resilience (one bad tick doesn't kill the feed)
• Graceful cancellation and cleanup

---

Architecture

PriceCache (Thread-safe In-Memory Store)
    ↑                    ↑                    ↓
  Writes              Writes              Reads
    │                    │                    │
┌─────────────┐  ┌────────────────┐  ┌──────────────┐
│ Simulator   │  │ Massive Client │  │ SSE Endpoint │
│ (GBM)       │  │ (REST Poller)  │  │ (Generator)  │
└─────────────┘  └────────────────┘  └──────────────┘

Both data sources implement MarketDataSource ABC and write to a shared PriceCache. The SSE endpoint reads from the cache and streams to connected clients.

---

Environment Variables

Variable	Default	Description
MASSIVE_API_KEY	"" (empty)	If set, use Massive; otherwise use simulator

---

Testing Requirements

Unit Tests (73 tests target, 84% coverage):

• test_models.py - PriceUpdate dataclass and properties
• test_cache.py - PriceCache thread safety and version counter
• test_simulator.py - GBM math, price positivity, correlation matrix
• test_simulator_source.py - SimulatorDataSource lifecycle and ticker management
• test_factory.py - Environment-based source selection
• test_massive.py - Massive API client with mocked responses

Integration Tests:

• SSE endpoint with mock client
• Cholesky decomposition for all 10 default tickers
• Error cases (invalid API key, malformed responses, network failures)
• Ticker add/remove during runtime

Demo (Optional):

• backend/market_data_demo.py - Rich terminal dashboard showing live prices, sparklines, and event log

---

Deliverables

• Complete backend/app/market/ module structure (8 files, ~500 lines)
• 73 unit and integration tests with 84%+ coverage
• Dependencies added to backend/pyproject.toml (numpy, massive)
• .env.example updated with MASSIVE_API_KEY
• FastAPI integration in backend/app/main.py (lifespan context manager)
• All edge cases handled (empty watchlist, price cache misses, API failures)
• Code review passed and all issues resolved

---

Related Documentation

• planning/PLAN.md - Full project specification
• planning/MARKET_DATA_DESIGN.md - Complete implementation guide with code examples
• planning/archive/MASSIVE_API.md - Massive API reference
• planning/archive/MARKET_INTERFACE.md - Interface design patterns