🚀 Algorithmic Trading Platform

A production-ready, cloud-native algorithmic trading platform built with modern data engineering and ML technologies. Real-time crypto market data processing, strategy backtesting, and automated trading execution.

🎯 Project Overview

This platform demonstrates advanced technical skills in:

Stream Processing: Real-time market data ingestion and processing
Time-Series Analytics: High-performance OLAP queries on billions of ticks
Machine Learning: Price prediction and trading signal generation
Distributed Systems: Event-driven microservices architecture
Cloud-Native Development: LocalStack for AWS services, managed databases
DevOps: Infrastructure as Code, local → cloud portability

Tech Stack:

Streaming & Messaging: Kafka (data streaming), RabbitMQ (task queues), Kinesis (AWS alternative)
Storage: S3 (LocalStack), ClickHouse (time-series), Redis (cache), PostgreSQL (metadata)
Application Layer: FastAPI, Python, AsyncIO, MLflow, Jupyter
Monitoring: Grafana, Prometheus
Cloud Services (LocalStack): Lambda, DynamoDB, EventBridge, SQS

📊 Architecture

Hybrid Messaging Architecture

┌─────────────────────────────────────────────────────────────────┐
│                  DATA STREAMING (Kafka)                         │
│  Exchange WebSockets → Kafka → [S3, ClickHouse, Redis]        │
│  • High throughput (60+ symbols)                               │
│  • Message replay for backtesting                              │
│  • Multiple consumers (analytics, ML)                          │
│  • 24h retention                                               │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                  TASK QUEUES (RabbitMQ)                         │
│  Trading API → RabbitMQ → Workers                              │
│  • Order execution (priority queues)                           │
│  • Backtest jobs (parallel workers)                            │
│  • Notifications (email, telegram)                             │
│  • Dead letter queues for failures                             │
└─────────────────────────────────────────────────────────────────┘

Local Development Stack (Phase 2)

Exchange REST API (ccxt)          Exchange WebSocket (Binance/Coinbase/Kraken)
    ↓                                          ↓
Sync Service (every 60s)          Market Data Ingestion Service
  - Initial backfill: 100 candles   - 60+ symbols, real-time trades
  - Regular sync: 5 latest candles  - Samples ~4% of trades
    ↓                                          ↓
ClickHouse (candles_1m/5m/1h)     Redis (latest_price:{exchange}:{symbol})
  ReplacingMergeTree, TTL 90d       Real-time signals for Phase 3+
    ↓
Indicator Service (every 60s + 10s delay)
  - Catch-up mode: all historical candles on startup
  - Calculates: SMA 20/50, EMA 12/26, RSI 14, MACD
  - Stores: ClickHouse (indicators) + Redis (60s TTL)
    ↓
Grafana (technical-analysis.json)
  └── 4 panels: Price+Indicators, RSI, MACD, Volume

Future Stack (Phase 3+)

Trading API (FastAPI) → RabbitMQ → Workers
    ↓
├── Strategy Engine
├── ML Pipeline (Jupyter)
├── Grafana (Docker) - Monitoring
└── Prometheus - Metrics collection

Cloud Production (Same code, different endpoints)

Replace LocalStack endpoints with real AWS services
- LocalStack S3 → AWS S3
- LocalStack Kinesis → AWS Kinesis
- LocalStack Lambda → AWS Lambda
- Docker ClickHouse → ClickHouse Cloud
- Docker Redis → AWS ElastiCache

📁 Project Structure

