Apppublic

ambarish0221/FlashLedger

sourceHugging Faceupdated 4mo agoView on Hugging Face

0likes

App README

FlashLedger

A production-grade, low-latency order matching engine with real-time Kafka streaming, a PySpark feature pipeline, LSTM and LightGBM ML models, Groq LLM integration, and a React trading dashboard.

Overview

FlashLedger replicates the core infrastructure of a modern crypto exchange. Orders enter through a REST API, get matched by a price-time priority engine in microseconds, and simultaneously flow through a streaming analytics pipeline that feeds trained machine learning models. A React frontend visualises every layer of the system live.

The primary technical focus is the ML training pipeline: raw trade microstructure data produced by the order book is transformed by PySpark Structured Streaming into engineered features, which train both an LSTM price-direction model and a LightGBM market-regime recommender. Groq LLM is integrated as an additive layer that synthesises model outputs into natural-language commentary and powers an autonomous AI trading agent.

Architecture

Browser (React + Vite)
    |
    | REST + WebSocket
    v
FastAPI (app/)
    |-- /order         --> Matching Engine --> Order Book (in-memory, heap)
    |-- /prediction    --> LSTM inference  (ml/predict.py)
    |-- /insights      --> LightGBM recommender + Groq commentary (ml/recommender.py)
    |-- /chat          --> Groq LLM with live market context
    |-- /demo/start    --> Rule-based simulation (5 agent types, 5 regimes)
    |-- /demo/ai/start --> Groq AI agent (orders generated by Llama 3.3)
    |-- /ws            --> WebSocket broadcast (orderbook, trades, metrics)
    |
    | Kafka topic: trades
    v
PySpark Structured Streaming (spark/feature_pipeline.py)
    |-- 10-second tumbling windows
    |-- Aggregates: trade_volume, vwap, order_imbalance, trade_velocity
    v
PostgreSQL
    |-- trades table
    |-- market_features table
    |
    +--> ml/train.py          (LSTM training)
    +--> ml/train_recommender.py  (LightGBM training on BTC/USD)

Stack

Layer	Technology
API	FastAPI 0.109, Uvicorn, Pydantic v2
Matching Engine	Python heapq, price-time priority, in-memory
Database	PostgreSQL 15, SQLAlchemy 2.0 async, asyncpg
Streaming	Apache Kafka 7.6 (Confluent), PySpark 3.5 Structured Streaming
ML - Time Series	PyTorch 2.2 LSTM (2-layer, dropout, ReduceLROnPlateau)
ML - Recommender	LightGBM 4.x, scikit-learn, yfinance (2yr BTC/USD hourly data)
LLM	Groq API, Llama 3.3 70B Versatile
Frontend	React 18, Vite 5, Tailwind CSS 3, Recharts, Framer Motion, Lucide
Infrastructure	Docker Compose (6 services), nginx

Project Structure

