OHLCV Python Connector

Kết nối và stream dữ liệu nến (OHLCV) 1 phút thời gian thực từ nhiều sàn giao dịch tiền mã hóa với độ tin cậy và hiệu năng cao, áp dụng Clean Architecture.

Tổng quan nhanh

Sàn hỗ trợ: Binance (Spot, USDT-M, Coin-M), OKX (Spot, Swap), Bybit (Spot, Linear, Inverse), Gate.io (Spot, USDT-M, Coin-M), Hyperliquid (Perp, Spot).
Kiến trúc: phân lớp domain/application/infrastructure, client WebSocket chung, circuit breaker, tự động reconnect + REST backfill, queue hai tầng kèm dedup.
Quan sát: log có cấu trúc, Prometheus metrics, healthcheck HTTP, theo dõi tình trạng kết nối sàn.
Vận hành: graceful shutdown, sẵn sàng K8s, cấu hình linh hoạt, tương thích ngược.

Cài đặt

git clone <repo-url>
cd ohlcv-python-connector
# Pip
pip install -r requirements.txt
# Hoặc Poetry
poetry install

Yêu cầu: Python 3.10+, kết nối mạng tới các endpoint REST/WS của sàn.

Sử dụng nhanh

Chạy CLI stream trực tiếp

# Binance spot
poetry run connector-cli binance BTCUSDT ETHUSDT --market spot --limit 10
# OKX swap
poetry run connector-cli okx BTC-USDT-SWAP ETH-USDT-SWAP --limit 10
# Bybit linear
poetry run connector-cli bybit BTCUSDT ETHUSDT --market linear --limit 10
# Gate.io spot
poetry run connector-cli gateio BTC_USDT ETH_USDT --market spot --limit 10

Chạy WebSocket server

poetry run connector-wss --host 0.0.0.0 --port 8765

Ví dụ client:

const ws = new WebSocket('ws://localhost:8765');
ws.onopen = () => {
  ws.send(JSON.stringify({
    exchange: "binance",
    contract_type: "spot",
    symbols: ["BTCUSDT", "ETHUSDT"],
    limit: 0
  }));
};
ws.onmessage = (event) => console.log(JSON.parse(event.data));

Healthcheck và Metrics

curl http://localhost:8766/health
curl http://localhost:8766/ready | jq
curl http://localhost:8766/metrics

Kiểm thử

Smoke test nhập khẩu: python test_imports.py
Test từng sàn: poetry run python test/test_<exchange>_*.py
Test batch toàn bộ symbol: PYTHONPATH=src poetry run python test/run_all_ws_subscriptions.py --batch-size 100 --limit 1 --concurrency 10 --message-timeout 20

Cấu trúc dự án

ohlcv-python-connector/
├── src/                 # Mã nguồn chính (config, domain, application, infrastructure, interfaces)
├── test/                # Test đơn lẻ và script integration WS
├── docs/                # Tài liệu chi tiết (Quickstart, Upgrade, Deployment, Mapping, Changelog, ...)
├── requirements.txt / pyproject.toml / poetry.lock
├── start.sh
└── README.md

Cấu hình nhanh

File mẫu: .env.example (tùy biến kết nối, logging, metrics, backoff).
Biến thường dùng (ví dụ): CONNECTOR_WS_HOST, CONNECTOR_WS_PORT, LOG_LEVEL, METRICS_PORT.

Ghi chú vận hành

Tăng độ tin cậy: bật circuit breaker, giữ message-timeout đủ lớn cho các batch symbol thanh khoản thấp.
Quan sát: xuất metrics Prometheus, theo dõi healthcheck để tích hợp liveness/readiness.
Dừng an toàn: gửi tín hiệu SIGTERM/SIGINT, hệ thống sẽ shutdown gracefull.

Tài liệu liên quan

Hướng dẫn nhanh: docs/QUICKSTART.md
Cấu trúc dự án: docs/PROJECT_STRUCTURE.md
Checklist triển khai: docs/DEPLOYMENT_CHECKLIST.md
Tổng quan triển khai: docs/IMPLEMENTATION_SUMMARY.md
Nguồn dữ liệu & mapping: docs/DATA_SOURCE_AND_MAPPING.md
Lịch sử thay đổi: docs/CHANGES.md
Kịch bản test WS: docs/TEST_VERIFICATION.md

When running the WebSocket server, health endpoints are automatically available:

# Liveness check
curl http://localhost:8766/health
# {"status":"healthy","timestamp":"2024-10-30T..."}

# Readiness check with exchange health
curl http://localhost:8766/ready | jq
# {
#   "status": "ready",
#   "exchanges": [
#     {
#       "exchange": "binance",
#       "active_connections": 2,
#       "healthy": true,
#       ...
#     }
#   ]
# }

# Prometheus metrics
curl http://localhost:8766/metrics

Architecture

This project follows Clean Architecture principles with clear separation of concerns:

src/
├── domain/              # Core business entities
│   ├── models.py        # PriceQuote dataclass
│   └── repositories.py  # Repository interfaces
│
├── application/         # Use cases
│   └── use_cases/
│       └── stream_prices.py
│
├── infrastructure/      # Exchange implementations
│   ├── common/          # Shared infrastructure
│   │   ├── client.py              # Base WebSocket client
│   │   ├── circuit_breaker.py     # Fault tolerance
│   │   ├── quote_queue.py         # Dual-pipeline queue
│   │   ├── deduplicator.py        # Message deduplication
│   │   ├── rest_pool.py           # Connection pooling
│   │   └── shutdown.py            # Graceful shutdown
│   │
│   ├── binance/         # Binance connector
│   ├── okx/             # OKX connector
│   ├── bybit/           # Bybit connector
│   ├── gateio/          # Gate.io connector
│   └── hyperliquid/     # Hyperliquid connector
│
├── interfaces/          # Entry points
│   ├── cli/
│   │   └── main.py      # CLI interface
│   ├── ws_server/
│   │   └── main.py      # WebSocket server
│   └── health_server.py # Health check HTTP server
│
├── metrics.py           # Metrics collection
└── config.py            # Configuration management

Data Flow

Exchange WebSocket → Circuit Breaker → Deduplicator → Dual Queue → Consumer
                          ↓ (on failure)
                    REST Backfill

Configuration

Environment Variables

Copy .env.example to .env and customize:

cp .env.example .env

Key Configuration Options

Core Settings

CONNECTOR_INACTIVITY_TIMEOUT=3.0          # Seconds before REST backfill
CONNECTOR_MAX_SYMBOL_PER_WS=50            # Symbols per WebSocket connection

Circuit Breaker (Fault Tolerance)

CONNECTOR_CIRCUIT_BREAKER_FAILURE_THRESHOLD=5      # Failures before opening
CONNECTOR_CIRCUIT_BREAKER_RECOVERY_TIMEOUT=30.0    # Base recovery time
CONNECTOR_CIRCUIT_BREAKER_HALF_OPEN_CALLS=1        # Test calls

Queue Management (Memory Safety)

CONNECTOR_CLOSED_QUEUE_MAXSIZE=1000       # Bounded queue for closed candles
CONNECTOR_OPEN_QUEUE_MAXSIZE=0            # 0 = unbounded LIFO stack

Deduplication (Data Quality)

CONNECTOR_DEDUPLICATION_WINDOW_SECONDS=120.0      # Sliding window
CONNECTOR_DEDUPLICATION_MAX_ENTRIES=10000         # Max tracked entries

Connection Pooling (Performance)

CONNECTOR_REST_POOL_CONNECTIONS=10        # Keep-alive connections
CONNECTOR_REST_POOL_MAXSIZE=20            # Max total connections

Health Checks (Monitoring)

CONNECTOR_WSS_HEALTH_CHECK_ENABLED=true   # Enable health endpoints
CONNECTOR_WSS_HEALTH_CHECK_PORT=8766      # Health check port

See .env.example for complete list with documentation.

Monitoring & Observability

Prometheus Metrics

Key metrics exposed on http://localhost:8766/metrics:

Metric	Type	Description
`connector_quotes_processed_total`	Counter	Total quotes processed
`connector_quote_latency_seconds`	Histogram	Processing latency
`connector_active_connections`	Gauge	Active WebSocket connections
`connector_connection_errors_total`	Counter	Connection errors
`connector_circuit_breaker_state`	Gauge	Circuit state (0/1/2)
`connector_queue_depth_closed`	Gauge	Closed queue depth
`connector_queue_blocking_events_total`	Counter	Backpressure events
`connector_duplicates_filtered_total`	Counter	Duplicate quotes

Grafana Dashboard (Example Queries)

# Quote processing rate
rate(connector_quotes_processed_total[1m])

# 95th percentile latency
histogram_quantile(0.95, rate(connector_quote_latency_seconds_bucket[5m]))

# Circuit breaker alerts
connector_circuit_breaker_state > 0

# Backpressure detection
rate(connector_queue_blocking_events_total[1m]) > 0

Health Endpoints

Liveness: GET /health

{
  "status": "healthy",
  "timestamp": "2024-10-30T12:34:56.789Z"
}

Readiness: GET /ready

{
  "status": "ready",
  "timestamp": "2024-10-30T12:34:56.789Z",
  "exchanges": [
    {
      "exchange": "binance",
      "contract_type": "spot",
      "active_connections": 2,
      "last_message_time": "2024-10-30T12:34:50Z",
      "total_quotes": 1523,
      "total_errors": 0,
      "circuit_state": "closed",
      "healthy": true
    }
  ]
}

Deployment

Docker

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8765 8766

CMD ["python", "-m", "interfaces.ws_server.main"]

# Build
docker build -t ohlcv-connector:latest .

# Run
docker run -p 8765:8765 -p 8766:8766 \
  -e CONNECTOR_WSS_HEALTH_CHECK_ENABLED=true \
  ohlcv-connector:latest

Docker Compose

