data_layer

data_layer is the market-data gateway for Bobby's trading stack. It centralizes provider connectivity, historical warmup, live streaming, fallback recovery, and service-to-service data contracts so alpha and execution services do not connect to exchanges directly.

The released V2 data plane is the primary market-data contract for its declared Binance USD-M and OKX Swap consumer demand. Python owns provider/API adapters and stable public contracts; Rust owns canonical identity, ordering, idempotency, L2 state and realtime normalization. V1 remains an explicit, observable rollback/compatibility route. VN/DNSE is separately governed and remains V1-primary until its market-hours certification is complete.

It currently serves:

• Binance crypto spot and USD-M futures market data.
• Binance derivatives metrics for basis-arbitrage research and execution.
• VN stock and derivative data through DNSE/vnstock integrations.
• Redis Pub/Sub streams for live consumers.
• REST endpoints for warmup, recovery, diagnostics, and health checks.

Quick Links

• Integration guide
• Fund-grade implementation tracker
• Fund-grade architecture and migration guide
• OKX V5 market-data implementation guide
• V2.0.13 release notes
• V2.0.13 machine-readable certificate
• Contributing guide
• Security policy
• Code of conduct
• License

Repository Policy

main is treated as the protected release branch. Do not commit directly to main.

Use this flow:

git checkout dev
git pull origin dev
git checkout -b feature/my-change

Open pull requests into dev; merge dev into main only through a release pull request after tests and smoke checks pass.

What This Service Owns

• Provider connections and retry/backoff policy.
• Normalized service contracts for downstream systems.
• Redis live-stream publication.
• VN preload parquet storage and materialized views.
• Latest-state recovery endpoints.
• Diagnostics for provider health and data freshness.

What This Service Does Not Own

• Trading decisions.
• Portfolio, risk, order routing, or broker account state.
• Alpha-specific signal logic.
• Direct broker execution.
• Long-term storage for ephemeral Binance derivatives metrics unless a design note explicitly approves it.

Features

• V2 stable data plane — manifest-authorized, mTLS/JWT V2 query and signed-cursor stream for Binance USD-M and OKX Swap
• Rust canonical core — provider-neutral identity, decimal/unit normalization, event idempotency, sequencing, gap/resync and L2 book state
• Durable projection — Kafka-compatible raw/canonical event planes with bounded SQLite/Redis projection, replay and typed freshness/quality state
• Execution-grade context — final BAR, TRADE, QUOTE, top-100 BOOK_SNAPSHOT/BOOK_DELTA and strict MARK_INDEX_PRICE for declared consumer demand
• Reference data — bounded provider wrappers for funding, OI, long/short, taker flow, mark/index, contract metadata and native/continuous basis
• Real-time streaming — WebSocket multiplexer for Binance (spot & futures trade + kline) and DNSE (VN stock live quotes)
• Automatic failover — DNSE as primary VN source, vnstock REST poller as secondary fallback
• Redis Pub/Sub distribution — single upstream connection shared across many downstream consumers
• Historical warmup (VN) — Parquet-backed preload service for 1-minute OHLCV candle warmup
• Binance derivatives REST wrappers — OHLCV, funding, open interest, long/short ratios, taker ratio, depth, and basis bundle endpoints
• Preload watchdog — auto-refreshes VN candle data during market hours, sleeps until next open otherwise
• ⚡ High-performance serialization — orjson throughout for minimal latency
• Alpha strategy example — moving-average crossover strategy included as a reference implementation
• 🐳 Docker-first — full Docker Compose stack with networking, volumes, and log management

V2 Stable Architecture

 Binance USD-M WS/REST                 OKX Swap WS/REST
          |                                      |
          +------------ Python venue adapters ---+
                       retry / pacing / native parsing
                                      |
                                      v
                    Rust canonical realtime core
      identity + decimals + event IDs + sequence + L2 state + gap/resync
                                      |
                       raw Kafka -> canonical Kafka
                                      |
                                      v
                  bounded durable projection layer
                   SQLite cache + Redis coordination
                                      |
                 +--------------------+--------------------+
                 |                                         |
                 v                                         v
         V2 query replicas                         V2 stream replicas
       mTLS/JWT, typed reads                  signed cursor replay -> LIVE
                 |                                         |
                 +--------------------+--------------------+
                                      v
                      qdl_sdk -> Trading System / alpha

 V1 FastAPI + Redis Pub/Sub remains an explicit compatibility and rollback path.
 VN/DNSE remains separately governed on V1 until its own V2 market-hours gate.

The consumer manifest is the authority for venue, canonical instrument, native symbol, feed, interval, depth, freshness and fallback policy. A client cannot use a provider adapter, another consumer's identity, a direct venue connection or a stale/gapped event as a shortcut around that contract.

V2 Data Products And Intended Use

