trading/.planning/codebase/ARCHITECTURE.md

Architecture

Analysis Date: 2025-02-23

Pattern Overview

Overall: Event-driven microservices architecture using Redis Streams for inter-service communication.

Key Characteristics:

• 8 independent Python services communicating asynchronously via Redis Streams
• Layered abstraction for extensibility: broker abstraction (BaseBroker), strategy abstraction (BaseStrategy), source abstraction for news
• Each service consumes from one or more Redis Streams, processes data, persists to PostgreSQL, and publishes downstream results
• Shared libraries provide domain models, database layer, telemetry, and abstractions
• React TypeScript frontend connected via WebSocket and REST API

Layers

Service Layer (Event Processors):

• Purpose: Independent microservices that transform data across the trading pipeline
• Location: services/*/main.py (entry points)
• Contains: RSS/Reddit fetchers, sentiment analysis, signal generation, trade execution, learning engine, market data subscriptions, API gateway
• Depends on: shared/ libraries, external APIs (Alpaca, Reddit, Ollama)
• Used by: Each other (via Redis Streams), dashboard (via API gateway)

Infrastructure/Abstraction Layer:

• Purpose: Pluggable interfaces and database access
• Location: shared/broker/, shared/strategies/, shared/db.py
• Contains: BaseBroker (Alpaca implementation), BaseStrategy (Momentum/MeanReversion/NewsDriven), database engine setup
• Depends on: SQLAlchemy, alpaca-py, external ML models
• Used by: Trade executor, signal generator, backtester

Data Layer:

• Purpose: Persistent storage of trades, signals, articles, market data, user credentials
• Location: shared/models/ (SQLAlchemy ORM models)
• Contains: 14 tables including trades, signals, articles, article_sentiments, market_data, positions, users, strategies, strategy weights history
• Depends on: PostgreSQL + TimescaleDB extension
• Used by: All services for read/write persistence

Messaging Layer:

• Purpose: Asynchronous event streaming between services
• Location: shared/redis_streams.py
• Contains: StreamPublisher (XADD), StreamConsumer (XREADGROUP with consumer groups)
• Depends on: Redis Streams
• Used by: All services for pub/sub communication

Frontend Layer:

• Purpose: User-facing dashboard for monitoring and control
• Location: dashboard/src/
• Contains: React pages, components, API client, WebAuthn auth
• Depends on: TanStack Query, TradingView charts, Recharts, React Router
• Used by: End users for real-time monitoring

Utility Layer:

• Purpose: Cross-cutting concerns
• Location: shared/config.py, shared/telemetry.py, shared/schemas/
• Contains: Configuration management (Pydantic BaseSettings), OpenTelemetry/Prometheus metrics setup, message schemas (Pydantic v2)
• Depends on: pydantic-settings, opentelemetry-sdk, prometheus-client
• Used by: All services

Data Flow

News Ingestion Pipeline:

1. services/news_fetcher/main.py polls RSS feeds and Reddit on independent schedules
2. Uses RSSSource and RedditSource to fetch articles
3. Deduplicates via Redis SET (news:seen_hashes)
4. Publishes RawArticle messages to news:raw Redis Stream

Sentiment Analysis:

1. services/sentiment_analyzer/main.py consumes from news:raw
2. Runs articles through FinBERTAnalyzer for sentiment scores
3. Falls back to OllamaAnalyzer if confidence < threshold (configured via SentimentAnalyzerConfig.finbert_confidence_threshold)
4. Extracts ticker mentions via regex in ticker_extractor.py
5. Persists Article + ArticleSentiment to database
6. Publishes ScoredArticle messages (one per ticker per article) to news:scored Redis Stream

Market Data Collection:

1. services/market_data/main.py fetches OHLCV bars from Alpaca (historical + live)
2. Publishes bars to market:bars Redis Stream as ticker-specific snapshots
3. Consumed by signal generator for technical indicators

Signal Generation:

1. services/signal_generator/main.py consumes from both news:scored and market:bars
2. Maintains in-memory buffers: per-ticker sentiment context (deque of last 50 scores) and market data (SMA, RSI)
3. Runs WeightedEnsemble combining three strategies:
   • MomentumStrategy: SMA cross-over based
   • MeanReversionStrategy: RSI-based
   • NewsDrivenStrategy: Sentiment-aware
4. Weights loaded from Redis cache (default equal weights: 0.333 each)
5. Only emits TradeSignal when combined strength > threshold (default 0.3)
6. Publishes to signals:generated Redis Stream, persists Signal model to database

Trade Execution:

1. services/trade_executor/main.py consumes from signals:generated
2. RiskManager performs pre-trade checks:
   • Market hours validation (9:30 AM - 4:00 PM ET)
   • Max daily loss limit
   • Per-ticker position limits
   • Cooldown period after exits
3. Calculates position size based on account equity and risk parameters
4. Submits order via AlpacaBroker (abstraction over alpaca-py)
5. Records Trade model with order details
6. Publishes TradeExecution to trades:executed Redis Stream

Learning & Weight Adjustment:

1. services/learning_engine/main.py consumes from trades:executed
2. Identifies position closes by tracking opening trades in Redis hash (positions:history)
3. TradeEvaluator calculates realized P&L
4. WeightAdjuster uses multi-armed bandit approach:
   • Attributes P&L credit to contributing strategies
   • Adjusts weights (with guardrails: min trades threshold, max shift limit, weight floor)
   • Persists adjustments to LearningAdjustment model for auditability
5. Saves updated weights to Redis cache

Portfolio Sync (Background Task):

1. services/api_gateway/tasks/portfolio_sync.py runs as background coroutine
2. Periodically queries Alpaca for current positions, cash, equity
3. Stores PortfolioSnapshot in database for historical tracking
4. Dashboard polls portfolio endpoint for real-time data

API Gateway & Dashboard:

1. services/api_gateway/main.py FastAPI app with:
   • REST endpoints (/api/signals, /api/trades, /api/portfolio, /api/news, /api/strategies, /api/backtest, /api/controls)
   • WebSocket endpoint for real-time updates
   • WebAuthn/JWT authentication
2. Dashboard (dashboard/src/App.tsx) consumed by authenticated users
3. Real-time market data, trades, signals streamed via WebSocket

State Management:

• Redis Streams: Messages flow unidirectionally; consumer groups ensure at-least-once processing
• Redis Cache: Strategy weights, position history, deduplication hashes
• PostgreSQL: Long-term storage of trades, signals, articles, market snapshots, users
• In-Memory: Services maintain per-ticker buffers (sentiment scores, market data) for current processing cycle; cleared/rebuilt as data arrives
• JWT Sessions: Issued at login, renewed via refresh token endpoint

Key Abstractions

BaseBroker:

• Purpose: Swap brokerage providers without changing services
• Examples: shared/broker/base.py (ABC), shared/broker/alpaca_broker.py (implementation)
• Pattern: Abstract methods for submit_order(), get_account(), get_positions(), cancel_order()
• Trade executor only knows BaseBroker interface; AlpacaBroker is a concrete implementation

BaseStrategy:

• Purpose: Pluggable trading strategy implementations
• Examples: shared/strategies/base.py (ABC), momentum.py, mean_reversion.py, news_driven.py
• Pattern: Each strategy implements async evaluate(ticker, market, sentiment) returning TradeSignal | None
• Signal generator instantiates all three at startup, runs via WeightedEnsemble.evaluate()

StreamPublisher / StreamConsumer:

• Purpose: Simplified Redis Streams wrapper with JSON serialization
• Examples: shared/redis_streams.py
• Pattern: Publisher.publish(dict) → XADD, Consumer.consume() → async iterator yielding (msg_id, data)
• Consumer groups ensure each message is processed once per group

Pydantic Schemas:

• Purpose: Message validation and serialization
• Examples: shared/schemas/news.py, shared/schemas/trading.py, shared/schemas/learning.py
• Pattern: Each stream has a schema (e.g., RawArticle, ScoredArticle, TradeSignal, TradeExecution)
• Services validate on consume, serialize with model_dump(mode="json") before publish

Entry Points

News Fetcher:

• Location: services/news_fetcher/main.py
• Triggers: Starts as standalone service, polling on timers
• Responsibilities: Fetch RSS/Reddit articles, deduplicate, publish to news:raw

Sentiment Analyzer:

• Location: services/sentiment_analyzer/main.py
• Triggers: Consumes messages from news:raw stream
• Responsibilities: Score sentiment (FinBERT + Ollama fallback), extract tickers, persist articles, publish to news:scored

Market Data Service:

• Location: services/market_data/main.py
• Triggers: Starts as standalone service, fetches on startup + live streaming
• Responsibilities: Fetch OHLCV bars from Alpaca, publish to market:bars

Signal Generator:

• Location: services/signal_generator/main.py
• Triggers: Consumes from both news:scored and market:bars streams
• Responsibilities: Combine sentiment + technical data, run strategy ensemble, publish to signals:generated

Trade Executor:

• Location: services/trade_executor/main.py
• Triggers: Consumes from signals:generated stream
• Responsibilities: Risk check, position sizing, order submission, trade recording, publish to trades:executed

Learning Engine:

• Location: services/learning_engine/main.py
• Triggers: Consumes from trades:executed stream
• Responsibilities: Evaluate closed positions, adjust strategy weights, audit adjustments

API Gateway:

• Location: services/api_gateway/main.py
• Triggers: HTTP server listening on configurable port (default 8000)
• Responsibilities: REST/WebSocket API, user authentication (WebAuthn), portfolio sync task, query database

Backtester:

• Location: backtester/engine.py
• Triggers: Called from API endpoint /api/backtest (POST)
• Responsibilities: Historical replay with SimulatedBroker, metrics calculation, equity curve generation

Error Handling

Strategy: Graceful degradation and event loop resilience.

Patterns:

• Service-level: Catch broad Exception in polling loops (news fetcher), log and continue; metrics counters increment for monitoring
• Fallback chains: Sentiment analyzer tries FinBERT first, falls back to Ollama on low confidence
• Risk checks: Trade executor rejects signals that fail risk checks (logs reason, continues)
• Database retries: If IntegrityError on article insert (duplicate), continue without re-raising
• Stream consumption: Consumer groups handle delivery guarantees; manual XACK after processing ensures fault tolerance
• Shutdown: asyncio.TaskGroup and signal handlers ensure graceful termination; cleanup of DB/Redis connections in FastAPI lifespan

Cross-Cutting Concerns

Logging:

• Each service uses Python logging module with service name
• All exceptions logged via logger.exception() for stack traces
• Key events logged at INFO level (article counts, trades executed, weight updates)

Validation:

• Pydantic v2 schemas validate all messages on consume
• Database models use SQLAlchemy constraints (NOT NULL, UNIQUE, FOREIGN KEY)
• Config classes extend BaseSettings with environment variable prefixes (TRADING_*)

Authentication:

• WebAuthn (passkey) via webauthn PyPI package (services/api_gateway/auth/)
• JWT tokens issued at login, verified on every protected endpoint via get_current_user dependency
• Credentials stored as UserCredential model (credential ID, public key, counter)

Telemetry:

• OpenTelemetry meters instantiated per service via setup_telemetry(service_name, metrics_port)
• Counters: articles_fetched, finbert_count, ollama_count, rejections, trades_executed
• Histograms: fill_latency, sentiment analysis duration, strategy evaluation time
• Prometheus scrapes /metrics endpoint (default port 9090) from each service
• No Prometheus/Grafana deployed; uses existing infrastructure for metrics collection