DataPlatform/
│
├── 📁 config/                           # Configuration Management
│   ├── settings.py                      # Pydantic Settings (from .env + YAML)
│   ├── loader.py                        # YAML config loaders
│   └── providers/                       # Per-service YAML configs
│       ├── databases.yaml               # ClickHouse, Redis, PostgreSQL
│       ├── exchanges.yaml               # WebSocket + REST API per exchange
│       ├── indicators.yaml              # Indicator definitions + service settings
│       ├── streaming.yaml               # Kafka/Kinesis topics
│       ├── storage.yaml                 # S3/GCS object storage
│       └── sync.yaml                    # Sync Service timing + REST API config
│
├── 📁 core/                             # Core Abstractions (Cloud-agnostic)
│   ├── interfaces/                      # Abstract Base Classes
│   │   ├── cache.py                     # BaseCacheClient (Redis, Memcached)
│   │   ├── database.py                  # BaseTimeSeriesDB (ClickHouse, TimescaleDB)
│   │   ├── market_data.py               # BaseExchangeWebSocket + BaseExchangeRestAPI
│   │   ├── indicators.py                # BaseIndicator
│   │   ├── storage.py                   # BaseStorageClient (S3, GCS)
│   │   ├── streaming_producer.py        # BaseStreamProducer (Kafka, Kinesis)
│   │   └── streaming_consumer.py        # BaseStreamConsumer
│   ├── models/
│   │   └── market_data.py               # Trade, Candle, OrderBook (Pydantic)
│   ├── validators/
│   │   └── market_data.py               # Spike detection, crossed book checks
│   └── utils/
│       └── config.py                    # load_yaml_safe()
│
├── 📁 providers/                        # Cloud Provider Implementations
│   ├── aws/
│   │   ├── kinesis.py                   # KinesisStreamProducer
│   │   └── s3.py                        # S3StorageClient
│   ├── opensource/
│   │   ├── clickhouse.py                # ClickHouseClient (connection pool)
│   │   ├── redis_client.py              # RedisClient (queued writes)
│   │   └── kafka_stream_producer.py     # KafkaStreamProducer
│   ├── binance/
│   │   ├── websocket.py                 # BinanceWebSocket
│   │   └── rest_api.py                  # BinanceRestAPI (ccxt)
│   ├── coinbase/
│   │   ├── websocket.py                 # CoinbaseWebSocket
│   │   └── rest_api.py                  # CoinbaseRestAPI (ccxt)
│   └── kraken/
│       ├── websocket.py                 # KrakenWebSocket
│       └── rest_api.py                  # KrakenRestAPI (ccxt)
│
├── 📁 factory/
│   └── client_factory.py                # create_exchange_rest_api(), create_timeseries_db(), etc.
│
├── 📁 domain/                           # Business Logic
│   └── indicators/
│       ├── moving_averages.py           # SMA, EMA, WMA
│       ├── momentum.py                  # RSI, MACD, Stochastic
│       └── registry.py                  # Load indicators from config
│
├── 📁 services/                         # Microservices (Phase 2)
│   ├── market_data_ingestion/           # WebSocket → Redis (real-time signals)
│   │   ├── main.py
│   │   ├── stream_processor.py
│   │   └── websocket_client.py
│   ├── sync_service/                    # REST API → ClickHouse (authoritative candles)
│   │   └── main.py                      # Runs every 60s, backfills 100 candles on startup
│   └── indicator_service/               # ClickHouse → Calculate → Redis + ClickHouse
│       ├── main.py                      # Runs every 60s (+10s delay after sync)
│       ├── calculator.py
│       ├── persistence.py
│       └── indicator_loader.py
│
├── 📁 infrastructure/
│   ├── docker/
│   │   ├── docker-compose.yml           # ClickHouse, Redis, PostgreSQL, Grafana
│   │   └── clickhouse/
│   │       ├── init-and-migrate.sh      # Auto-applies migrations on startup
│   │       └── migrations/              # 000–006 SQL migrations
│   └── terraform/                       # LocalStack (S3, Kinesis)
│
├── 📁 monitoring/
│   └── grafana/
│       ├── dashboards/
│       │   └── technical-analysis.json  # Price+Indicators, RSI, MACD, Volume
│       └── provisioning/
│           └── datasources.yml          # ClickHouse, Prometheus, PostgreSQL
│
├── 📁 tests/
│   ├── unit/                            # 183 tests, no Docker required
│   │   ├── indicators/                  # SMA, EMA, RSI, MACD
│   │   ├── services/                    # stream_processor, websocket_client
│   │   ├── test_config/                 # Settings Phase 2
│   │   ├── test_providers/              # ClickHouse pool, queued cache/db/streaming, REST API
│   │   ├── test_services/               # indicator calculator + persistence
│   │   ├── test_factory.py
│   │   └── test_models.py
│   └── integration/                     # Requires Docker + Terraform
│       ├── pipelines/                   # End-to-end pipeline tests
│       ├── infrastructure/              # ClickHouse, Redis, Docker readiness
│       ├── idempotency/                 # Candle/indicator/cache quality
│       └── factory/                     # Exchange factory + REST API network
│
├── 📁 docs/
│   ├── DOCKER_SETUP.md
│   ├── TERRAFORM_SETUP.md
│   ├── KINESIS_VS_KAFKA.md              # Architecture decision record
│   ├── COMPARISON.md
│   └── WHEN_TO_USE_CLOUD_DB.md
│
├── .env.example                         # Environment template
├── .gitignore
├── pyproject.toml                       # uv dependencies
├── Makefile                             # docker-up/down, terraform-apply, etc.
└── CLAUDE.md                            # AI assistant context + architecture guide

🔑 Key Design Principles

1. Separation of Concerns:

core/interfaces/ - Abstract contracts (cloud-agnostic)
providers/ - Concrete implementations (AWS, GCP, Azure, LocalStack)
factory/ - Auto-create clients based on config
domain/ - Business logic (trading strategies, NOT cloud code)
services/ - Microservices (API, ingestion, strategy engine)

2. Dependency Flow:

services/ → domain/ → factory/ → providers/ → core/interfaces/
                         ↓
                     config/

3. Cloud Portability:

Code in services/ and domain/ is 100% cloud-agnostic
Switch clouds by changing .env file only
No vendor lock-in

4. Testability:

Mock implementations of core/interfaces/ for testing
Fixtures in tests/fixtures/
Separate unit and integration tests

📊 Architecture Maturity (vs. Quant Fund Standard)

Mapping hiện trạng vào 5 lớp workflow của quỹ định lượng:

Layer 1 — Data Infrastructure      ██████████  ✅ Complete
          Multi-exchange WebSocket + REST API, ClickHouse OHLCV,
          Redis cache, PostgreSQL paper trading, MLflow artifacts

Layer 2 — Feature / Factor Eng.    ████░░░░░░  ~40%
          10 regime features + 16 sizing features (price-based only)
          Missing: funding rate, open interest, cross-asset, alternative data
          → Phase 8C target

Layer 3 — Alpha Validation          ███░░░░░░░  ~30%
          Backtest engine + walk-forward runner exist
          Missing: IC (information coefficient), decay analysis,
          factor correlation matrix, signal quality filter
          → Phase 8B target

Layer 4 — Portfolio Construction    ████████░░  ~80%
          MVO optimizer, regime gates (Layer 1 + Layer 2),
          meta-labeling filter, vol overlay, 4-slot MSAPE engine

Layer 5 — Risk & Execution          █████░░░░░  ~50%
          RiskManager, PaperBroker, BacktestBroker
          Missing: slippage model, market impact, execution algos

📈 Backtest Results Summary (2022-01-01 → 2026-02-27, Binance 1h)

Strategy	Symbol	Return	Sharpe	MaxDD	Trades
`volatility_breakout`	SOLUSDT	+7.11%	0.580	3.50%	234
`volatility_breakout`	ETHUSDT	+3.92%	0.575	1.60%	232
`volatility_breakout`	BTCUSDT	+0.78%	0.142	2.50%	248
`sma_crossover`	ETHUSDT	+1.42%	0.099	9.04%	830
`sma_crossover`	BTCUSDT	-1.27%	-0.070	7.00%	862
`rsi_mean_reversion`	ETHUSDT	-4.68%	-0.224	10.90%	295
`rsi_mean_reversion`	BTCUSDT	-6.75%	-0.465	8.67%	254
`ma_trend`	BTCUSDT	-7.89%	-0.678	8.38%	1,561
`volatility_breakout`	BNBUSDT	-5.56%	-0.884	6.16%	246
`adaptive_trend`	BTCUSDT	-12.92%	-0.747	16.97%	2,187
`lower_band_reversion`	BTCUSDT	-15.46%	-1.397	16.67%	1,558
`fade_breakout`	BTCUSDT	-13.93%	-2.106	14.33%	1,594