Product	Delivery	Intended use
Final BAR	Durable query/history and signed stream	Closed-candle signal calculation only
TRADE, QUOTE	Durable query plus hot signed stream	Current market context and execution monitoring
BOOK_SNAPSHOT, BOOK_DELTA	Snapshot bootstrap/resync plus ordered delta stream	Limit/post-only price selection and market-impact analysis
MARK_INDEX_PRICE	Bounded provider-authentic reference read	Declared trigger/risk reference only; stale values are rejected
Funding, OI, long/short, taker flow, metadata, basis	Bounded reference:batch provider reads	Research/feature/diagnostic input under a manifest requirement

For the released Binance/OKX scope, the full certificate covers 299 consumer products: 234 durable and 65 on-demand, including 150 BAR requirements over 140 physical final-BAR bindings, 24 TRADE, 20 QUOTE, 20 BOOK_SNAPSHOT, 20 BOOK_DELTA, 20 MARK_INDEX_PRICE and 45 other reference products. The V2.0.13 patch additionally proves the five-liquid Trading System paper scope for BTC, ETH, SOL, DOGE and BNB across both venues.

Certified V2 Benchmark Snapshot

These figures are real-provider/no-order evidence, not a blanket network SLA. Request round-trip is consumer -> V2 query/cache -> consumer; it is distinct from provider event age and from final-BAR materialization. The durable price/ bar query buckets below contain 30 reads each: five liquid symbols, three samples, two query replicas, exact identity, complete=true and no open gap.

V2 durable read	p50	p95	p99
Binance USD-M TRADE	7.52ms	22.03ms	40.52ms
Binance USD-M QUOTE	8.83ms	38.84ms	45.03ms
Binance USD-M final BAR 1m query	13.73ms	55.64ms	64.61ms
OKX Swap TRADE	6.70ms	29.02ms	31.65ms
OKX Swap QUOTE	6.65ms	31.26ms	41.43ms
OKX Swap final BAR 1m query	534.35ms	1182.97ms	1196.38ms

• A warmup:batch of 10 x 100 = 1,000 final 1m bars completed in 813.88ms and 952.09ms on the two replicas. Warm up once, retain a bounded FIFO buffer, then append/deduplicate closed bars; do not poll full history per signal cycle.
• At a real 1m close, close-to-final-BAR availability was p50 2.151s, p95 3.376s, p99 4.196s, max 4.400s. The active bar edge has a 0.10s settlement start and 1.0s bounded retry; the remaining time is provider finality plus durable publication, not an intentional 10-second sleep.
• warmup_then_stream TRADE handoff measured p50 294.66ms, p95 908.86ms, p99 994.62ms; durable event age was p50 342.57ms, p95 406.12ms, p99 407.23ms. A quiet but live session is typed QUIET, not misclassified as a disconnected provider.
• The SOL/OKX MARK_INDEX_PRICE repair measured source age 39ms/70ms and receive age 3ms/9ms at the two V2 query replicas. The 2,000ms execution freshness boundary is strict: 2,001ms is typed STALE and not execution eligible.

BOOK_SNAPSHOT and BOOK_DELTA are certified for top-100 depth and sequence/ gap/resync correctness across the five-liquid Binance/OKX books. A separate raw WebSocket-ingress p99 and per-reference-metric p99 are not published SLOs yet; callers must rely on typed freshness, completeness and gap state rather than infer those figures from this benchmark.

See the implementation journal, V2.0.12 full certificate and V2.0.13 patch certificate for scope, provenance and exclusions.

V1 Compatibility Architecture

                    ┌─────────────────────────────────────────┐
                    │            data_layer service            │
                    │                                          │
  Binance WS ──────►│  Async WS Multiplexer                   │
  (trade + kline)   │  (spot & futures)                       │
                    │            │                             │
  DNSE WS ─────────►│  DNSE Stream Manager  ──────────────────►  Redis Pub/Sub
  (VN primary)      │  (primary VN source)                    │  stream:trade:{symbol}
                    │            │                             │  stream:kline:{interval}:{symbol}
  vnstock REST ─────►│  VN Poller            ──────────────────►  stream:vn:{symbol}
  (VN fallback)     │  (fallback if DNSE stale)               │
                    │                                          │
                    │  Preload Watchdog                        │  REST API
                    │  (Parquet 1m OHLCV)  ──────────────────►  GET /v1/preload/{symbol}
                    └─────────────────────────────────────────┘

Downstream services (alpha strategies, paper trading engines, execution services) connect to:

• V2 (preferred where the manifest admits it): deployment-provided V2 query/stream endpoints through qdl_sdk, mTLS/JWT and signed cursors
• V1 compatibility: Redis Pub/Sub plus REST API (http://data_layer:8100) for legacy warmup, latest-state recovery and health checks

Tech Stack

Component	Technology
Framework	FastAPI + Uvicorn
Cache & Streaming	Redis 5+ with hiredis
Data Processing	Pandas + PyArrow (Parquet)
Serialization	orjson
Canonical Realtime Core	Rust (qdl-core, qdl-realtime-core, qdl-kafka)
Durable Event Backbone	Kafka-compatible raw/canonical topics + SQLite/Redis bounded projection
V2 Security	mTLS workload identity + JWT manifests + signed replay cursors
VN Market Data	vnstock + DNSE WebSocket
Package Manager	Poetry
Runtime	Python 3.10+
Container	Docker + Docker Compose

Development Checks

This repo is Docker-first. Prefer container tests over installing Python packages directly on the server:

docker compose run --rm test_runner python -m unittest discover -s tests

Optional pre-commit hooks:

pre-commit install
pre-commit run --all-files

Project Structure

data_layer/
├── app/
│   ├── main.py               # FastAPI app, lifespan, all REST endpoints
│   ├── config.py             # Environment config loader
│   ├── logging_config.py     # Structured logging setup
│   ├── alpha/
│   │   ├── strategy.py       # DataLayerClient + MovingAverageCrossAlpha example
│   │   └── run_alpha.py      # Alpha service entrypoint
│   ├── cache/
│   │   └── redis_cache.py    # Redis abstraction layer
│   ├── database/
│   │   ├── preload.py        # Parquet-based VN candle preload logic
│   │   └── dnse_fallback.py  # DNSE REST fallback for historical data
│   ├── diagnostics/          # Data source health check scripts
│   ├── openapi_sdk/          # Generated DNSE OpenAPI client (Python + JS)
│   └── stream/
│       ├── async_live_feed.py   # Async Binance WS multiplexer
│       ├── binance_ws.py        # Binance WebSocket client
│       ├── dnse_ws.py           # DNSE WebSocket client & stream manager
│       ├── feed_builder.py      # Stream task builder
│       ├── feed_parsers.py      # Normalized payload parsers
│       ├── price_manager.py     # Price aggregation logic
│       └── vnstock_poller.py    # vnstock REST polling fallback
├── tests/
│   ├── test_alpha_strategy.py
│   └── test_data_layer_client.py
├── data/                     # Parquet preload storage (gitignored)
├── logs/                     # Application logs (gitignored)
├── symbols.json              # Binance symbols configuration
├── symbols_vn.yaml           # VN stock symbols configuration
├── docker-compose.yml        # Full service stack
├── Dockerfile
├── pyproject.toml
└── DATA_LAYER_SERVICE_ACCESS_GUIDE.md  # Integration guide for downstream services

Getting Started

Prerequisites

• Docker & Docker Compose
• A running Redis instance accessible as redis_service on bobby_network
• (Optional) DNSE API credentials for VN stock live data
• (Optional) vnstock API key

1. Clone & Configure

git clone <your-repo-url>
cd data_layer
cp .env.example .env

Edit .env and fill in your credentials:

# Required for VN stock live data (DNSE)
DNSE_API_KEY=your_dnse_api_key
DNSE_API_SECRET_KEY=your_dnse_secret
DNSE_API_VERSION=2026-07-23
# Explicit opt-in only when an approved egress proxy is required.
DNSE_REST_USE_ENV_PROXY=false

# Optional: vnstock API key
VNSTOCK_API_KEY=your_vnstock_key

2. Create Docker Networks (if not already present)

docker network create bobby_network
docker network create executor_network

3. Build & Run

docker compose up -d data_layer

The service will be available at http://localhost:8100.

4. Verify

curl http://localhost:8100/v1/health

Expected response:

{
  "status": "ok",
  "redis": true,
  "binance_trade_stream": true,
  "binance_kline_stream": true
}

API Reference

V2 Stable Query And Stream

V2 is the preferred contract for manifest-admitted Binance USD-M and OKX Swap consumers. Use deployment-provided query/stream addresses and qdl_sdk; port 8100 and the V1 endpoints below are not V2 authority.

Method	Endpoint	Purpose
GET	/v2/instruments	Canonical instrument catalog
GET	/v2/market-data/{instrument_uid}/snapshot	Current typed durable snapshot
GET	/v2/market-data/{instrument_uid}/warmup	Bounded final-BAR warmup
GET	/v2/market-data/{instrument_uid}/history	Bounded historical final-BAR read
POST	/v2/market-data/warmup:batch	Multi-instrument warmup
POST	/v2/market-data/reference:batch	Manifest-authorized provider reference data
GET	/v2/feeds/{instrument_uid}/status	Typed freshness, session and gap state
POST	/v2/system/readiness:check	Requirement-level readiness admission

The full client, startup/reconnect sequence, stream cursor behavior and V1 fallback rules are in the production integration guide.

Health

Method	Endpoint	Description
GET	/v1/health	Service health + Redis status

Binance

Method	Endpoint	Description
GET	/v1/binance/price/{symbol}	Latest cached trade price
GET	/v1/binance/kline/{symbol}?interval=1m	Latest cached kline
GET	/v1/binance/klines/{symbol}?interval=1m&limit=500	Historical klines proxy

VN Stock

Method	Endpoint	Description
GET	/v1/vn/quote/{symbol}	Live quote (short TTL)
GET	/v1/vn/quote-last/{symbol}	Last known snapshot (survives market close)
GET	/v1/vn/board	Full VN price board snapshot

Preload (VN Historical Warmup)

Method	Endpoint	Description
GET	/v1/preload/status	Preload inventory & timestamp metadata
GET	/v1/preload/{symbol}?limit=1000	Latest N warmup candles (ascending time)
POST	/v1/preload/run	Trigger full preload for all symbols
POST	/v1/preload/append/{symbol}	Append delta for a single symbol

Redis Stream Channels

Downstream services subscribe to these channels:

Channel	Purpose
stream:trade:{symbol}	Binance live trade price (execution/papertrade)
stream:kline:{interval}:{symbol}	Binance kline/candle data
stream:vn:{symbol}	VN stock live quote

All payloads are serialized with orjson. Clients should use orjson.loads(message['data']).

Quick Subscribe Example (Python)

import redis
import orjson

r = redis.Redis(host='redis_service', port=6379, db=2)
pubsub = r.pubsub()
pubsub.subscribe("stream:trade:BTCUSDT")

for message in pubsub.listen():
    if message['type'] == 'message':
        data = orjson.loads(message['data'])
        print(f"Price: {data['price']}")

Recommended Boot Sequence for Downstream Services

1. Check GET /v1/health — verify data_layer is reachable
2. Fetch GET /v1/preload/{symbol}?limit=N — warm up with historical candles (VN only)
3. Call the latest-state endpoint for your feed to recover any missed updates
4. Subscribe to the appropriate Redis channel for live streaming

See DATA_LAYER_SERVICE_ACCESS_GUIDE.md for the full integration contract.

Normalized Payload Contracts

All feeds emit normalized payloads regardless of provider. See app/stream/feed_parsers.py for the canonical field definitions.

Feed	Key Fields
Binance Trade	symbol, price, quantity, trade_id, event_time, trade_time, side, source
Binance Kline	symbol, open, high, low, close, volume, timestamp, is_closed
VN Quote	symbol, price, quantity, open, high, low, source, timestamp

Configuration

All configuration is via environment variables. See .env.example for the full list.

Variable	Default	Description
REDIS_HOST	redis_service	Redis hostname
REDIS_PORT	6379	Redis port
REDIS_DB	2	Redis database index
DNSE_API_KEY	—	DNSE OpenAPI key
DNSE_API_SECRET_KEY	—	DNSE OpenAPI secret
DNSE_API_VERSION	2026-07-23	Explicit DNSE REST contract revision
DNSE_REST_USE_ENV_PROXY	false	Opt in to an approved environment proxy; TLS verification remains mandatory
VNSTOCK_API_KEY	—	vnstock API key
VNSTOCK_SOURCE	KBS	vnstock data source
VNSTOCK_POLL_INTERVAL	3.0	Polling interval (seconds)
PRELOAD_DIR	/app/data/preload/1m	Parquet storage path
PRELOAD_MONTHS	6	Months of history to preload
BINANCE_SYMBOLS_FILE	/app/symbols.json	Binance symbols config
API_HOST	0.0.0.0	FastAPI bind host
API_PORT	8100	FastAPI bind port

Development

Install Dependencies (local, without Docker)

pip install poetry
poetry install

Run Locally

# Set up environment
cp .env.example .env
# Edit .env to point REDIS_HOST to your local Redis instance

uvicorn app.main:app --host 0.0.0.0 --port 8100 --reload

Run Tests

# Via Docker Compose
docker compose run --rm test_runner

# Or locally
python -m unittest discover tests

Run Alpha Strategy Example

docker compose up -d alpha_service

Run Data Source Diagnostics

docker compose run --rm data_source_checker

Docker Services

Service	Description
data_layer	Main FastAPI data gateway (port 8100)
alpha_service	Moving-average crossover alpha strategy example
test_runner	Unit test runner
data_source_checker	Data source health diagnostics

VN Market Schedule

The preload watchdog respects VN stock exchange hours (UTC+7):

• Morning session: 09:00 – 11:30
• Afternoon session: 13:00 – 14:30
• Weekends: skipped

Contributing

Contributions are welcome! Please:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/your-feature)
3. Commit your changes (git commit -m 'Add your feature')
4. Push to the branch (git push origin feature/your-feature)
5. Open a Pull Request

License

This project is licensed under the MIT License — see the LICENSE file for details.