version: '3.8'
services:
  connector:
    build: .
    ports:
      - "8765:8765"
      - "8766:8766"
    environment:
      - CONNECTOR_CIRCUIT_BREAKER_FAILURE_THRESHOLD=5
      - CONNECTOR_CLOSED_QUEUE_MAXSIZE=1000
      - CONNECTOR_WSS_HEALTH_CHECK_ENABLED=true
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8766/health"]
      interval: 30s
      timeout: 10s
      retries: 3
    restart: unless-stopped

Kubernetes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ohlcv-connector
spec:
  replicas: 2
  selector:
    matchLabels:
      app: ohlcv-connector
  template:
    metadata:
      labels:
        app: ohlcv-connector
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8766"
        prometheus.io/path: "/metrics"
    spec:
      containers:
      - name: connector
        image: ohlcv-connector:latest
        ports:
        - containerPort: 8765
          name: websocket
        - containerPort: 8766
          name: health
        env:
        - name: CONNECTOR_WSS_HEALTH_CHECK_ENABLED
          value: "true"
        livenessProbe:
          httpGet:
            path: /health
            port: 8766
          initialDelaySeconds: 10
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /ready
            port: 8766
          initialDelaySeconds: 5
          periodSeconds: 10
        resources:
          requests:
            memory: "256Mi"
            cpu: "200m"
          limits:
            memory: "512Mi"
            cpu: "500m"

Testing

Run All Tests

# Unit tests
poetry run pytest

# Specific exchange test
poetry run python test/test_binance_btc_spot.py

# Batch test all exchanges
poetry run python test/run_all_ws_subscriptions.py \
  --batch-size 100 \
  --limit 60 \
  --concurrency 24

Code Quality

# Linting
poetry run ruff check src

# Type checking
poetry run mypy src

Documentation

Comprehensive documentation is available in the /docs directory:

Document	Description
Quick Start Guide	5-minute setup with examples
Upgrade Guide	Detailed feature documentation
Deployment Checklist	Step-by-step deployment
Implementation Summary	Technical details
Changelog	Version history
Data Sources & Mapping	Exchange API details

Advanced Usage

Custom Configuration for High Load

# .env for 1000+ symbols
CONNECTOR_CLOSED_QUEUE_MAXSIZE=2000
CONNECTOR_OPEN_QUEUE_MAXSIZE=10000
CONNECTOR_MAX_SYMBOL_PER_WS=100
CONNECTOR_REST_POOL_MAXSIZE=50
CONNECTOR_REST_POOL_CONNECTIONS=20

Programmatic Usage

from application.use_cases.stream_prices import StreamPrices
from interfaces.repository_factory import build_price_feed_repository

# Build repository
repository = build_price_feed_repository("binance", "spot")

# Create use case
use_case = StreamPrices(repository)

# Stream quotes
async for quote in use_case.execute(["BTCUSDT", "ETHUSDT"]):
    print(f"{quote.symbol}: {quote.close} @ {quote.timestamp}")

Reliability Features

Circuit Breaker

Automatically opens after 5 consecutive failures, preventing resource waste:

Normal → CLOSED (allow all)
  ↓ (5 failures)
Failing → OPEN (block all, wait 30s)
  ↓ (timeout)
Testing → HALF_OPEN (allow 1 test)
  ↓ (success)        ↓ (failure)
Back to CLOSED    Back to OPEN (wait 60s)

Dual-Pipeline Queue

Memory-safe message routing:

Closed candles (is_closed=True) → Bounded queue (backpressure enforced)
Open candles (is_closed=False) → LIFO stack (latest data priority)

Consumer always drains closed candles first, ensuring completed data is prioritized.

Deduplication

Prevents duplicate processing during reconnections:

Tracks (symbol, timestamp) for 120 seconds
Automatic cleanup of old entries
Typically filters 0.01% duplicates in production

Performance Benchmarks

Metric	v0.1.0	v0.2.0	Improvement
Memory (1000 symbols)	850 MB	320 MB	62% reduction
JSON parse time	12 ms	3 ms	4x faster
REST backfill	850 ms	180 ms	4.7x faster
Duplicate rate	5-15%	<0.1%	99%+ reduction
Connection reuse	0%	90%+	New capability

Contributing

This is an internal project, but improvements are welcome:

Follow Clean Architecture principles
Add tests for new features
Update documentation
Run linting: poetry run ruff check src
Run type checking: poetry run mypy src

License

MIT License - See LICENSE file for details.

Exchange API References

Support

For issues or questions:

Check documentation
Review troubleshooting guide
Check health endpoints: curl http://localhost:8766/ready
Review logs for errors

Roadmap

Version: 0.2.0 Status: Production-Ready Architecture: Clean Architecture Language: Python 3.10+

Built with for high-load, business-critical cryptocurrency market data streaming.

Name		Name	Last commit message	Last commit date
src		src
test		test
.env.example		.env.example
.gitignore		.gitignore
README.md		README.md
poetry.lock		poetry.lock
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt
start.sh		start.sh

Latest commit

History

Folders and files

Repository files navigation