Portfolio MSAPE — 4 slots, regime filter ON:

Period	Return	Ann.	Sharpe	MaxDD	AvgLev
2022–2026 (4yr)	+2.38%	+0.58%/yr	0.295	-2.56%	1.47×
2024 only	+2.58%	+2.90%/yr	1.20	-1.60%	1.36×

Key insight: ma_trend (1,561 trades → $1,512 commission) and adaptive_trend (2,187 trades → $2,082 commission) lose 14–20% of capital to fees. Regime filter cuts them to ~200 trades and turns Sharpe positive.

🗺️ Development Phases

📍 PHASE 1: Foundation & Real-time Data ✅ COMPLETED

Goal: Stream live crypto market data into the system

Minimal Features (Required):

LocalStack setup (S3, Kinesis, Lambda)
Docker Compose for stateful services (ClickHouse, Redis, Grafana, Kafka)
WebSocket connection to Binance (BTC, ETH)
Stream to Kafka/Kinesis → S3 (raw data backup)
ClickHouse table: trades (symbol, price, quantity, timestamp)
Grafana dashboard: Real-time price line chart
Environment config: .env + YAML configs (cloud-agnostic)

Advanced Features (Completed):

Multi-exchange support (Binance, Coinbase, Kraken)
60+ crypto symbols streaming (20+ per exchange)
Order book depth data with quality validation
Data quality validation (spike detection, crossed book checks)
Kafka migration (KRaft mode, dual listeners)
Cloud-agnostic architecture (factory pattern, generic naming)
Comprehensive testing (62/62 tests passing: 46 unit, 16 integration)
Git pre-push hooks (secrets detection, lint, format, tests)
YAML-based configuration (public) + .env secrets (gitignored)

Services:

Docker: Kafka, ClickHouse, Redis, PostgreSQL, Grafana, Prometheus
LocalStack: S3, Kinesis (optional alternative to Kafka)
Python: Market data ingestion service with async WebSocket clients

Tech Showcase:

Cloud-Agnostic Design: Factory pattern, swappable providers (Kafka/Kinesis, S3/GCS, ClickHouse/TimescaleDB)
Multi-Exchange Integration: Binance, Coinbase, Kraken with unified interface
Kafka Streaming: KRaft mode, dual listeners (Docker + host), 24h retention
Data Quality: Spike detection, order book validation, data validators
Testing: 62/62 tests (unit + integration), pytest markers, timezone-aware
DevOps: Git hooks, ruff linting/formatting, uv dependency management
Monitoring: Grafana dashboards with multi-exchange metrics

📍 PHASE 2: Data Processing & Analytics ✅ COMPLETED

Goal: Transform raw ticks into analytical datasets

Minimal Features (Required):

OHLCV candlesticks (1m, 5m, 1h) via REST API
ClickHouse candle tables (direct inserts, no MVs)
Technical indicators: SMA(20, 50), EMA(12, 26), RSI, MACD
Redis caching for latest indicators
Grafana technical analysis dashboard

Advanced Features (Completed):

Advanced indicators: RSI, MACD with custom implementations
Multiple timeframes (1m, 5m, 1h)
Initial backfill (100 candles on startup)
Catch-up mode (calculate indicators for all historical candles)
Sequential processing (avoid ClickHouse connection conflicts)

Services:

Redis (cache for indicators)
ClickHouse (candles + indicators storage)
NEW: Sync Service - Scheduled REST API klines fetching
NEW: Indicator Service - Scheduled indicator calculation

Architecture: Industry Standard - Scheduled Jobs

Exchange REST API (authoritative OHLCV data)
   ↓
Sync Service (every 60 seconds)
   ├─ Fetches latest klines via ccxt
   ├─ Initial backfill: 100 candles on startup
   ├─ Regular sync: 5 latest candles
   └─ Stores in ClickHouse (candles_1m, 5m, 1h)
   ↓
Indicator Service (every 60 seconds + 10s delay)
   ├─ Reads candles from ClickHouse
   ├─ Catch-up mode: Process all historical candles on startup
   ├─ Calculates: SMA 20/50, EMA 12/26, RSI, MACD
   ├─ Stores in ClickHouse (indicators table)
   └─ Caches in Redis (60s TTL)
   ↓
Grafana Dashboard
   └─ 4-panel layout: Price+Indicators, RSI, MACD, Volume

Key Design Decisions:

REST API over WebSocket: Exchange REST API provides authoritative, complete OHLCV data (WebSocket only captures ~4% of trades via sampling)
Scheduled jobs over event-driven: Simple, predictable, easier to debug than Kafka consumers
Sequential processing: Avoid ClickHouse single-connection conflicts (clickhouse_driver limitation)
Catch-up mode: Calculate indicators for all backfilled data on startup
No Materialized Views: Direct inserts from REST API, simpler schema

Tech Showcase:

ccxt Integration: Unified exchange REST API abstraction (Binance, Coinbase, Kraken)
BaseExchangeRestAPI Pattern: Cloud-agnostic REST API abstraction (mirrors WebSocket pattern)
Scheduled Job Architecture: 60-second intervals, 10-second offset between services
ClickHouse Optimization: Sequential queries to avoid connection conflicts
Custom Indicator Implementation: SMA, EMA, RSI, MACD without TA-Lib dependency
Grafana Advanced Visualization: Candlestick charts with indicator overlays, styled panels
Initial Backfill + Catch-up: 100 candles fetched on startup, all historical data processed

📍 PHASE 3: Trading API & Strategy Framework (Day 5-6)

Goal: Build API and strategy execution engine with paper trading simulator

Minimal Features (✅ Complete):

Advanced Features (Optional — Phase 4+):

Strategy hot-reload (no restart needed)
Order types: Limit, Stop-Loss, Trailing Stop (Phase 3: market only)
Position sizing: Kelly Criterion (Phase 3: flat 10% per position)
Strategy parameter optimization via grid search (needs backtesting first)
RabbitMQ task queues (deferred — no value for paper trading, added in Phase 4)