FlashLedger/
├── app/
│   ├── main.py                   FastAPI entry point, CORS, lifespan
│   ├── api/
│   │   ├── routes.py             All REST + WebSocket endpoints
│   │   └── ws_manager.py         WebSocket broadcast manager
│   ├── engine/
│   │   ├── matching_engine.py    Price-time priority matching
│   │   └── order_book.py         Heap-based order book
│   ├── db/
│   │   ├── models.py             Trade + MarketFeature ORM models
│   │   └── database.py           Async PostgreSQL connection
│   ├── kafka/
│   │   └── producer.py           Confluent-Kafka producer (graceful degradation)
│   ├── ai/
│   │   └── groq_client.py        Groq AsyncGroq client (commentary + chat)
│   └── demo/
│       ├── runner.py             Rule-based simulation (market makers, regimes)
│       └── ai_runner.py          Groq-powered autonomous trading agent
├── spark/
│   └── feature_pipeline.py       PySpark Structured Streaming consumer
├── ml/
│   ├── train.py                  LSTM training (market_features -> model.pt)
│   ├── predict.py                LSTM inference module
│   ├── train_recommender.py      LightGBM training (BTC/USD yfinance data)
│   └── recommender.py            LightGBM inference + KNN similarity
├── frontend/
│   ├── src/
│   │   ├── App.jsx               Main layout, demo state management
│   │   ├── components/
│   │   │   ├── Header.jsx        Demo controls, regime badge, AI indicator
│   │   │   ├── OrderBook.jsx     Animated bid/ask depth bars
│   │   │   ├── TradeFeed.jsx     Live trade stream with side colouring
│   │   │   ├── PriceChart.jsx    Recharts price line + volume bars
│   │   │   ├── PredictionWidget.jsx  LSTM direction + confidence ring
│   │   │   ├── InsightsWidget.jsx    LightGBM recommendation + Groq analysis
│   │   │   ├── OrderForm.jsx     Buy/sell order entry form
│   │   │   ├── MetricsPanel.jsx  Engine metrics cards
│   │   │   ├── SparkMetrics.jsx  Pipeline topology documentation
│   │   │   └── ChatWidget.jsx    Floating Groq chat panel
│   │   ├── hooks/
│   │   │   └── useWebSocket.js   Auto-reconnect WS hook
│   │   └── api/
│   │       └── client.js         Axios client for all endpoints
│   ├── Dockerfile                Multi-stage node build + nginx serve
│   └── nginx.conf                SPA fallback + API reverse proxy
├── tests/
│   ├── test_order_book.py
│   ├── test_matching_engine.py
│   └── load_test.py
├── docker-compose.yml            6 services: postgres, zookeeper, kafka, app, spark, frontend
├── Dockerfile
├── requirements.txt
└── .env                          GROQ_API_KEY, DATABASE_URL, KAFKA_BOOTSTRAP_SERVERS

ML Pipeline

Feature Engineering (PySpark)

The Spark job reads the trades Kafka topic, applies 10-second tumbling windows with a 30-second watermark, and writes four features per window to the market_features PostgreSQL table:

Feature	Description
`trade_volume`	Total quantity traded in window
`vwap`	Volume-weighted average price
`order_imbalance`	Buy aggressor volume minus sell aggressor volume
`trade_velocity`	Trade count divided by window seconds

Run the pipeline:

bash

spark-submit \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,org.postgresql:postgresql:42.7.1 \
  spark/feature_pipeline.py

LSTM Price Direction Model

Reads market_features rows, builds 20-step sliding windows, and trains a 2-layer LSTM to predict binary price direction (up / down).

bash

python -m ml.train

Output: ml/model.pt (contains model weights, scaler params, feature metadata)

LightGBM Market Recommender

Downloads 2 years of BTC/USD hourly OHLCV data via yfinance (17,400+ bars), engineers 6 features matching the live PySpark output, and trains a 3-class classifier:

—Classes: BUY / HOLD / SELL (forward 3-period return, 0.5% threshold)
—Validation accuracy: 44.3% (baseline random = 33.3%)
—HOLD precision: 72% (the model is most reliable when recommending inaction)

bash

python -m ml.train_recommender

Output: ml/recommender.pkl, ml/rec_scaler.pkl, ml/rec_history.npz (17,485 embedding vectors for KNN similarity search)

Demo Modes

Rule-Based Demo (Primary)

Starts five agent types that simulate realistic exchange microstructure. Automatically cycles through five market regimes.

POST /api/v1/demo/start
POST /api/v1/demo/stop
GET  /api/v1/demo/status

Agents: mm_alpha, mm_beta (market makers), trend_1-10 (trend followers), noise_1-4 (noise traders), whale (regime-transition bursts), panic_seller (flash crash only)

Regimes: ranging -> bull / bear / high_vol -> flash_crash -> ranging

Tick interval: 0.4 seconds. Each tick broadcasts fresh orderbook + metrics via WebSocket.

