Giới Thiệu
Thị trường tiền mã hóa năm 2026 đang chứng kiến sự cạnh tranh khốc liệt giữa các sàn giao dịch, và
Bybit tiếp tục khẳng định vị thế với hạ tầng WebSocket tốc độ cao. Với độ trễ chỉ
2-5ms trên kết nối incremental_book_L2, traders định lượng có thể xây dựng chiến lược arbitrage, market making với độ chính xác cao.
Trong bài viết này, tôi sẽ hướng dẫn chi tiết cách kết nối
Bybit incremental_book_L2 vào hệ thống giao dịch định lượng, từ cấu hình WebSocket đến xử lý dữ liệu real-time. Đặc biệt, tôi sẽ chỉ ra cách tích hợp
HolySheep AI để phân tích signal từ order book thông qua LLM.
Bybit Incremental Book L2 Là Gì?
incremental_book_L2 là endpoint WebSocket của Bybit cung cấp dữ liệu order book theo thời gian thực. Khác với full order book snapshot, incremental update chỉ gửi những thay đổi (insert/update/delete) giúp:
- Giảm bandwidth tiêu thụ 70-80%
- Độ trễ trung bình 2-8ms
- Thích hợp cho HFT và arbitrage strategies
- Hỗ trợ tất cả cặp spot và futures
Kiến Trúc Hệ Thống
Trước khi vào code, hãy xem kiến trúc tổng quan:
┌─────────────────────────────────────────────────────────────────┐
│ QUANT TRADING SYSTEM ARCHITECTURE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Bybit WS │ ───► │ OrderBook │ ───► │ Strategy │ │
│ │ L2 Feed │ │ Manager │ │ Engine │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Reconnect │ │ Signal │ │ Executor │ │
│ │ Handler │ │ Generator │ │ (Orders) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐│
│ │ HolySheep AI ││
│ │ (LLM Risk) ││
│ └──────────────┘│
└─────────────────────────────────────────────────────────────────┘
Cài Đặt Môi Trường
# Cài đặt dependencies
pip install websockets asyncio aiofiles pandas numpy
Cấu trúc project
quant-bybit/
├── config/
│ └── settings.py
├── core/
│ ├── bybit_websocket.py
│ ├── orderbook_manager.py
│ └── signal_generator.py
├── strategies/
│ └── arbitrage_strategy.py
├── ai/
│ └── risk_analyzer.py
├── main.py
└── requirements.txt
Triển Khai Chi Tiết
1. Cấu Hình Settings
# config/settings.py
import os
from dataclasses import dataclass
@dataclass
class BybitConfig:
"""Cấu hình Bybit WebSocket - incremental_book_L2"""
WS_URL: str = "wss://stream.bybit.com/v5/public/linear"
SYMBOLS: list = None
DEPTH_LIMIT: int = 50
RECONNECT_DELAY: float = 1.0
MAX_RECONNECT: int = 10
PING_INTERVAL: int = 20
PING_TIMEOUT: int = 10
def __post_init__(self):
self.SYMBOLS = self.SYMBOLS or ["BTCUSDT", "ETHUSDT"]
@dataclass
class HolySheepConfig:
"""Cấu hình HolySheep AI - phân tích signal định lượng"""
BASE_URL: str = "https://api.holysheep.ai/v1"
API_KEY: str = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
MODEL: str = "gpt-4.1"
# Chi phí 2026: GPT-4.1 = $8/MTok (so với OpenAI $30/MTok)
COST_PER_MTOK: float = 8.0
SIGNAL_ANALYSIS_PROMPT: str = """
Phân tích order book signal và đưa ra khuyến nghị:
- Momentum score (0-100)
- Risk level (LOW/MEDIUM/HIGH)
- Action: BUY/SELL/HOLD
"""
@dataclass
class TradingConfig:
"""Cấu hình trading strategy"""
MIN_SPREAD: float = 0.001 # 0.1%
MAX_POSITION: float = 1000.0 # USDT
STOP_LOSS: float = 0.005 # 0.5%
TAKE_PROFIT: float = 0.01 # 1%
LEVERAGE: int = 3
Singleton instances
bybit_config = BybitConfig()
holysheep_config = HolySheepConfig()
trading_config = TradingConfig()
2. WebSocket Client - Kết Nối Bybit L2
# core/bybit_websocket.py
import asyncio
import json
import logging
from datetime import datetime
from typing import Callable, Optional, Dict, Any
import websockets
from websockets.exceptions import ConnectionClosed, InvalidStatusCode
from config.settings import bybit_config
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class BybitWebSocketClient:
"""
WebSocket client cho Bybit incremental_book_L2
- Tự động reconnect khi mất kết nối
- Xử lý subscription/unsubscription
- Callback pattern cho data processing
"""
def __init__(
self,
on_message: Optional[Callable] = None,
on_connect: Optional[Callable] = None,
on_disconnect: Optional[Callable] = None,
on_error: Optional[Callable] = None
):
self.ws_url = bybit_config.WS_URL
self.symbols = bybit_config.SYMBOLS
self.websocket = None
self.running = False
self.reconnect_count = 0
# Callbacks
self.on_message = on_message
self.on_connect = on_connect
self.on_disconnect = on_disconnect
self.on_error = on_error
# Metrics
self.messages_received = 0
self.last_message_time = None
self.connection_start_time = None
logger.info(f"🔌 Bybit WS Client khởi tạo cho: {self.symbols}")
async def connect(self) -> bool:
"""Thiết lập kết nối WebSocket"""
try:
self.websocket = await websockets.connect(
self.ws_url,
ping_interval=bybit_config.PING_INTERVAL,
ping_timeout=bybit_config.PING_TIMEOUT
)
self.running = True
self.connection_start_time = datetime.now()
self.reconnect_count = 0
logger.info(f"✅ Kết nối Bybit thành công: {self.ws_url}")
if self.on_connect:
await self.on_connect()
return True
except Exception as e:
logger.error(f"❌ Kết nối thất bại: {e}")
if self.on_error:
await self.on_error(e)
return False
async def subscribe_orderbook(self, symbols: list = None) -> bool:
"""
Subscribe incremental_book_L2 cho các cặp giao dịch
Response format:
{
"topic": "orderbook.50.{symbol}",
"type": "snapshot" | "delta",
"data": {
"s": "BTCUSDT",
"b": [["price", "size"], ...],
"a": [["price", "size"], ...],
"u": 123456,
"seq": 123456789
}
}
"""
symbols = symbols or self.symbols
subscribe_msg = {
"op": "subscribe",
"args": [f"orderbook.50.{symbol}" for symbol in symbols]
}
try:
await self.websocket.send(json.dumps(subscribe_msg))
logger.info(f"📡 Đã subscribe: {symbols}")
# Đợi confirmation
response = await asyncio.wait_for(
self.websocket.recv(),
timeout=5.0
)
resp_data = json.loads(response)
if resp_data.get("success"):
logger.info(f"✅ Subscribe thành công: {resp_data}")
return True
else:
logger.error(f"❌ Subscribe thất bại: {resp_data}")
return False
except asyncio.TimeoutError:
logger.error("⏰ Timeout chờ subscribe confirmation")
return False
except Exception as e:
logger.error(f"❌ Lỗi subscribe: {e}")
return False
async def listen(self):
"""Vòng lặp nhận messages từ Bybit"""
logger.info("👂 Bắt đầu lắng nghe messages...")
while self.running:
try:
message = await self.websocket.recv()
self.messages_received += 1
self.last_message_time = datetime.now()
data = json.loads(message)
if self.on_message:
await self.on_message(data)
except ConnectionClosed as e:
logger.warning(f"⚠️ Kết nối đóng: {e}")
await self.handle_reconnect()
break
except Exception as e:
logger.error(f"❌ Lỗi nhận message: {e}")
if self.on_error:
await self.on_error(e)
async def handle_reconnect(self):
"""Xử lý tự động reconnect với exponential backoff"""
if self.reconnect_count >= bybit_config.MAX_RECONNECT:
logger.error("🚫 Đạt max reconnect attempts")
return
delay = bybit_config.RECONNECT_DELAY * (2 ** self.reconnect_count)
self.reconnect_count += 1
logger.info(f"🔄 Reconnect attempt {self.reconnect_count} sau {delay}s...")
await asyncio.sleep(delay)
if await self.connect():
await self.subscribe_orderbook()
asyncio.create_task(self.listen())
async def unsubscribe(self, symbols: list = None):
"""Unsubscribe khỏi topics"""
symbols = symbols or self.symbols
unsubscribe_msg = {
"op": "unsubscribe",
"args": [f"orderbook.50.{symbol}" for symbol in symbols]
}
await self.websocket.send(json.dumps(unsubscribe_msg))
logger.info(f"📤 Đã unsubscribe: {symbols}")
async def close(self):
"""Đóng kết nối"""
self.running = False
if self.websocket:
await self.websocket.close()
logger.info("🔚 WebSocket đã đóng")
def get_metrics(self) -> Dict[str, Any]:
"""Lấy metrics của connection"""
return {
"messages_received": self.messages_received,
"last_message": self.last_message_time,
"uptime_seconds": (
datetime.now() - self.connection_start_time
).total_seconds() if self.connection_start_time else 0,
"reconnect_count": self.reconnect_count
}
3. Order Book Manager
# core/orderbook_manager.py
import asyncio
from collections import OrderedDict
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Tuple, Optional
from threading import Lock
import logging
import numpy as np
logger = logging.getLogger(__name__)
@dataclass
class OrderBookLevel:
"""Một mức giá trong order book"""
price: float
size: float
timestamp: datetime = field(default_factory=datetime.now)
def __post_init__(self):
self.price = float(self.price)
self.size = float(self.size)
@dataclass
class OrderBook:
"""
Quản lý order book state từ Bybit incremental updates
- Duy trì sorted order book
- Tính spread, mid price, imbalance
"""
symbol: str
bids: OrderedDict = field(default_factory=OrderedDict) # price -> OrderBookLevel
asks: OrderedDict = field(default_factory=OrderedDict)
last_update_id: int = 0
seq_num: int = 0
last_update_time: datetime = field(default_factory=datetime.now)
update_count: int = 0
def update_bid(self, price: float, size: float):
"""Cập nhật bid - size = 0 means delete"""
if size == 0:
self.bids.pop(price, None)
else:
self.bids[float(price)] = OrderBookLevel(price, size)
self.update_count += 1
self.last_update_time = datetime.now()
def update_ask(self, price: float, size: float):
"""Cập nhật ask - size = 0 means delete"""
if size == 0:
self.asks.pop(price, None)
else:
self.asks[float(price)] = OrderBookLevel(price, size)
self.update_count += 1
self.last_update_time = datetime.now()
def get_spread(self) -> float:
"""Tính bid-ask spread"""
if not self.bids or not self.asks:
return float('inf')
best_bid = max(self.bids.keys())
best_ask = min(self.asks.keys())
return best_ask - best_bid
def get_spread_pct(self) -> float:
"""Spread percentage"""
spread = self.get_spread()
mid_price = self.get_mid_price()
return (spread / mid_price * 100) if mid_price else 0
def get_mid_price(self) -> float:
"""Giá trung bình"""
if not self.bids or not self.asks:
return 0
best_bid = max(self.bids.keys())
best_ask = min(self.asks.keys())
return (best_bid + best_ask) / 2
def get_bid_imbalance(self, levels: int = 10) -> float:
"""Bid/Ask imbalance ratio (-1 to 1)"""
bid_vol = sum(
list(self.bids.values())[:levels]
)
ask_vol = sum(
list(self.asks.values())[:levels]
)
if bid_vol + ask_vol == 0:
return 0
return (bid_vol - ask_vol) / (bid_vol + ask_vol)
def get_top_of_book(self, n: int = 5) -> Dict:
"""Lấy N mức giá tốt nhất"""
sorted_bids = sorted(self.bids.items(), reverse=True)[:n]
sorted_asks = sorted(self.asks.items())[:n]
return {
"symbol": self.symbol,
"mid_price": self.get_mid_price(),
"spread_pct": self.get_spread_pct(),
"bids": [(p, float(v.size)) for p, v in sorted_bids],
"asks": [(p, float(v.size)) for p, v in sorted_asks],
"bid_imbalance": self.get_bid_imbalance(5),
"timestamp": self.last_update_time.isoformat()
}
class OrderBookManager:
"""
Quản lý nhiều order books cho các cặp giao dịch
Thread-safe với asyncio
"""
def __init__(self):
self.books: Dict[str, OrderBook] = {}
self.lock = Lock()
self._callbacks = []
def register_callback(self, callback):
"""Đăng ký callback khi order book update"""
self._callbacks.append(callback)
async def process_update(self, data: dict):
"""
Xử lý incremental update từ Bybit WebSocket
data format:
{
"topic": "orderbook.50.BTCUSDT",
"type": "delta",
"data": {
"s": "BTCUSDT",
"b": [["price", "size"], ...],
"a": [["price", "size"], ...],
"u": 123456,
"seq": 123456789
}
}
"""
try:
symbol = data.get("data", {}).get("s")
if not symbol:
return
update_type = data.get("type", "delta")
update_data = data.get("data", {})
with self.lock:
# Khởi tạo book nếu chưa có (snapshot)
if symbol not in self.books:
self.books[symbol] = OrderBook(symbol=symbol)
book = self.books[symbol]
if update_type == "snapshot":
# Clear và rebuild từ snapshot
book.bids.clear()
book.asks.clear()
for price, size in update_data.get("b", []):
book.update_bid(float(price), float(size))
for price, size in update_data.get("a", []):
book.update_ask(float(price), float(size))
else: # delta update
for price, size in update_data.get("b", []):
book.update_bid(float(price), float(size))
for price, size in update_data.get("a", []):
book.update_ask(float(price), float(size))
book.last_update_id = update_data.get("u", 0)
book.seq_num = update_data.get("seq", 0)
# Trigger callbacks
for callback in self._callbacks:
await callback(book, update_type)
except Exception as e:
logger.error(f"❌ Lỗi process update: {e}")
def get_book(self, symbol: str) -> Optional[OrderBook]:
"""Lấy order book cho symbol"""
with self.lock:
return self.books.get(symbol)
def get_all_books(self) -> Dict[str, OrderBook]:
"""Lấy tất cả order books"""
with self.lock:
return self.books.copy()
4. Signal Generator - Chiến Lược Arbitrage
# core/signal_generator.py
import asyncio
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime
from enum import Enum
import logging
from core.orderbook_manager import OrderBook, OrderBookManager
from config.settings import trading_config
logger = logging.getLogger(__name__)
class SignalType(Enum):
BUY = "BUY"
SELL = "SELL"
HOLD = "HOLD"
@dataclass
class TradingSignal:
"""Signal giao dịch từ strategy"""
symbol: str
signal_type: SignalType
entry_price: float
stop_loss: float
take_profit: float
confidence: float # 0-100
spread_pct: float
imbalance: float
timestamp: datetime
reasoning: str
def to_dict(self) -> Dict:
return {
"symbol": self.symbol,
"signal": self.signal_type.value,
"entry": self.entry_price,
"sl": self.stop_loss,
"tp": self.take_profit,
"confidence": self.confidence,
"spread": f"{self.spread_pct:.4f}%",
"imbalance": f"{self.imbalance:.4f}",
"timestamp": self.timestamp.isoformat(),
"reasoning": self.reasoning
}
class ArbitrageSignalGenerator:
"""
Chiến lược arbitrage dựa trên:
- Cross-exchange spread
- Order book imbalance
- Volume analysis
"""
def __init__(self, orderbook_manager: OrderBookManager):
self.ob_manager = orderbook_manager
self.min_spread = trading_config.MIN_SPREAD
self.stop_loss = trading_config.STOP_LOSS
self.take_profit = trading_config.TAKE_PROFIT
self.history: List[TradingSignal] = []
async def analyze(self, book: OrderBook) -> Optional[TradingSignal]:
"""Phân tích order book và tạo signal"""
try:
spread_pct = book.get_spread_pct()
imbalance = book.get_bid_imbalance(10)
mid_price = book.get_mid_price()
# Điều kiện signal BUY: imbalance > threshold
if imbalance > 0.15 and spread_pct > self.min_spread:
signal = TradingSignal(
symbol=book.symbol,
signal_type=SignalType.BUY,
entry_price=mid_price,
stop_loss=mid_price * (1 - self.stop_loss),
take_profit=mid_price * (1 + self.take_profit),
confidence=min(100, 50 + imbalance * 100),
spread_pct=spread_pct,
imbalance=imbalance,
timestamp=datetime.now(),
reasoning=f"BUY: Imbalance={imbalance:.3f}, Spread={spread_pct:.4f}%"
)
self.history.append(signal)
return signal
# Điều kiện signal SELL: imbalance < -threshold
elif imbalance < -0.15 and spread_pct > self.min_spread:
signal = TradingSignal(
symbol=book.symbol,
signal_type=SignalType.SELL,
entry_price=mid_price,
stop_loss=mid_price * (1 + self.stop_loss),
take_profit=mid_price * (1 - self.take_profit),
confidence=min(100, 50 + abs(imbalance) * 100),
spread_pct=spread_pct,
imbalance=imbalance,
timestamp=datetime.now(),
reasoning=f"SELL: Imbalance={imbalance:.3f}, Spread={spread_pct:.4f}%"
)
self.history.append(signal)
return signal
return None
except Exception as e:
logger.error(f"❌ Lỗi phân tích signal: {e}")
return None
def get_statistics(self) -> Dict:
"""Thống kê signals đã tạo"""
if not self.history:
return {"total": 0}
return {
"total": len(self.history),
"buy_signals": sum(1 for s in self.history if s.signal_type == SignalType.BUY),
"sell_signals": sum(1 for s in self.history if s.signal_type == SignalType.SELL),
"avg_confidence": sum(s.confidence for s in self.history) / len(self.history),
"last_signal": self.history[-1].to_dict() if self.history else None
}
5. Tích Hợp HolySheep AI - Phân Tích Risk
# ai/risk_analyzer.py
import asyncio
import json
import logging
from typing import Dict, List, Optional
from datetime import datetime
from dataclasses import dataclass
import aiohttp
from config.settings import holysheep_config
logger = logging.getLogger(__name__)
@dataclass
class RiskAnalysis:
"""Kết quả phân tích risk từ AI"""
momentum_score: int # 0-100
risk_level: str # LOW/MEDIUM/HIGH
action: str # BUY/SELL/HOLD
reasoning: str
confidence: float
estimated_cost: float # USD
class HolySheepRiskAnalyzer:
"""
Sử dụng HolySheep AI để phân tích order book signal
HolySheep ưu điểm:
- Giá chỉ $8/MTok (so với $30 của OpenAI)
- Độ trễ <50ms
- Hỗ trợ WeChat/Alipay thanh toán
"""
def __init__(self, api_key: str = None):
self.base_url = holysheep_config.BASE_URL
self.api_key = api_key or holysheep_config.API_KEY
self.model = holysheep_config.MODEL
self.cost_per_mtok = holysheep_config.COST_PER_MTOK
self.session: Optional[aiohttp.ClientSession] = None
async def _ensure_session(self):
"""Đảm bảo session tồn tại"""
if self.session is None or self.session.closed:
self.session = aiohttp.ClientSession()
async def analyze_risk(
self,
orderbook_data: Dict,
trading_signal: Dict
) -> Optional[RiskAnalysis]:
"""
Phân tích risk sử dụng LLM qua HolySheep API
"""
await self._ensure_session()
# Build prompt với context đầy đủ
prompt = f"""Bạn là chuyên gia phân tích rủi ro giao dịch tiền mã hóa.
Dữ liệu Order Book (Bybit):
- Symbol: {orderbook_data.get('symbol')}
- Mid Price: ${orderbook_data.get('mid_price', 0):.2f}
- Spread: {orderbook_data.get('spread_pct', 0):.4f}%
- Bid Imbalance: {orderbook_data.get('bid_imbalance', 0):.4f}
Signal Chiến lược:
- Action: {trading_signal.get('signal')}
- Confidence: {trading_signal.get('confidence', 0):.1f}%
- Entry Price: ${trading_signal.get('entry', 0):.2f}
- Stop Loss: ${trading_signal.get('sl', 0):.2f}
- Take Profit: ${trading_signal.get('tp', 0):.2f}
Yêu cầu:
Phân tích và trả lời CHÍNH XÁC format JSON:
{{
"momentum_score": 0-100,
"risk_level": "LOW|MEDIUM|HIGH",
"action": "BUY|SELL|HOLD",
"reasoning": "giải thích ngắn gọn",
"confidence": 0.0-1.0
}}
"""
try:
# Gọi HolySheep API
async with self.session.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [
{
"role": "system",
"content": "Bạn là chuyên gia phân tích rủi ro. Chỉ trả lời JSON hợp lệ."
},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
},
timeout=aiohttp.ClientTimeout(total=5)
) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"❌ HolySheep API error {response.status}: {error_text}")
return None
result = await response.json()
# Parse response
content = result["choices"][0]["message"]["content"]
# Extract JSON từ response
try:
# Thử parse trực tiếp
analysis_data = json.loads(content)
except json.JSONDecodeError:
# Tìm JSON trong text
start = content.find("{")
end = content.rfind("}") + 1
if start != -1 and end != 0:
analysis_data = json.loads(content[start:end])
else:
logger.error("❌ Không tìm thấy JSON trong response")
return None
# Tính chi phí
tokens_used = result.get("usage", {}).get("total_tokens", 0)
estimated_cost = (tokens_used / 1_000_000) * self.cost_per_mtok
logger.info(
f"💰 HolySheep analysis: {tokens_used} tokens, "
f"~$ {estimated_cost:.4f}"
)
return RiskAnalysis(
momentum_score=analysis_data.get("momentum_score", 50),
risk_level=analysis_data.get("risk_level", "MEDIUM"),
action=analysis_data.get("action", "HOLD"),
reasoning=analysis_data.get("reasoning", ""),
confidence=analysis_data.get("confidence", 0.5),
estimated_cost=estimated_cost
)
except asyncio.TimeoutError:
logger.error("⏰ HolySheep API timeout")
return None
except Exception as e:
logger.error(f"❌ Lỗi HolySheep analysis: {e}")
return None
async def batch_analyze(
self,
signals: List[Dict]
) -> List[RiskAnalysis]:
"""Phân tích nhiều signals song song"""
tasks = []
for signal in signals:
orderbook_data = signal.get("orderbook", {})
tasks.append(self.analyze_risk(orderbook_data, signal))
results = await asyncio.gather(*tasks, return_exceptions=True)
return [r for r in results if isinstance(r, RiskAnalysis)]
async def close(self):
"""Đóng session"""
if self.session:
await self.session.close()
6. Main - Tích Hợp Toàn Bộ Hệ Thống
# main.py
import asyncio
import logging
from datetime import datetime
from config.settings import bybit_config, holysheep_config, trading_config
from core.bybit_websocket import BybitWebSocketClient
from core.orderbook_manager import OrderBookManager
from core.signal_generator import ArbitrageSignalGenerator, TradingSignal
from ai.risk_analyzer import HolySheepRiskAnalyzer
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class QuantTradingSystem:
"""
Hệ thống giao dịch định lượng hoàn chỉnh
- Kết nối Bybit WebSocket (incremental_book_L2)
- Xử lý order book real-time
- Tạo signals từ chiến lược
- Phân tích risk với HolySheep AI
"""
def __init__(self):
self.ws_client = BybitWebSocketClient(
on_message=self.handle_message,
on_connect=self.handle_connect,
on_disconnect=self.handle_disconnect
)
self.ob_manager = OrderBookManager()
self.signal_generator = ArbitrageSignalGenerator(self.ob_manager)
self.risk_analyzer = HolySheepRiskAnalyzer(
api_key=holysheep_config.API_KEY
)
self.trade_count = 0
self.last_signal_print = datetime.now()
# Đăng ký callback cho order book updates
self.ob_manager.register_callback(self.on_orderbook_update)
async def handle_connect(self):
"""Xử lý khi kết nối thành công"""
logger.info("🎉 Kết nối thành công - Subscribing order book...")
await self.ws_client.subscribe_orderbook(bybit_config.SYMBOLS)
async def handle_disconnect(self, error=None):
"""Xử lý khi mất kết nối"""
logger.warning(f"⚠️ Mất kết nối: {error}")
async def handle_message(self, data: dict):
"""Xử lý message từ Bybit WebSocket"""
try:
topic = data.get("topic", "")
if "orderbook" in topic:
await self.ob_manager.process_update(data)
except Exception as e:
logger.error(f"❌ Lỗi xử lý message: {e}")
async def on_orderbook_update(self, book, update_type: str):
"""
Callback khi order book được cập nhật
- Tạo trading signal
- Phân tích risk với AI
"""
# Chỉ xử lý mỗi 100ms để tránh spam
now = datetime.now()
if (now - self.last_signal_print).total_seconds() < 0.1:
return
# Tạo signal
signal = await self.signal_generator.analyze(book)
if signal and signal.signal_type.value != "HOLD":
self.last_signal_print = now
logger.info(f"📊 Signal: {signal.signal_type.value} {signal.symbol} "
f"@ ${signal.entry_price:.2f} "
f"(Confidence: {signal.confidence:.1f}%)")
# Phân tích risk với HolySheep AI
risk_analysis = await self.risk_analyzer.analyze_risk(
orderbook_data=book.get_top_of_book(10),
trading_signal=signal.to_dict()
)
if risk_analysis:
logger.info(
f"🛡️ Risk Analysis: {risk_analysis.action} "
f"(Score: {risk_analysis.momentum_score}, "
f"Risk: {risk_analysis.risk_level}, "
f"Cost: ${risk_analysis.estimated_cost:.4f})"
)
# Kiểm tra điều kiện thực thi
if (risk_analysis.action == signal.signal_type.value
and risk_analysis.risk_level != "HIGH"
and
Tài nguyên liên quan
Bài viết liên quan