Architecture Decision — No RabbitMQ in Phase 3:

Paper trading has no unreliable external boundary (no real exchange calls, no retries, no burst). A task queue adds zero value. Instead:

InlineTaskQueue — direct call, zero infra overhead
Phase 4 swaps to RabbitMQTaskQueue by changing 1 line in factory/client_factory.py

Services:

PostgreSQL (Docker) — source of truth for all paper trading state
FastAPI (Python service) — REST + WebSocket, port 8000
Strategy Engine (Python service) — 60s polling loop
Kafka (existing) — audit trail only, no new consumers

Tech Showcase:

Cloud-agnostic factory pattern (BaseBroker, BaseOrderStore, BaseTaskQueue)
PostgreSQL as source of truth (no in-memory state — crash-safe)
UUID generated at OrderManager — idempotent retries, no double-fills
Mark-to-market equity curve: cash + Σ(qty × live_price_redis) updated every 60s
WebSocket fan-out for real-time price streaming
Kafka audit trail (fire-and-forget) for Phase 4 backtest replay

📍 PHASE 4: Backtesting Engine ✅ COMPLETED

Goal: Test strategies against historical data with zero look-ahead bias

Backtester — fill at next-bar open, equity at current-bar close (intentional 1-bar lag)
BacktestBroker / BacktestOrderStore — in-memory, mirrors paper trading accounting
BacktestResult + compute_metrics — Sharpe (annualized √8760), MaxDD, win rate
CLI runner (services/backtest_runner/main.py) — argparse, ClickHouse only, --debug flag
JSON + CSV reports with per-trade log
WalkForwardRunner — sliding train/val windows with MLflow child runs
query_candles_history() — 2-year ClickHouse retention table for ML training data
36 unit tests (including 4 look-ahead bias tests)

Key design decisions:

candles_1h_history table (730d TTL) separate from candles_1h (90d TTL) — backtests need full history
No Redis/Postgres required — ClickHouse only for speed
limit=None in query_candles for full date-range (never mix numeric limit + date range)

📍 PHASE 5: ML/AI Integration ✅ COMPLETED

Goal: LightGBM regime classifier + volatility sizing model + MLflow tracking

📍 PHASE 6: Multi-Strategy Adaptive Portfolio Engine (MSAPE) ✅ COMPLETED

Goal: MVO-based multi-strategy portfolio with two-layer regime gating

Two-layer architecture:

Layer 1 (per-strategy regime gate):
  MATrendStrategy:            HOLD if regime_trending < 0.55
  RSIMeanReversionStrategy:   HOLD if regime_ranging  < 0.40
  VolatilityBreakoutStrategy: HOLD if regime_trending < 0.55

Layer 2 (portfolio MVO weights):
  EMA-smoothed Sharpe → SLSQP MVO → MRC cap → weekly rebalance
  Vol overlay: leverage = target_vol / port_vol, clamped [0.5, 1.5]

domain/portfolio/ — pure numpy/scipy computation, no I/O
- returns_matrix.py — align N equity curves → forward-filled ReturnsMatrix
- performance_engine.py — rolling Sharpe/Vol/MaxDD/LedoitWolf covariance
- allocation_engine.py — stateful EMA-smoothed Sharpe → MVO
- vol_overlay.py — leverage scaling
PortfolioBacktester — N sequential sub-backtests → bar-by-bar MVO simulation
PortfolioEngine (live) — reads Redis indicators → MVO weights → paper orders
CLI (services/portfolio_runner/main.py) — --use-yaml-slots, --regime-filter
MATrendStrategy — EMA50/200 crossover; requires "close" key in indicators dict
create_portfolio_backtester(use_regime_filter=False) — loads classifier from MLflow
Grafana portfolio dashboard
46 new unit tests

📍 PHASE 7a: Meta-Labeling ✅ COMPLETED

Goal: Filter strategy signals by trade quality confidence

triple_barrier_label() — PT/SL/timeout barriers; entry at next bar's open (look-ahead-free)
MetaLabeler — LightGBM binary; drops y==0 (timeout); threshold=0.55; untrained returns 0.5
Backtester meta_labeler param — after signal, if meta_confidence < threshold → HOLD
--meta-labeler flag in backtest_runner/main.py
--train-meta-labeler --strategy <name> in ml_trainer/main.py
Factory: create_meta_labeler(), load_meta_labeler()
Config: ml.yaml meta_labeler section (pt_pct=0.015, sl_pct=0.008, max_hold_bars=48)
17 unit tests

📍 PHASE 7b: Short Selling + AdaptiveTrendStrategy ✅ COMPLETED

Goal: Long-short capability with ATR trailing stop

Signal.SHORT + Signal.COVER added to signal enum
BacktestOrderStore — SHORT (negative qty, cash+=), COVER (cash-=, PnL=(entry-cover)×qty-commission)
BacktestBroker / PaperBroker — margin validation for SHORT, position validation for COVER
RiskManager — reject double-short, reject COVER with no short; symbol param in calculate_quantity
AdaptiveTrendStrategy — EMA+ATR long-short; BUY/SHORT on entry, SELL/COVER on ATR trailing stop; stateful; regime gate
19 unit tests (short selling), 17 unit tests (adaptive trend)
Total: 470 unit tests passing

📍 PHASE 7c: SHORT/COVER Integration Fixes ✅ COMPLETED

Goal: Close real bugs found after Phase 7b code review

OrderManager.process_signal() — add symbol=symbol to calculate_quantity call (bug: multi-symbol SELL/COVER used wrong position)
ml_trainer/main.py — capture Signal.SHORT in addition to Signal.BUY for meta-labeler training
Docstring updates: generate_signal(), validate(), process_signal()

📍 PHASE 8: Alpha Factory 🔜 NEXT

Goal: Move from ~7 alphas to 20-30 validated signals using quant-grade validation

Execution order: B → C → A

8B — Signal Quality Filter (Layer 3: Alpha Validation)

Problem diagnosed from backtest: ma_trend (1,561 trades) and adaptive_trend (2,187 trades) lose 14–20% of capital to commission. Signal frequency is too high — not signal direction.