AI Demo (Additional)

Calls Groq Llama 3.3 every 4 seconds with live market context (mid, spread, regime, VWAP delta, order imbalance, trade velocity) and submits the LLM-generated orders to the matching engine.

POST /api/v1/demo/ai/start
POST /api/v1/demo/ai/stop
GET  /api/v1/demo/ai/status

Requires GROQ_API_KEY in .env. Can run simultaneously with the rule-based demo.

API Reference

Order Submission

POST /api/v1/order

json

{
  "user_id": "trader1",
  "side": "buy",
  "price": 100.50,
  "quantity": 5.0
}

Response includes matched trades, remaining quantity, and matching latency in milliseconds.

Order Book

GET /api/v1/orderbook?depth=10

Recent Trades

GET /api/v1/trades?limit=100

Engine Metrics

GET /api/v1/metrics

LSTM Prediction

GET /api/v1/prediction

Returns: direction (up/down), confidence, model (lstm or heuristic), window_rows

Market Insights

GET /api/v1/insights

Returns: action (BUY/SELL/HOLD), confidence, regime, probabilities, insights, similar_conditions (KNN), rsi, feature_values, ai_commentary (Groq narrative, optional)

AI Chat

POST /api/v1/chat

json

{ "message": "What does the current order imbalance signal?" }

Returns a response from Llama 3.3 with live orderbook and metrics injected as context.

WebSocket

WS /api/v1/ws

Message types: orderbook_update, trade_executed, metrics_update. Send ping to receive pong.

Setup

Prerequisites

—Docker and Docker Compose
—Python 3.11+
—Node.js 20+ (frontend only)
—Groq API key (free tier at console.groq.com)

Environment

Copy .env and fill in your key:

DATABASE_URL=postgresql+asyncpg://flashledger:flashledger@localhost:5432/flashledger
KAFKA_BOOTSTRAP_SERVERS=kafka:9092
GROQ_API_KEY=gsk_...

Full Stack with Docker

bash

docker compose up --build

Services started: PostgreSQL, Zookeeper, Kafka, FlashLedger API, PySpark pipeline, React frontend

Frontend: http://localhost:3000 API docs: http://localhost:8000/docs

Local Development

bash

# Backend
pip install -r requirements.txt
uvicorn app.main:app --reload --port 8000

# Frontend
cd frontend
npm install
npm run dev

Train the Models

Both commands are independent and can be run in any order. The recommender downloads its own training data; the LSTM requires the Spark pipeline to have produced market_features rows first.

bash

# LightGBM recommender (downloads BTC/USD data automatically, ~2 min)
python -m ml.train_recommender

# LSTM (requires live market_features data from the Spark pipeline)
python -m ml.train

Testing

bash

# Unit tests
pytest tests/ -v

# Load test (requires running server)
python tests/load_test.py --orders 10000 --concurrency 100

License

MIT

MVP Features

—✅ Accept buy and sell orders via REST API
—✅ Maintain in-memory order book with heap-based priority queues
—✅ Match orders using price-time priority
—✅ Execute trades automatically
—✅ Record trades in PostgreSQL
—✅ Expose API endpoints for orders and order book
—✅ Sub-5ms local latency target

System Flow

User submits order → API receives → Engine checks book → Match occurs → Trade executed → Book updated → Trade stored

Architecture

┌─────────────┐     ┌────────────────┐     ┌─────────────┐
│   FastAPI   │────▶│  Matching      │────▶│ Order Book  │
│   /order    │     │  Engine        │     │ (In-Memory) │
└─────────────┘     └────────────────┘     └─────────────┘
                           │
                           ▼
                    ┌─────────────┐
                    │ PostgreSQL  │
                    │  (Trades)   │
                    └─────────────┘

Core Concepts

Order Book Structure

—Buy Side: Max heap (highest price first)
—Sell Side: Min heap (lowest price first)

Matching Rules

