Đêm khuya, hệ thống trading của tôi báo lỗi ConnectionError: timeout after 30s khi cố gắng lấy dữ liệu từ 5 sàn giao dịch cùng lúc. Đó là khoảnh khắc tôi nhận ra: xử lý đồng bộ (synchronous) không còn đủ cho trading thời gian thực. Bài viết này sẽ chia sẻ cách tôi xây dựng framework xử lý bất đồng bộ (asyncio) để handle hàng triệu data point mỗi ngày với độ trễ dưới 50ms.
Vì Sao Xử Lý Bất Đồng Bộ Quan Trọng Trong Trading?
Khi bạn cần lấy dữ liệu từ nhiều sàn (Binance, Coinbase, Kraken...) theo cách đồng bộ:
- Mỗi request chờ request trước hoàn thành
- Tổng thời gian = tổng tất cả API calls
- Ví dụ: 5 sàn × 200ms = 1000ms (1 giây trễ)
Với asyncio, tất cả requests chạy song song:
- Tổng thời gian = thời gian của request chậm nhất
- 5 sàn × 200ms song song = ~200ms (tiết kiệm 80%)
- Thông lượng tăng gấp 5-10 lần
Kiến Trúc Framework Xử Lý Bất Đồng Bộ
1. Class Cơ Bản: AsyncExchangeConnector
import asyncio
import aiohttp
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime
import json
@dataclass
class TickerData:
symbol: str
price: float
volume_24h: float
change_24h: float
timestamp: datetime
exchange: str
class AsyncExchangeConnector:
"""
Kết nối bất đồng bộ với nhiều sàn giao dịch
Thiết kế: Non-blocking I/O với asyncio
"""
def __init__(self, timeout: int = 10):
self.timeout = timeout
self.session: Optional[aiohttp.ClientSession] = None
self._rate_limiter = asyncio.Semaphore(5) # Tối đa 5 requests đồng thời
async def __aenter__(self):
connector = aiohttp.TCPConnector(
limit=100, # Tối đa 100 connections
limit_per_host=10, # Tối đa 10 connections/sàn
ttl_dns_cache=300 # Cache DNS 5 phút
)
self.session = aiohttp.ClientSession(
connector=connector,
timeout=aiohttp.ClientTimeout(total=self.timeout)
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def fetch_binance_ticker(self, symbol: str) -> Optional[TickerData]:
"""Lấy ticker từ Binance với xử lý lỗi"""
url = f"https://api.binance.com/api/v3/ticker/24hr?symbol={symbol}"
async with self._rate_limiter: # Kiểm soát concurrency
try:
async with self.session.get(url) as response:
if response.status == 200:
data = await response.json()
return TickerData(
symbol=symbol,
price=float(data['lastPrice']),
volume_24h=float(data['quoteVolume']),
change_24h=float(data['priceChangePercent']),
timestamp=datetime.fromtimestamp(data['closeTime']/1000),
exchange='binance'
)
elif response.status == 429:
# Rate limit - retry với exponential backoff
await asyncio.sleep(2 ** 2) # 4 giây
return await self.fetch_binance_ticker(symbol)
else:
print(f"Binance error: {response.status}")
return None
except asyncio.TimeoutError:
print(f"Timeout fetching {symbol} from Binance")
return None
except Exception as e:
print(f"Error Binance: {e}")
return None
Sử dụng
async def main():
async with AsyncExchangeConnector(timeout=10) as connector:
# Lấy 10 cặp tiền song song
symbols = ['BTCUSDT', 'ETHUSDT', 'BNBUSDT', 'ADAUSDT',
'DOGEUSDT', 'XRPUSDT', 'DOTUSDT', 'LTCUSDT',
'LINKUSDT', 'MATICUSDT']
tasks = [connector.fetch_binance_ticker(s) for s in symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
valid_results = [r for r in results if isinstance(r, TickerData)]
print(f"Lấy thành công: {len(valid_results)}/{len(symbols)} tickers")
Chạy
asyncio.run(main())2. Stream Processing Với Queue
import asyncio
from collections import deque
from typing import Callable, Awaitable
class AsyncDataStream:
"""
Luồng dữ liệu bất đồng bộ với buffering và backpressure
Phù hợp: Xử lý real-time data từ WebSocket streams
"""
def __init__(self, maxsize: int = 1000):
self._queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
self._running = False
self._processors: List[Callable] = []
async def put(self, item):
"""Thêm data vào stream - blocking nếu queue full"""
await self._queue.put(item)
async def get(self):
"""Lấy data từ