Plan:

Add min_confirmation_bars: int parameter to MATrendStrategy and AdaptiveTrendStrategy
- Require N consecutive bars of signal confirmation before entry (default: 3)
- Reduces ma_trend trades: 1,561 → ~200, commission $1,512 → ~$200
IC (Information Coefficient) framework in domain/alpha/ic.py
- compute_ic(factor_values, forward_returns, periods=[1,4,8,24]) — Spearman rank correlation
- IC > 0.02 = weak signal, > 0.05 = usable, > 0.10 = strong
Decay analysis — IC rolling window to detect when signal loses edge
Factor correlation matrix — prevent redundant alphas from dominating portfolio
services/alpha_lab/ — standalone script to evaluate new factors before wiring into strategies

Key files:

domain/alpha/
  ic.py              # IC + ICIR computation
  decay.py           # Rolling IC decay analysis
  correlation.py     # Factor cross-correlation matrix
services/alpha_lab/
  evaluate.py        # CLI: evaluate a factor against forward returns

8C — Data Expansion (Layer 2: Feature Engineering)

Problem: All current features are price-derived. Quant funds use data price cannot provide.

Plan:

Funding rate — Binance perpetual futures (ccxt: exchange.fetch_funding_rate)
- Negative funding = shorts paying longs = forced short covering imminent
- Extreme positive funding = overleveraged longs = mean reversion setup
Open interest — exchange.fetch_open_interest via ccxt
- Rising OI + rising price = trend confirmation
- Rising OI + falling price = capitulation risk
Cross-asset momentum — BTC 24h return as feature for ETH/SOL/BNB strategies
- BTC leads altcoins by ~1-4 hours on large moves
Basis — spot vs futures price spread (when futures available)
Store in trading.market_context ClickHouse table (hourly snapshots)
Inject into FeaturePipeline as additional features
Add to indicators.yaml and config/settings.py

Key files:

providers/binance/futures.py         # Funding rate + OI fetcher
services/context_service/main.py     # Hourly market context collector
domain/ml/features.py                # Add FUNDING_FEATURES, OI_FEATURES
infrastructure/docker/clickhouse/migrations/008_market_context.sql

8A — Alpha Factory Scale-up (Layer 2+3: More Alphas + Validation)

Goal: Expand from 7 alphas to 20-30 using IC-validated factors

Candidate alpha families:

Family	Examples	IC target
Momentum	4h/12h/24h/72h price return, ROC	0.03–0.06
Seasonality	hour-of-day, day-of-week return patterns	0.02–0.04
Microstructure	bid-ask spread proxy, trade imbalance	0.04–0.08
Funding carry	funding rate z-score, cumulative funding	0.05–0.10
Cross-asset	BTC/ETH relative strength, BTC dominance delta	0.03–0.07
Volatility regime	realized vol percentile rank, VIX analogue	0.03–0.06

Process for each alpha:

Compute factor → 2. IC test (needs IC > 0.03) → 3. Decay check (still predictive at 8h+?) → 4. Correlation check (< 0.60 with existing alphas) → 5. Add to strategy or FeaturePipeline

Key files:

domain/alpha/
  momentum.py        # Multi-period price momentum factors
  seasonality.py     # Calendar-based return patterns
  carry.py           # Funding rate carry factors
  cross_asset.py     # BTC-relative strength signals
services/alpha_lab/
  backtest_alpha.py  # Fast vectorized alpha backtest (no strategy overhead)
  report.py          # IC, decay, correlation summary report

🛠️ Tech Stack Details

Messaging Architecture: Kafka vs RabbitMQ

When to use Kafka (Data Streaming):

✅ Market data ingestion (trades, order books)
✅ Event streaming with replay capability
✅ Multiple consumers need same data
✅ High throughput (60+ symbols, thousands msg/sec)
✅ Message retention (24h+) for backtesting
✅ Analytics pipelines

When to use RabbitMQ (Task Queues):

✅ Order execution (need acknowledgment)
✅ Priority queues (urgent stop-loss first)
✅ Dead letter queues (failed task handling)
✅ Work distribution (backtest jobs to workers)
✅ Notifications (email, telegram)
✅ Long-running async jobs

Implementation:

# factory/client_factory.py
def create_stream_client() -> BaseStreamClient:
    """Kafka/Kinesis for data streaming"""
    if settings.CLOUD_PROVIDER == "opensource":
        return KafkaStreamClient()
    elif settings.CLOUD_PROVIDER == "aws":
        return KinesisStreamClient()

def create_task_queue() -> BaseTaskQueue:
    """RabbitMQ/SQS for task queues (Phase 3+)"""
    if settings.CLOUD_PROVIDER == "opensource":
        return RabbitMQClient()
    elif settings.CLOUD_PROVIDER == "aws":
        return SQSClient()

LocalStack (AWS Services Emulation)

S3: Object storage for raw data, backups, ML artifacts
Kinesis: Real-time data streaming (alternative to Kafka)
Lambda: Serverless compute for event processing
DynamoDB: NoSQL database for high-throughput data
EventBridge: Event bus for decoupled architecture
SQS/SNS: Message queuing and pub/sub
API Gateway: REST/WebSocket API endpoints
CloudWatch: Logging and metrics
Secrets Manager: Secure credential storage
Step Functions (optional): Workflow orchestration

Stateful Services (Docker or Cloud Managed)

Data Storage

ClickHouse: Time-series OLAP database (billions of ticks)
- Local: Docker container
- Cloud: ClickHouse Cloud, Altinity Cloud
Redis: In-memory caching (latest prices, positions)
- Local: Docker container
- Cloud: Upstash Redis, AWS ElastiCache
PostgreSQL: Relational metadata (strategies, users, configs)
- Local: Docker container
- Cloud: Supabase, Neon, AWS RDS

ML & Analytics

Jupyter: Interactive data exploration
MLflow: Experiment tracking, model registry
scikit-learn, XGBoost, LightGBM: Traditional ML
TensorFlow/PyTorch: Deep learning (LSTM, Transformers)

Monitoring & Visualization