Orders match when: Buy Price >= Sell Price

Example:

Buy order:  Price=100, Qty=10
Sell order: Price=99,  Qty=5

Match happens at price=99, quantity=5
Remaining buy: 5 units

Partial Fills

Buy 10 → Sell 4 available
Trade executes for 4
Remaining buy: 6 (stays in book)

Quick Start

With Docker (Recommended)

bash

# Start services
docker-compose up -d

# API available at http://localhost:8000
# Health check: http://localhost:8000/health

Local Development

bash

# Install dependencies
pip install -r requirements.txt

# Start PostgreSQL (or use Docker)
docker run -d --name postgres \
  -e POSTGRES_DB=flashledger \
  -e POSTGRES_USER=flashledger \
  -e POSTGRES_PASSWORD=flashledger \
  -p 5432:5432 \
  postgres:15-alpine

# Run the server
uvicorn app.main:app --reload --port 8000

API Endpoints

POST /api/v1/order

Submit a new order.

Request:

json

{
  "user_id": "user1",
  "side": "buy",
  "price": 100,
  "quantity": 10
}

Response:

json

{
  "order_id": "uuid",
  "order_status": "matched",
  "trades": [
    {
      "price": 99,
      "quantity": 5,
      "trade_id": "uuid"
    }
  ],
  "remaining_qty": 5,
  "latency_ms": 0.234
}

GET /api/v1/orderbook

Get current order book state.

Response:

json

{
  "symbol": "FLASH",
  "bids": [
    {"price": 99, "quantity": 100, "orders": 3}
  ],
  "asks": [
    {"price": 101, "quantity": 50, "orders": 2}
  ],
  "bid_count": 1,
  "ask_count": 1
}

GET /api/v1/trades

Get recent executed trades.

Response:

json

[
  {
    "trade_id": "uuid",
    "buy_order_id": "uuid",
    "sell_order_id": "uuid",
    "buyer_id": "user1",
    "seller_id": "user2",
    "price": 100,
    "quantity": 10,
    "timestamp": "2024-01-01T12:00:00"
  }
]

DELETE /api/v1/order/{order_id}

Cancel an existing order.

GET /api/v1/metrics

Get engine metrics.

Project Structure

flashledger/
├── app/
│   ├── main.py              # FastAPI entry point
│   ├── engine/
│   │   ├── order_book.py    # Heap-based order book
│   │   └── matching_engine.py  # Price-time priority matching
│   ├── api/
│   │   └── routes.py        # REST endpoints
│   └── db/
│       ├── models.py        # Trade model
│       └── database.py      # PostgreSQL connection
├── tests/
│   ├── test_order_book.py
│   ├── test_matching_engine.py
│   └── load_test.py         # 10k order benchmark
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
└── README.md

Testing

Unit Tests

bash

pytest tests/ -v

Load Test (10k Orders)

bash

# Start the server first
python tests/load_test.py --orders 10000 --concurrency 100

Sample Output:

Orders: 10000
Duration: 2.5 seconds
Orders/second: 4000
Median latency: 2.1 ms
P95 latency: 4.2 ms

✅ PASS: Median latency (2.1ms) < 5ms target

Data Flow Example

1.User A submits: BUY 10 @ $100
2.Order rests in buy book (no match)
3.User B submits: SELL 5 @ $99
4.Match found: Buy $100 >= Sell $99
5.Trade executed: 5 units @ $99
6.User A's order: 5 remaining in book
7.User B's order: fully filled
8.Trade persisted to PostgreSQL

MVP Success Criteria

Criteria	Status
Order submission works	✅
Orders match correctly	✅
Partial fills work	✅
Order book updates correctly	✅
Trades persist in database	✅
< 5ms local latency	✅

Tech Stack

—Framework: FastAPI (Python 3.11)
—Database: PostgreSQL 15
—ORM: SQLAlchemy 2.0 (async)
—Container: Docker Compose

License

MIT