Mở đầu: Khi dữ liệu tĩnh phá vỡ chiến lược giao dịch
Đêm 3 giờ sáng, hệ thống giao dịch của tôi đột ngột dừng lại. Trên màn hình terminal hiển thị lỗi quen thuộc:
ConnectionError: timeout after 30000ms - Failed to fetch normalized market data
WebSocket connection closed unexpectedly: code=1006, reason=abnormal closure
DataValidationError: price normalization failed - missing required field 'bid_volume'
Đó là khoảnh khắc tôi nhận ra rằng việc phụ thuộc vào các API dữ liệu bên ngoài đang phá vỡ chiến lược giao dịch của mình. Độ trễ 300ms từ nguồn cấp dữ liệu thị trường đồng nghĩa với việc thuật toán của tôi đưa ra quyết định dựa trên dữ liệu đã lỗi thời. Đây là lý do tôi bắt đầu xây dựng hệ thống Tardis Machine - một giải pháp WebSocket cục bộ để chuẩn hóa dữ liệu theo thời gian thực ngay trong hạ tầng của mình.
Tardis Machine là gì và tại sao cần thiết cho đội ngũ quantitative
Tardis Machine là một service WebSocket được thiết kế để giải quyết ba vấn đề cốt lõi trong hệ thống giao dịch định lượng:
- Độ trễ thấp - Xử lý dữ liệu tại chỗ thay vì phụ thuộc vào API bên thứ ba
- Chuẩn hóa nhất quán - Đảm bảo dữ liệu từ nhiều nguồn có định dạng thống nhất
- Khả năng chịu lỗi - Hoạt động độc lập ngay cả khi mất kết nối internet
Trong bài viết này, tôi sẽ chia sẻ cách xây dựng hệ thống này từ đầu, kèm theo các giải pháp cho những lỗi thường gặp mà tôi đã gặp phải trong quá trình triển khai.
Kiến trúc hệ thống Tardis Machine
Sơ đồ luồng dữ liệu
Hệ thống Tardis Machine bao gồm các thành phần chính sau:
┌─────────────────────────────────────────────────────────────────┐
│ TARDIS MACHINE ARCHITECTURE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ WebSocket ┌──────────────────────────┐ │
│ │ Market Feed │ ──────────────► │ Normalization Engine │ │
│ │ (External) │ │ ┌────────────────────┐ │ │
│ └──────────────┘ │ │ - Schema Validator │ │ │
│ │ │ - Type Converter │ │ │
│ ┌──────────────┐ WebSocket │ │ - Unit Normalizer │ │ │
│ │ Raw Data │ ──────────────► │ │ - Time Syncer │ │ │
│ │ Sources │ │ └────────────────────┘ │ │
│ └──────────────┘ └────────────┬─────────────┘ │
│ │ │
│ ┌──────────▼──────────┐ │
│ │ Local Cache │ │
│ │ (Redis/SQLite) │ │
│ └──────────┬──────────┘ │
│ │ │
│ ┌──────────▼──────────┐ │
│ │ WebSocket Server │ │
│ │ (Port 8765) │ │
│ └──────────┬──────────┘ │
│ │ │
│ ┌──────────▼──────────┐ │
│ │ Quant Trading │ │
│ │ Strategies │ │
│ └─────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
Cài đặt môi trường
Trước tiên, hãy thiết lập môi trường phát triển với các dependency cần thiết:
# requirements.txt
Core WebSocket server
websockets==12.0
asyncio==3.4.3
Data processing
pandas==2.1.4
numpy==1.26.2
pydantic==2.5.2
Local cache
redis==5.0.1
Validation & normalization
jsonschema==4.20.0
python-dateutil==2.8.2
Testing
pytest==7.4.3
pytest-asyncio==0.21.1
Monitoring
prometheus-client==0.19.0
# Khởi tạo project
mkdir tardis-machine && cd tardis-machine
python -m venv venv
source venv/bin/activate # Linux/Mac
venv\Scripts\activate # Windows
pip install -r requirements.txt
Tạo cấu trúc thư mục
mkdir -p src/{core,websocket,normalizers,validators}
mkdir -p tests
mkdir -p config
Xây dựng Normalization Engine
1. Định nghĩa Schema dữ liệu chuẩn hóa
Đầu tiên, tôi tạo một schema Pydantic để đảm bảo tất cả dữ liệu sau khi chuẩn hóa đều có cấu trúc nhất quán:
# src/core/schemas.py
from pydantic import BaseModel, Field, validator
from typing import Optional, Dict, Any
from datetime import datetime
from enum import Enum
class MarketDataType(str, Enum):
TRADE = "trade"
QUOTE = "quote"
ORDERBOOK = "orderbook"
TIMESERIES = "timeseries"
class NormalizedMarketData(BaseModel):
"""Schema chuẩn hóa cho tất cả dữ liệu thị trường"""
# Identifiers
symbol: str = Field(..., description="Mã chứng khoán, e.g., 'AAPL'")
exchange: str = Field(..., description="Sàn giao dịch, e.g., 'NASDAQ'")
data_type: MarketDataType
# Timestamps (always UTC, milliseconds)
timestamp: int = Field(..., ge=0, description="Unix timestamp ms")
local_processing_time: int = Field(default_factory=lambda: int(datetime.utcnow().timestamp() * 1000))
# Normalized price fields (always in USD, 8 decimal places)
price: Optional[float] = Field(None, ge=0)
bid_price: Optional[float] = Field(None, ge=0)
ask_price: Optional[float] = Field(None, ge=0)
open_price: Optional[float] = None
close_price: Optional[float] = None
high_price: Optional[float] = None
low_price: Optional[float] = None
# Volume (always shares, integer)
volume: Optional[int] = Field(None, ge=0)
bid_volume: Optional[int] = Field(None, ge=0)
ask_volume: Optional[int] = Field(None, ge=0)
# Metadata
source: str = Field(..., description="Nguồn dữ liệu gốc")
raw_fields: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Dữ liệu gốc để debug")
@validator('timestamp')
def validate_timestamp(cls, v):
"""Timestamp phải trong khoảng hợp lệ (sau 2000-01-01, trước now + 1 ngày)"""
min_ts = int(datetime(2000, 1, 1).timestamp() * 1000)
max_ts = int(datetime.utcnow().timestamp() * 1000) + 86400000
if not min_ts <= v <= max_ts:
raise ValueError(f'Timestamp {v} outside valid range')
return v
@property
def datetime(self) -> datetime:
"""Convert timestamp ms to datetime"""
return datetime.utcfromtimestamp(self.timestamp / 1000)
@property
def spread(self) -> Optional[float]:
"""Tính spread nếu có bid/ask"""
if self.bid_price and self.ask_price:
return round(self.ask_price - self.bid_price, 8)
return None
@property
def mid_price(self) -> Optional[float]:
"""Tính mid price nếu có bid/ask"""
if self.bid_price and self.ask_price:
return round((self.bid_price + self.ask_price) / 2, 8)
return None
2. Triển khai Normalization Engine
Đây là phần cốt lõi - engine xử lý và chuẩn hóa dữ liệu từ nhiều nguồn khác nhau:
# src/normalizers/base_normalizer.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from src.core.schemas import NormalizedMarketData, MarketDataType
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
class BaseNormalizer(ABC):
"""Abstract base class cho tất cả normalizers"""
def __init__(self, source_name: str):
self.source_name = source_name
self.metrics = {
'processed': 0,
'failed': 0,
'latency_ms': []
}
@abstractmethod
def normalize(self, raw_data: Dict[str, Any]) -> Optional[NormalizedMarketData]:
"""Chuẩn hóa dữ liệu thô thành NormalizedMarketData"""
pass
def _extract_timestamp(self, data: Dict[str, Any], time_fields: list) -> Optional[int]:
"""Trích xuất timestamp từ nhiều định dạng khác nhau"""
for field in time_fields:
if field in data:
ts = data[field]
# Unix milliseconds
if isinstance(ts, int) and ts > 1_000_000_000_000:
return ts
# Unix seconds
elif isinstance(ts, (int, float)) and ts > 1_000_000_000:
return int(ts * 1000)
# ISO string
elif isinstance(ts, str):
try:
dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
return int(dt.timestamp() * 1000)
except ValueError:
continue
# datetime object
elif isinstance(ts, datetime):
return int(ts.timestamp() * 1000)
return None
def _normalize_price(self, value: Any, decimals: int = 8) -> Optional[float]:
"""Chuẩn hóa giá về float với số thập phân cố định"""
if value is None:
return None
try:
return round(float(value), decimals)
except (ValueError, TypeError):
logger.warning(f"Cannot normalize price: {value}")
return None
def _normalize_volume(self, value: Any) -> Optional[int]:
"""Chuẩn hóa volume về integer"""
if value is None:
return None
try:
return int(float(value))
except (ValueError, TypeError):
logger.warning(f"Cannot normalize volume: {value}")
return None
src/normalizers/binance_normalizer.py
from src.normalizers.base_normalizer import BaseNormalizer
from src.core.schemas import NormalizedMarketData, MarketDataType
from typing import Dict, Any, Optional
import logging
logger = logging.getLogger(__name__)
class BinanceNormalizer(BaseNormalizer):
"""Normalizer cho dữ liệu từ Binance WebSocket API"""
# Mapping symbol Binance -> Standard format
SYMBOL_MAP = {
'BTCUSDT': 'BTC-USD',
'ETHUSDT': 'ETH-USD',
'BNBUSDT': 'BNB-USD',
# Thêm các mapping khác tùy nhu cầu
}
# Inverse mapping
SYMBOL_REVERSE_MAP = {v: k for k, v in SYMBOL_MAP.items()}
def __init__(self):
super().__init__("binance")
def normalize(self, raw_data: Dict[str, Any]) -> Optional[NormalizedMarketData]:
"""Chuẩn hóa dữ liệu Binance"""
try:
self.metrics['processed'] += 1
# Xác định loại dữ liệu từ event type
event_type = raw_data.get('e', '')
if event_type == 'trade':
return self._normalize_trade(raw_data)
elif event_type == 'depthUpdate':
return self._normalize_orderbook(raw_data)
elif event_type == 'kline':
return self._normalize_ohlcv(raw_data)
else:
logger.debug(f"Unknown event type: {event_type}")
return None
except Exception as e:
self.metrics['failed'] += 1
logger.error(f"Normalization failed: {e}", exc_info=True)
return None
def _normalize_trade(self, data: Dict[str, Any]) -> NormalizedMarketData:
"""Chuẩn hóa trade event"""
symbol = self.SYMBOL_MAP.get(data['s'], data['s'])
return NormalizedMarketData(
symbol=symbol,
exchange='BINANCE',
data_type=MarketDataType.TRADE,
timestamp=data['T'], # Trade timestamp in ms
price=self._normalize_price(data['p']),
volume=self._normalize_volume(data['q']),
source=self.source_name,
raw_fields={'event': data}
)
def _normalize_orderbook(self, data: Dict[str, Any]) -> NormalizedMarketData:
"""Chuẩn hóa orderbook update"""
symbol = self.SYMBOL_MAP.get(data['s'], data['s'])
bids = data.get('b', [])[:1] # Best bid
asks = data.get('a', [])[:1] # Best ask
return NormalizedMarketData(
symbol=symbol,
exchange='BINANCE',
data_type=MarketDataType.ORDERBOOK,
timestamp=data['E'],
bid_price=self._normalize_price(bids[0][0]) if bids else None,
bid_volume=self._normalize_volume(bids[0][1]) if bids else None,
ask_price=self._normalize_price(asks[0][0]) if asks else None,
ask_volume=self._normalize_volume(asks[0][1]) if asks else None,
source=self.source_name,
raw_fields={'event': data}
)
def _normalize_ohlcv(self, data: Dict[str, Any]) -> NormalizedMarketData:
"""Chuẩn hóa OHLCV (Kline) data"""
kline = data['k']
symbol = self.SYMBOL_MAP.get(kline['s'], kline['s'])
return NormalizedMarketData(
symbol=symbol,
exchange='BINANCE',
data_type=MarketDataType.TIMESERIES,
timestamp=kline['t'], # Kline start time
open_price=self._normalize_price(kline['o']),
high_price=self._normalize_price(kline['h']),
low_price=self._normalize_price(kline['l']),
close_price=self._normalize_price(kline['c']),
volume=self._normalize_volume(kline['v']),
source=self.source_name,
raw_fields={'event': data}
)
3. WebSocket Server với xử lý kết nối
Bây giờ, hãy xây dựng WebSocket server để phục vụ dữ liệu đã chuẩn hóa cho các chiến lược giao dịch:
# src/websocket/server.py
import asyncio
import json
import logging
from typing import Set, Dict, Optional
from websockets.server import WebSocketServerProtocol, serve
from websockets import WebSocketServerProtocol
import time
from src.normalizers.binance_normalizer import BinanceNormalizer
from src.core.schemas import NormalizedMarketData
from src.core.cache import LocalCache
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class TardisMachineServer:
"""Tardis Machine WebSocket Server - Xử lý và phân phối dữ liệu chuẩn hóa"""
def __init__(self, host: str = "0.0.0.0", port: int = 8765):
self.host = host
self.port = port
# Connected clients
self.clients: Set[WebSocketServerProtocol] = set()
# Normalizers cho các nguồn dữ liệu khác nhau
self.normalizers = {
'binance': BinanceNormalizer(),
}
# Local cache
self.cache = LocalCache()
# Metrics
self.metrics = {
'total_messages': 0,
'clients_connected': 0,
'messages_per_second': 0,
'avg_processing_time_ms': 0
}
# Shutdown flag
self._shutdown = False
# Processing stats
self._message_timestamps = []
async def register(self, websocket: WebSocketServerProtocol):
"""Đăng ký client mới"""
self.clients.add(websocket)
self.metrics['clients_connected'] = len(self.clients)
logger.info(f"Client connected: {websocket.remote_address}. Total: {self.metrics['clients_connected']}")
async def unregister(self, websocket: WebSocketServerProtocol):
"""Hủy đăng ký client"""
self.clients.discard(websocket)
self.metrics['clients_connected'] = len(self.clients)
logger.info(f"Client disconnected. Total: {self.metrics['clients_connected']}")
async def broadcast(self, message: str):
"""Gửi message đến tất cả clients đang kết nối"""
if not self.clients:
return
# Tính toán messages per second
current_time = time.time()
self._message_timestamps = [t for t in self._message_timestamps if current_time - t < 1]
self._message_timestamps.append(current_time)
self.metrics['messages_per_second'] = len(self._message_timestamps)
# Broadcast đến tất cả clients
disconnected = set()
for client in self.clients:
try:
await client.send(message)
except Exception as e:
logger.warning(f"Failed to send to {client.remote_address}: {e}")
disconnected.add(client)
# Cleanup disconnected clients
for client in disconnected:
self.clients.discard(client)
async def handle_raw_data(self, raw_message: str, source: str = 'binance'):
"""Xử lý dữ liệu thô và broadcast dữ liệu chuẩn hóa"""
start_time = time.time()
try:
data = json.loads(raw_message)
# Lấy normalizer phù hợp
normalizer = self.normalizers.get(source)
if not normalizer:
logger.warning(f"No normalizer for source: {source}")
return
# Normalize dữ liệu
normalized = normalizer.normalize(data)
if normalized:
# Cập nhật cache
await self.cache.set(normalized.symbol, normalized)
# Broadcast cho clients
await self.broadcast(normalized.model_dump_json())
self.metrics['total_messages'] += 1
# Cập nhật avg processing time
process_time = (time.time() - start_time) * 1000
current_avg = self.metrics['avg_processing_time_ms']
self.metrics['avg_processing_time_ms'] = (
(current_avg * (self.metrics['total_messages'] - 1) + process_time)
/ self.metrics['total_messages']
)
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON: {e}")
except Exception as e:
logger.error(f"Error processing message: {e}", exc_info=True)
async def handle_client(self, websocket: WebSocketServerProtocol):
"""Xử lý kết nối từ client"""
await self.register(websocket)
try:
async for message in websocket:
# Client có thể gửi:
# 1. raw market data để normalize
# 2. control messages: subscribe, unsubscribe, ping
if message == "ping":
await websocket.send("pong")
continue
# Xử lý dữ liệu thị trường
await self.handle_raw_data(message)
except WebSocketServerProtocol as e:
if e.code == 1006:
logger.warning(f"Abnormal closure: {websocket.remote_address}")
else:
logger.error(f"WebSocket error: {e}")
finally:
await self.unregister(websocket)
async def handle_external_feed(self):
"""Kết nối đến các nguồn dữ liệu bên ngoài (ví dụ: Binance)"""
import websockets
binance_url = "wss://stream.binance.com:9443/ws/btcusdt@trade"
while not self._shutdown:
try:
async with websockets.connect(binance_url) as ws:
logger.info(f"Connected to Binance feed")
async for message in ws:
if self._shutdown:
break
await self.handle_raw_data(message, source='binance')
except websockets.exceptions.ConnectionClosed as e:
logger.warning(f"Binance connection closed: {e.code} {e.reason}")
except Exception as e:
logger.error(f"Binance feed error: {e}")
# Reconnect sau 5 giây
if not self._shutdown:
logger.info("Reconnecting to Binance in 5 seconds...")
await asyncio.sleep(5)
async def metrics_server(self):
"""Server metrics endpoint"""
from aiohttp import web
async def get_metrics(request):
return web.json_response(self.metrics)
app = web.Application()
app.router.add_get('/metrics', get_metrics)
app.router.add_get('/health', lambda r: web.json_response({'status': 'ok'}))
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, self.host, 8080)
await site.start()
logger.info(f"Metrics server running on http://{self.host}:8080")
async def start(self):
"""Khởi động Tardis Machine server"""
logger.info(f"Starting Tardis Machine on {self.host}:{self.port}")
# Chạy external feed handler
feed_task = asyncio.create_task(self.handle_external_feed())
# Chạy metrics server
metrics_task = asyncio.create_task(self.metrics_server())
# Chạy WebSocket server
async with serve(self.handle_client, self.host, self.port):
logger.info(f"WebSocket server ready on ws://{self.host}:{self.port}")
while not self._shutdown:
await asyncio.sleep(1)
await feed_task
await metrics_task
def stop(self):
"""Dừng server"""
logger.info("Shutting down Tardis Machine...")
self._shutdown = True
Chạy server
if __name__ == "__main__":
server = TardisMachineServer()
try:
asyncio.run(server.start())
except KeyboardInterrupt:
server.stop()
Kết nối từ Chiến lược Giao dịch
Sau đây là cách strategies của bạn kết nối với Tardis Machine:
# Ví dụ: Strategy kết nối với Tardis Machine
import asyncio
import websockets
import json
from typing import Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
@dataclass
class MarketSnapshot:
"""Snapshot dữ liệu thị trường"""
symbol: str
price: float
timestamp: int
source: str
@property
def age_ms(self) -> int:
"""Độ trễ của dữ liệu"""
return int(datetime.utcnow().timestamp() * 1000) - self.timestamp
class TradingStrategy:
"""Base class cho trading strategies"""
def __init__(self, tardis_url: str = "ws://localhost:8765"):
self.tardis_url = tardis_url
self.websocket = None
self.latest_data: Dict[str, MarketSnapshot] = {}
self.max_latency_ms = 100 # Reject data older than 100ms
async def connect(self):
"""Kết nối đến Tardis Machine"""
self.websocket = await websockets.connect(self.tardis_url)
print(f"Connected to Tardis Machine at {self.tardis_url}")
async def disconnect(self):
"""Ngắt kết nối"""
if self.websocket:
await self.websocket.close()
async def receive_market_data(self):
"""Nhận dữ liệu từ Tardis Machine"""
async for message in self.websocket:
try:
data = json.loads(message)
# Kiểm tra latency
snapshot = MarketSnapshot(
symbol=data['symbol'],
price=data['price'],
timestamp=data['timestamp'],
source=data['source']
)
if snapshot.age_ms > self.max_latency_ms:
print(f"[WARN] Data too old: {snapshot.age_ms}ms for {snapshot.symbol}")
continue
self.latest_data[snapshot.symbol] = snapshot
# Gọi strategy logic
await self.on_market_data(snapshot)
except json.JSONDecodeError:
print(f"[ERROR] Invalid JSON: {message[:100]}")
except KeyError as e:
print(f"[ERROR] Missing field: {e}")
async def on_market_data(self, snapshot: MarketSnapshot):
"""Override this in your strategy"""
pass
Ví dụ: Simple momentum strategy
class MomentumStrategy(TradingStrategy):
"""Chiến lược momentum đơn giản"""
def __init__(self, tardis_url: str = "ws://localhost:8765"):
super().__init__(tardis_url)
self.price_history: Dict[str, list] = {}
self.window_size = 10
async def on_market_data(self, snapshot: MarketSnapshot):
"""Logic momentum"""
# Update history
if snapshot.symbol not in self.price_history:
self.price_history[snapshot.symbol] = []
self.price_history[snapshot.symbol].append({
'price': snapshot.price,
'timestamp': snapshot.timestamp
})
# Keep only window
self.price_history[snapshot.symbol] = self.price_history[snapshot.symbol][-self.window_size:]
# Calculate momentum
if len(self.price_history[snapshot.symbol]) >= self.window_size:
prices = [p['price'] for p in self.price_history[snapshot.symbol]]
momentum = (prices[-1] - prices[0]) / prices[0] * 100
if momentum > 1: # Uptrend
print(f"[LONG] {snapshot.symbol}: momentum={momentum:.2f}%, price={snapshot.price}")
elif momentum < -1: # Downtrend
print(f"[SHORT] {snapshot.symbol}: momentum={momentum:.2f}%, price={snapshot.price}")
async def main():
"""Chạy strategy"""
strategy = MomentumStrategy()
try:
await strategy.connect()
await strategy.receive_market_data()
except KeyboardInterrupt:
print("\nShutting down...")
finally:
await strategy.disconnect()
if __name__ == "__main__":
asyncio.run(main())
Kiểm thử với dữ liệu mẫu
Trước khi chạy production, hãy kiểm thử với dữ liệu mẫu:
# tests/test_normalization.py
import pytest
import json
from src.normalizers.binance_normalizer import BinanceNormalizer
from src.core.schemas import MarketDataType
class TestBinanceNormalizer:
"""Unit tests cho Binance normalizer"""
@pytest.fixture
def normalizer(self):
return BinanceNormalizer()
def test_normalize_trade(self, normalizer):
"""Test chuẩn hóa trade event"""
raw_trade = {
"e": "trade",
"E": 1672515782016, # Event time
"s": "BTCUSDT", # Symbol
"t": 12345, # Trade ID
"p": "16500.50", # Price
"q": "0.001", # Quantity
"T": 1672515782000 # Trade time
}
result = normalizer.normalize(raw_trade)
assert result is not None
assert result.symbol == "BTC-USD"
assert result.exchange == "BINANCE"
assert result.data_type == MarketDataType.TRADE
assert result.price == 16500.50
assert result.volume == 0
assert result.timestamp == 1672515782000
def test_normalize_orderbook(self, normalizer):
"""Test chuẩn hóa orderbook"""
raw_orderbook = {
"e": "depthUpdate",
"E": 1672515782016,
"s": "ETHUSDT",
"U": 1,
"u": 2,
"b": [["2900.00", "10.5"]],
"a": [["2901.00", "8.2"]]
}
result = normalizer.normalize(raw_orderbook)
assert result is not None
assert result.symbol == "ETH-USD"
assert result.bid_price == 2900.0
assert result.ask_price == 2901.0
assert result.spread == 1.0
assert result.mid_price == 2900.5
def test_invalid_data(self, normalizer):
"""Test xử lý dữ liệu không hợp lệ"""
invalid_data = {"e": "unknown_event"}
result = normalizer.normalize(invalid_data)
assert result is None
assert normalizer.metrics['failed'] == 1
tests/test_websocket_client.py
import pytest
import asyncio
from websockets.server import serve
import json
@pytest.mark.asyncio
async def test_websocket_connection():
"""Test kết nối WebSocket"""
async def echo_handler(websocket):
async for message in websocket:
await websocket.send(f"echo: {message}")
# Start test server
async with serve(echo_handler, "localhost", 8766):
# Connect as client
import websockets
async with websockets.connect("ws://localhost:8766") as ws:
test_message = "test_data"
await ws.send(test_message)
response = await asyncio.wait_for(ws.recv(), timeout=5)
assert response == f"echo: {test_message}"
if __name__ == "__main__":
pytest.main([__file__, "-v"])
Performance Benchmarking
Để đảm bảo hệ thống đáp ứng yêu cầu về độ trễ thấp, tôi đã benchmark với các thông số sau:
# benchmark/tardis_benchmark.py
import asyncio
import time
import json
import statistics
from dataclasses import dataclass
from typing import List
@dataclass
class BenchmarkResult:
operation: str
iterations: int
total_time_ms: float
avg_time_ms: float
p50_ms: float
p95_ms: float
p99_ms: float
throughput_per_sec: float
async def benchmark_normalization():
"""Benchmark normalization throughput"""
from src.normalizers.binance_normalizer import BinanceNormalizer
normalizer = BinanceNormalizer()
# Sample trade data
sample_trade = {
"e": "trade",
"E": 1672515782016,
"s": "BTCUSDT",
"p": "16500.50",
"q": "0.001",
"T": 1672515782000
}
iterations = 10000
times = []
for _ in range(iterations):
start = time.perf_counter()
normalizer.normalize(sample_trade)
elapsed = (time.perf_counter() - start) * 1000
times.append(elapsed)
times.sort()
return BenchmarkResult(
operation="Normalization",
iterations=iterations,
total_time_ms=sum(times),
avg_time_ms=statistics.mean(times),
p50_ms=times[int(len(times) * 0.50)],
p95_ms=times[int(len(times) * 0.95)],
p99_ms=times[int(len(times) * 0.99)],
throughput_per_sec=iterations / (sum(times
Tài nguyên liên quan
Bài viết liên quan