Grafana: Real-time operational dashboards (Docker)
Prometheus: Metrics collection (Docker)

Application Layer

Python 3.10+: Primary language
boto3: AWS SDK (works with LocalStack)
FastAPI: High-performance API framework
Pydantic: Data validation
SQLAlchemy: ORM for PostgreSQL
clickhouse-connect: ClickHouse Python client
redis-py: Redis Python client

Infrastructure as Code

AWS CDK (recommended): Define LocalStack + AWS resources in Python
Terraform (alternative): HCL-based IaC
LocalStack Docker Compose: Service orchestration

📦 Services Breakdown

Minimal Stack (LocalStack + Core Services - ~3GB RAM)

LocalStack (unified container):
- All AWS services (S3, Kinesis, Lambda, DynamoDB, etc.) - 512MB

Docker Stateful Services:
- clickhouse (1.5GB)
- redis (256MB)
- postgres (512MB)
- grafana (256MB)

Python Services (run locally, not Docker):
- market-data-ingestion (256MB)
- trading-api (512MB)

Full Stack (All Services - ~8GB RAM)

+ LocalStack Pro features (optional)
+ jupyter (1GB)
+ mlflow (512MB)
+ prometheus (512MB)
+ Additional Lambda functions
+ Step Functions workflows

Cloud Alternative (Zero Local Resources)

Replace all with managed services:
- LocalStack → Real AWS (free tier)
- Docker ClickHouse → ClickHouse Cloud ($300 credit)
- Docker Redis → Upstash Redis (free tier)
- Docker Postgres → Supabase (free tier)
- Python services → AWS Lambda or Railway

🚀 Quick Start

Prerequisites

Docker & Docker Compose
Python 3.10+
8GB+ RAM (minimal), 16GB+ recommended (full stack)
20GB+ disk space

1. Clone & Setup

git clone <repo-url>
cd DataPlatform

# Copy environment template
cp .env.example .env.local

2. Start Minimal Stack (Phase 1)

# Start LocalStack + stateful services
docker-compose up -d

# Services started:
# - localstack (all AWS services)
# - clickhouse
# - redis
# - postgres
# - grafana

# Check services health
docker-compose ps

# View logs
docker-compose logs -f localstack

3. Run Services (Phase 2)

Run in this order to ensure proper data flow:

# Terminal 1: Market Data Ingestion (WebSocket → Redis signals)
uv run python services/market_data_ingestion/main.py

# Terminal 2: Sync Service (REST API → ClickHouse candles, every 60s)
uv run python services/sync_service/main.py

# Terminal 3: Indicator Service (ClickHouse → Calculate → Redis/ClickHouse, every 60s)
uv run python services/indicator_service/main.py

4. Access Dashboards

LocalStack Dashboard: http://localhost:4566/_localstack/health
Grafana: http://localhost:3000 (admin/admin)
ClickHouse: http://localhost:8123
Redis: localhost:6379
PostgreSQL: localhost:5432

5. Verify Data Flow

# Check LocalStack health
curl http://localhost:4566/_localstack/health

# List S3 buckets (LocalStack)
aws --endpoint-url=http://localhost:4566 s3 ls

# Check Kinesis stream (LocalStack)
aws --endpoint-url=http://localhost:4566 kinesis list-streams

# Query ClickHouse
curl "http://localhost:8123/?query=SELECT count() FROM trades"

🌐 Local → Cloud Migration

Configuration-Based Deployment

All services use environment variables for endpoints:

# .env.local (LocalStack + Docker)
AWS_ENDPOINT_URL=http://localhost:4566
AWS_REGION=us-east-1
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test

CLICKHOUSE_HOST=localhost
CLICKHOUSE_PORT=8123
REDIS_URL=redis://localhost:6379
POSTGRES_URL=postgresql://admin:password@localhost:5432/trading

# .env.cloud (Production AWS)
AWS_ENDPOINT_URL=  # Empty = use real AWS
AWS_REGION=us-east-1
AWS_ACCESS_KEY_ID=<real-key>
AWS_SECRET_ACCESS_KEY=<real-secret>

CLICKHOUSE_HOST=abc123.clickhouse.cloud
REDIS_URL=rediss://username:password@redis.cloud:6379
POSTGRES_URL=postgresql://user:pass@db.supabase.com/postgres

Switch Environments

# Local development with LocalStack
cp .env.local .env
docker-compose up -d
python services/market-data-ingestion/main.py

# Production (real AWS + managed databases)
cp .env.cloud .env
# Deploy Lambda functions
cd infra
cdk deploy  # or terraform apply

# Code remains IDENTICAL - boto3 SDK uses env vars!

Cloud Deployment Options

Option A: Docker on Cloud VM (easiest)

# On AWS EC2 / GCP Compute / Azure VM
git clone <repo>
docker-compose -f docker-compose.cloud.yml up -d

Option B: Managed Services

Kafka: Confluent Cloud, AWS MSK, Aiven
ClickHouse: ClickHouse Cloud, Altinity.Cloud
Redis: AWS ElastiCache, Redis Cloud
PostgreSQL: AWS RDS, Google Cloud SQL, Supabase
Airflow: Astronomer, Google Cloud Composer

Option C: Kubernetes

# Using Helm charts
helm install trading ./helm/trading-platform

📈 Project Milestones

Phase	Status	Can Demo	Technical Highlights
Phase 1	✅ Complete	Real-time prices	WebSocket, multi-exchange, ClickHouse
Phase 2	✅ Complete	Charts + indicators	REST API sync, scheduled jobs, SMA/EMA/RSI/MACD
Phase 3	✅ Complete	Paper trading live	FastAPI, strategy engine, PostgreSQL
Phase 4	✅ Complete	Backtest CLI + reports	Backtester, walk-forward, JSON/CSV output
Phase 5	✅ Complete	ML regime classifier	LightGBM 3-class, MLflow, vol sizing model
Phase 6	✅ Complete	Portfolio engine	MSAPE, MVO, two-layer regime gates
Phase 7a	✅ Complete	Meta-labeling	Triple-barrier labels, LightGBM quality filter
Phase 7b	✅ Complete	Long-short trading	SHORT/COVER signals, AdaptiveTrendStrategy
Phase 7c	✅ Complete	Bug fixes	symbol param in calculate_quantity, ml_trainer SHORT capture
Phase 8B	🔜 Next	Signal quality	IC framework, confirmation bars, decay analysis
Phase 8C	🔜 Planned	Alt data	Funding rate, open interest, cross-asset momentum
Phase 8A	🔜 Planned	Alpha factory	20-30 IC-validated factors, alpha lab CLI

Current: 470 unit tests passing | 12 strategies backtested | 4-asset portfolio live

🎓 Learning Outcomes

By completing this project, you will master:

Data Engineering

Stream processing architectures
Time-series database optimization
Data pipeline orchestration
ETL/ELT patterns

Backend Development

REST + WebSocket APIs
Event-driven microservices
Caching strategies
Database design

Machine Learning

Time-series forecasting
Feature engineering
Model training & evaluation
MLOps practices

DevOps

Containerization (Docker)
Infrastructure as Code
Monitoring & alerting
CI/CD pipelines

System Design

Distributed systems
Scalability patterns
Fault tolerance
Performance optimization

📊 Performance Targets

Latency

Market data ingestion → ClickHouse: < 100ms (p99)
Strategy signal generation: < 50ms
API response time: < 100ms (p95)
ML model inference: < 50ms

Throughput

Market data ingestion: 10,000+ trades/second
Strategy evaluations: 1,000+ per second
API requests: 5,000+ req/second

Data Volume

100M+ ticks per day (~1GB/day compressed)
1B+ ticks for backtesting (2020-2024)
Query performance: Sub-second on billions of rows

🔐 Security Considerations

Minimal (Phase 1-4)

Environment variables for secrets
Docker network isolation
Read-only API keys (exchange APIs)

Production (Phase 6)

API authentication (JWT)
Rate limiting
Input validation
HTTPS/TLS encryption
Secrets management (Vault, AWS Secrets Manager)
Network policies (firewall rules)
Regular security audits

🧪 Testing Strategy

Unit Tests

Strategy logic
Technical indicators calculation
Data validation

Integration Tests

Kafka → ClickHouse pipeline
API endpoints
ML model inference

Performance Tests

Load testing (Locust, K6)
Stress testing (max throughput)
Latency benchmarks

Backtests

Historical strategy validation
Out-of-sample testing
Walk-forward analysis

📚 Resources & References

Documentation

Data Sources

Binance WebSocket API
yfinance - Historical stock data
ccxt - Multi-exchange crypto library

Learning

Kaggle: Crypto/Stock datasets
QuantConnect: Algorithmic trading tutorials
TradingView: Technical analysis

🤝 Contributing

This is a learning project showcasing technical skills. Feel free to:

Fork and customize for your use case
Submit issues for bugs/improvements
Share your results and learnings

⚠️ Disclaimer

Educational & Research Purposes Only

This platform is built for:

Learning data engineering and ML concepts
Portfolio demonstration
Academic research

NOT intended for:

Real money trading without extensive modifications
Production use without proper risk management
Financial advice

Trading involves significant risk of loss. Past performance does not guarantee future results.

🔄 Database Migrations (Alembic) - Phase 4+

Why Alembic?

Phase 1-3 (Development): Use postgres/init.sql (simple, fast iteration)

Schema is simple and changes infrequently
Can docker-compose down -v to recreate database
No production data to preserve

Phase 4+ (Production): Migrate to Alembic (version control, safety)

Have production data (cannot recreate database)
Schema changes need to be tracked and reversible
Team collaboration requires migration history

Setup Alembic

# Install
pip install alembic psycopg2-binary

# Initialize (creates migrations/ folder)
alembic init migrations

# Configure
# Edit migrations/env.py to use config/settings.py

migrations/env.py:

from config.settings import get_settings
settings = get_settings()

config.set_main_option("sqlalchemy.url", settings.postgres_dsn)

Create Migrations

Auto-generate from SQLAlchemy models:

# After changing models in domain/models/
alembic revision --autogenerate -m "add phone_number to users"

# Review the generated migration
# Edit migrations/versions/xxx_add_phone_number.py if needed

Manual migration:

alembic revision -m "add index on orders.symbol"

migrations/versions/001_add_phone.py:

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column('users',
        sa.Column('phone_number', sa.String(20), nullable=True)
    )
    op.create_index('idx_users_phone', 'users', ['phone_number'])

def downgrade():
    op.drop_index('idx_users_phone')
    op.drop_column('users', 'phone_number')

Apply Migrations

Local development:

# Apply all pending migrations
alembic upgrade head

# Rollback one migration
alembic downgrade -1

# Check current version
alembic current

# View migration history
alembic history

Production deployment:

# In Dockerfile or docker-compose
CMD alembic upgrade head && python main.py

docker-compose.yml:

services:
  trading-api:
    build: .
    command: >
      sh -c "alembic upgrade head &&
             uvicorn main:app --host 0.0.0.0"
    depends_on:
      postgres:
        condition: service_healthy

Migration Best Practices

1. Always test migrations:

# Test upgrade
alembic upgrade head

# Test downgrade
alembic downgrade -1

# Test re-upgrade
alembic upgrade head

2. Never modify old migrations:

Create new migration to fix issues
Old migrations are history (like Git commits)

3. Review auto-generated migrations:

Alembic may miss some changes
Check nullable, defaults, indexes

4. Use transactions:

def upgrade():
    with op.get_context().autocommit_block():
        # DDL statements here
        pass

5. Data migrations:

def upgrade():
    # Schema change
    op.add_column('users', sa.Column('status', sa.String(20)))

    # Data migration
    connection = op.get_bind()
    connection.execute(
        "UPDATE users SET status = 'active' WHERE is_active = true"
    )

CI/CD Integration

GitHub Actions:

- name: Run migrations
  run: |
    alembic upgrade head

- name: Run tests
  run: pytest

Deployment script:

#!/bin/bash
# deploy.sh

# Backup database
pg_dump $DB_URL > backup_$(date +%Y%m%d_%H%M%S).sql

# Run migrations
alembic upgrade head

# Deploy app
docker-compose up -d --build

Converting from init.sql to Alembic

Step 1: Initial migration from existing schema

# Start with empty migrations
alembic init migrations

# Create initial migration matching current init.sql
alembic revision -m "initial schema"

Step 2: Copy init.sql content to migration:

# migrations/versions/001_initial.py
def upgrade():
    # Copy CREATE TABLE statements from init.sql
    op.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
        ...
    )
    """)

Step 3: Mark as applied (don't re-run on existing DB):

# On existing database
alembic stamp head

# On new database
alembic upgrade head

Step 4: Future changes use Alembic

# All new schema changes
alembic revision --autogenerate -m "add new column"
alembic upgrade head

Troubleshooting

Migration conflicts:

# Multiple developers created migrations
# Merge and renumber
alembic merge heads -m "merge migrations"

Reset migrations (development only!):

# WARNING: Deletes all data!
docker-compose down -v
rm -rf migrations/versions/*
alembic revision -m "initial"
# Recreate schema in migration
alembic upgrade head

📝 License

MIT License - See LICENSE file for details

🎯 Next Steps (Phase 8B)

# Run all strategy backtests (2022–2026)
CLICKHOUSE_HOST=localhost uv run python services/backtest_runner/main.py \
  --strategy volatility_breakout --symbol SOLUSDT \
  --exchange binance --timeframe 1h \
  --start 2022-01-01 --end 2026-02-27

# Run portfolio with regime filter
CLICKHOUSE_HOST=localhost uv run python services/portfolio_runner/main.py \
  --use-yaml-slots --exchange binance --timeframe 1h \
  --start 2022-01-01 --end 2026-02-27 --regime-filter

# Run unit tests
uv run pytest tests/unit/ -v          # 470 tests

# Train meta-labeler
uv run python services/ml_trainer/main.py \
  --train-meta-labeler --strategy adaptive_trend \
  --symbol BTCUSDT --start 2022-01-01 --end 2024-12-31

Questions or Issues?

Check CLAUDE.md for architecture guide and design decisions
Check /docs folder for detailed guides

Built with ❤️ for learning and showcasing technical expertise

Keywords: #DataEngineering #MachineLearning #AlgorithmicTrading #StreamProcessing #ClickHouse #Kafka #Python #FastAPI #Docker #CloudNative

Name		Name	Last commit message	Last commit date
Latest commit History 33 Commits
config		config
core		core
data		data
domain		domain
factory		factory
infrastructure		infrastructure
monitoring		monitoring
notebooks		notebooks
providers		providers
scripts		scripts
services		services
tests		tests
.env.example		.env.example
.env.terraform		.env.terraform
.gitignore		.gitignore
.python-version		.python-version
CLAUDE.md		CLAUDE.md
Makefile		Makefile
README.md		README.md
main.py		main.py
pyproject.toml		pyproject.toml
pytest.ini		pytest.ini
ruff.toml		ruff.toml
uv.lock		uv.lock

Folders and files

Latest commit

History

Repository files navigation

🚀 Algorithmic Trading Platform

🎯 Project Overview

📊 Architecture

Hybrid Messaging Architecture

Local Development Stack (Phase 2)

Future Stack (Phase 3+)

Cloud Production (Same code, different endpoints)

📁 Project Structure

🔑 Key Design Principles

📊 Architecture Maturity (vs. Quant Fund Standard)

📈 Backtest Results Summary (2022-01-01 → 2026-02-27, Binance 1h)

🗺️ Development Phases

📍 PHASE 1: Foundation & Real-time Data ✅ COMPLETED

Minimal Features (Required):

Advanced Features (Completed):

📍 PHASE 2: Data Processing & Analytics ✅ COMPLETED

Minimal Features (Required):

Advanced Features (Completed):

📍 PHASE 3: Trading API & Strategy Framework (Day 5-6)

Minimal Features (✅ Complete):

Advanced Features (Optional — Phase 4+):

📍 PHASE 4: Backtesting Engine ✅ COMPLETED

📍 PHASE 5: ML/AI Integration ✅ COMPLETED

📍 PHASE 6: Multi-Strategy Adaptive Portfolio Engine (MSAPE) ✅ COMPLETED

📍 PHASE 7a: Meta-Labeling ✅ COMPLETED

📍 PHASE 7b: Short Selling + AdaptiveTrendStrategy ✅ COMPLETED

📍 PHASE 7c: SHORT/COVER Integration Fixes ✅ COMPLETED

📍 PHASE 8: Alpha Factory 🔜 NEXT

8B — Signal Quality Filter (Layer 3: Alpha Validation)

8C — Data Expansion (Layer 2: Feature Engineering)

8A — Alpha Factory Scale-up (Layer 2+3: More Alphas + Validation)

🛠️ Tech Stack Details

Messaging Architecture: Kafka vs RabbitMQ

LocalStack (AWS Services Emulation)

Stateful Services (Docker or Cloud Managed)

Data Storage

ML & Analytics

Monitoring & Visualization

Application Layer

Infrastructure as Code

📦 Services Breakdown

Minimal Stack (LocalStack + Core Services - ~3GB RAM)

Full Stack (All Services - ~8GB RAM)

Cloud Alternative (Zero Local Resources)

🚀 Quick Start

Prerequisites

1. Clone & Setup

2. Start Minimal Stack (Phase 1)

3. Run Services (Phase 2)

4. Access Dashboards

5. Verify Data Flow

🌐 Local → Cloud Migration

Configuration-Based Deployment

Switch Environments

Cloud Deployment Options

Option A: Docker on Cloud VM (easiest)

Option B: Managed Services

Option C: Kubernetes

📈 Project Milestones

🎓 Learning Outcomes

Data Engineering

Backend Development

Machine Learning

DevOps

System Design

📊 Performance Targets

Latency

Throughput

Data Volume

🔐 Security Considerations

Minimal (Phase 1-4)

Production (Phase 6)

🧪 Testing Strategy

Unit Tests

Integration Tests

Performance Tests