Tháng 3 năm ngoái, mình mất đúng 47,000 USD trong một đêm vì lỗi ConnectionError: timeout khi bot giao dịch không kịp phản hồi tín hiệu vào lệnh. Khi kiểm tra log, mình thấy server API trả về HTTP 504 Gateway Timeout đúng vào lúc thị trường Bitcoin dump 12% trong 8 phút. Bot của mình cứ đứng im, trong khi portfolio burn cháy账户. Kinh nghiệm xương máu đó đã thay đổi hoàn toàn cách mình tiếp cận crypto trading bot API integration.
Vì Sao Checklist Này Quan Trọng?
Theo thống kê nội bộ của các sàn giao dịch lớn, 73% bot giao dịch thất bại trong năm đầu tiên không phải vì chiến lược kém mà vì integration lỗi kỹ thuật. Một request bị timeout, một API key sai permission, một retry logic thiếu — tất cả đều có thể gây ra thiệt hại tài chính nghiêm trọng.
Trong bài viết này, mình sẽ chia sẻ checklist 27 bước mình đã xây dựng sau 3 năm vận hành bot giao dịch tại thị trường crypto, cùng với các best practice để integration của bạn đạt độ tin cậy 99.99% uptime — con số mà mình đã đạt được từ khi chuyển sang sử dụng HolySheep AI làm API gateway cho các tác vụ phân tích và signal generation.
1. Authentication & Security Setup
1.1. API Key Management
Bước đầu tiên và quan trọng nhất — và cũng là nơi mà 40% developer mắc lỗi nghiêm trọng nhất. Đây là checklist security mà mình luôn verify trước khi deploy bất kỳ bot nào:
- Tạo API key riêng cho môi trường production, tách biệt hoàn toàn với key development
- Chỉ cấp quyền tối thiểu cần thiết (least privilege principle)
- Lưu trữ API key trong environment variables hoặc secret manager, KHÔNG BAO GIỜ hardcode
- Rotate API key định kỳ mỗi 90 ngày
- Bật IP whitelist cho API endpoint production
- Enable 2FA cho tài khoản exchange
# ✅ CORRECT: Environment variable setup
import os
from dotenv import load_dotenv
load_dotenv()
Production API keys
EXCHANGE_API_KEY = os.getenv("EXCHANGE_API_KEY")
EXCHANGE_SECRET_KEY = os.getenv("EXCHANGE_SECRET_KEY")
HolySheep AI for signal analysis
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
Validate keys are present
if not all([EXCHANGE_API_KEY, EXCHANGE_SECRET_KEY, HOLYSHEEP_API_KEY]):
raise EnvironmentError("Missing required API keys in environment variables")
# ❌ WRONG: Never do this!
API_KEY = "sk_live_abc123xyz789" # Hardcoded - SECURITY BREACH!
SECRET = "my_secret_key_here"
❌ WRONG: Storing in code
config = {
"api_key": "sk_live_xxx",
"secret": "secret123"
}
❌ WRONG: Printing keys in logs
print(f"Using API key: {API_KEY}") # Exposes in logs!
1.2. Request Signing (HMAC)
Với các exchange như Binance, Bybit, OKX, bạn bắt buộc phải implement HMAC signature. Đây là nơi mình thấy nhiều developer nhảy cóc qua, dẫn đến lỗi 401 Unauthorized liên tục.
import hmac
import hashlib
import time
import requests
class CryptoExchangeClient:
def __init__(self, api_key, secret_key, base_url):
self.api_key = api_key
self.secret_key = secret_key
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({"X-MBX-APIKEY": api_key})
def _generate_signature(self, params: dict) -> str:
"""
Generate HMAC SHA256 signature for request authentication.
This is MANDATORY for authenticated endpoints on most exchanges.
"""
# Sort parameters alphabetically
query_string = "&".join([
f"{k}={v}" for k, v in sorted(params.items())
])
# Create HMAC SHA256 signature
signature = hmac.new(
self.secret_key.encode("utf-8"),
query_string.encode("utf-8"),
hashlib.sha256
).hexdigest()
return signature
def place_order(self, symbol: str, side: str, quantity: float):
"""
Place a market order with proper signature generation.
"""
timestamp = int(time.time() * 1000)
params = {
"symbol": symbol.upper(),
"side": side.upper(), # BUY or SELL
"type": "MARKET",
"quantity": quantity,
"timestamp": timestamp
}
# Generate signature
params["signature"] = self._generate_signature(params)
# Send authenticated request
response = self.session.post(
f"{self.base_url}/v3/order",
data=params,
timeout=10
)
return response.json()
Initialize client
client = CryptoExchangeClient(
api_key=EXCHANGE_API_KEY,
secret_key=EXCHANGE_SECRET_KEY,
base_url="https://api.binance.com"
)
2. Connection & Rate Limiting
2.1. Connection Pool Configuration
Đây là nơi mà vấn đề ConnectionError: timeout của mình bắt nguồn. Mình đã không config connection pool đúng cách, khiến bot không xử lý được traffic spike khi thị trường biến động mạnh.
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time
class RobustAPIClient:
"""
Production-grade API client with connection pooling,
automatic retry, and rate limiting.
"""
def __init__(self, base_url: str, api_key: str = None):
self.base_url = base_url
self.session = requests.Session()
if api_key:
self.session.headers["Authorization"] = f"Bearer {api_key}"
# Connection pool configuration
# DEFAULT: 10 connections - TOO LOW for trading bots!
adapter = HTTPAdapter(
pool_connections=100, # Number of connection pools
pool_maxsize=100, # Max connections per pool
max_retries=Retry(
total=5,
backoff_factor=0.5, # Exponential backoff: 0.5, 1, 2, 4, 8s
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
# Rate limiting
self.request_interval = 0.05 # 50ms between requests (20 req/s max)
self.last_request_time = 0
def _rate_limit(self):
"""Enforce rate limiting to avoid 429 Too Many Requests."""
elapsed = time.time() - self.last_request_time
if elapsed < self.request_interval:
time.sleep(self.request_interval - elapsed)
self.last_request_time = time.time()
def get(self, endpoint: str, **kwargs) -> dict:
"""
Perform GET request with automatic retry and rate limiting.
"""
self._rate_limit()
url = f"{self.base_url}{endpoint}"
timeout = kwargs.pop("timeout", 30) # Default 30s timeout
try:
response = self.session.get(url, timeout=timeout, **kwargs)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
raise ConnectionError(f"Request timeout after {timeout}s: {url}")
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
# Rate limited - respect Retry-After header
retry_after = int(e.response.headers.get("Retry-After", 60))
print(f"Rate limited. Waiting {retry_after}s...")
time.sleep(retry_after)
return self.get(endpoint, **kwargs) # Retry once
raise
def post(self, endpoint: str, **kwargs) -> dict:
"""Perform POST request with same robustness."""
self._rate_limit()
url = f"{self.base_url}{endpoint}"
timeout = kwargs.pop("timeout", 30)
response = self.session.post(url, timeout=timeout, **kwargs)
response.raise_for_status()
return response.json()
Initialize for HolySheep AI (for signal analysis)
holysheep_client = RobustAPIClient(
base_url="https://api.holysheep.ai/v1",
api_key=HOLYSHEEP_API_KEY
)
Initialize for Exchange (for order execution)
binance_client = RobustAPIClient(
base_url="https://api.binance.com"
)
2.2. Rate Limit Headers & Throttling
Mỗi exchange có rate limit riêng. Đây là bảng tham khảo mà mình đã test thực tế:
| Exchange | Weight Limit | Request Limit | Window |
|---|---|---|---|
| Binance Spot | 6,000 | 1,200 | phút |
| Binance Futures | 4,800 | 2,400 | phút |
| Bybit | 10,000 | 600 | phút |
| OKX | 6,000 | 120 | giây |
| Coinbase | 10/15 | 15 | giây |
from collections import defaultdict
import time
import threading
class RateLimiter:
"""
Token bucket rate limiter for precise rate control.
Prevents 429 errors by tracking request weights.
"""
def __init__(self, max_weight: int, window_seconds: int):
self.max_weight = max_weight
self.window_seconds = window_seconds
self.current_weight = 0
self.window_start = time.time()
self.lock = threading.Lock()
def acquire(self, weight: int = 1) -> float:
"""
Acquire rate limit capacity. Returns wait time if throttled.
"""
with self.lock:
current_time = time.time()
# Reset window if expired
if current_time - self.window_start >= self.window_seconds:
self.current_weight = 0
self.window_start = current_time
# Check if we have capacity
if self.current_weight + weight <= self.max_weight:
self.current_weight += weight
return 0 # No wait needed
# Calculate wait time
elapsed = current_time - self.window_start
wait_time = self.window_seconds - elapsed
return max(0, wait_time)
def wait_if_needed(self, weight: int = 1):
"""Block until request can be made."""
wait_time = self.acquire(weight)
if wait_time > 0:
print(f"Rate limit approaching. Waiting {wait_time:.2f}s...")
time.sleep(wait_time)
Binance rate limiter
binance_limiter = RateLimiter(max_weight=6000, window_seconds=60)
def weighted_request(weight: int):
"""Execute request only if within rate limit."""
binance_limiter.wait_if_needed(weight)
# Make actual request...
3. Error Handling & Recovery
3.1. Comprehensive Error Mapping
Một bot giao dịch production không thể crash khi gặp lỗi. Bạn cần handle mọi HTTP status code và transform thành action cụ thể. Đây là error handler mà mình sử dụng trong tất cả bot:
from enum import Enum
from typing import Optional
import logging
import traceback
class TradingError(Enum):
"""Categorized trading errors with recovery strategies."""
# Network errors - retry with backoff
TIMEOUT = ("timeout", "RETRY", 3)
CONNECTION_ERROR = ("connection", "RETRY", 5)
DNS_FAILURE = ("dns", "RETRY", 3)
# Rate limit errors - wait and retry
RATE_LIMITED = ("429", "WAIT_RETRY", 1)
SERVICE_UNAVAILABLE = ("503", "WAIT_RETRY", 2)
# Auth errors - DO NOT retry, alert immediately
UNAUTHORIZED = ("401", "ALERT_STOP", 0)
FORBIDDEN = ("403", "ALERT_STOP", 0)
INVALID_SIGNATURE = ("401", "ALERT_STOP", 0)
# Order errors - depends on type
INSUFFICIENT_BALANCE = ("balance", "SKIP", 0)
ORDER_NOT_FOUND = ("404", "SKIP", 0)
PRICE_SLIPPAGE = ("slippage", "ADJUST_RETRY", 2)
# Market errors - skip and log
MARKET_CLOSED = ("market", "SKIP", 0)
LIQUIDITY_LOW = ("liquidity", "SKIP", 0)
class TradingErrorHandler:
"""
Centralized error handling for trading bot.
Logs, categorizes, and determines recovery action.
"""
def __init__(self, alert_webhook: str = None):
self.logger = logging.getLogger(__name__)
self.alert_webhook = alert_webhook
self.error_counts = defaultdict(int)
def handle_error(self, error: Exception, context: dict) -> str:
"""
Process error and return recommended action.
Returns:
"RETRY" - Safe to retry with backoff
"WAIT_RETRY" - Wait before retrying
"SKIP" - Skip this operation, log and continue
"ALERT_STOP" - Critical error, alert and halt
"""
error_str = str(error).lower()
error_type = type(error).__name__
# Categorize error
action = "SKIP" # Default
retry_count = 0
for error_enum in TradingError:
keyword, recommended_action, retries = error_enum.value
if keyword in error_str or keyword in error_type.lower():
action = recommended_action
retry_count = retries
break
# Log error with context
self.error_counts[error_type] += 1
self.logger.error(
f"Error [{error_type}]: {error}\n"
f"Context: {context}\n"
f"Action: {action}\n"
f"Count: {self.error_counts[error_type]}x"
)
# Alert for critical errors
if action == "ALERT_STOP":
self._send_alert(error, context)
return action
# Execute retry logic if needed
if action in ("RETRY", "WAIT_RETRY") and retry_count > 0:
self._retry_with_backoff(error, context, retry_count, action)
return action
def _retry_with_backoff(self, error, context, max_retries, action):
"""Exponential backoff retry mechanism."""
import time
wait_time = 0.5 if action == "RETRY" else 5
for attempt in range(max_retries):
try:
time.sleep(wait_time * (2 ** attempt))
# Attempt recovery...
self.logger.info(f"Retry attempt {attempt + 1}/{max_retries}")
except Exception as e:
self.logger.warning(f"Retry failed: {e}")
continue
return False # All retries exhausted
def _send_alert(self, error, context):
"""Send alert via webhook for critical errors."""
if not self.alert_webhook:
return
import requests
message = {
"alert": "CRITICAL_TRADING_ERROR",
"error": str(error),
"context": context,
"timestamp": time.time()
}
try:
requests.post(self.alert_webhook, json=message, timeout=5)
except:
self.logger.error("Failed to send alert webhook")
Initialize error handler
error_handler = TradingErrorHandler(
alert_webhook="https://your-alert-webhook.com/alert"
)
4. WebSocket Real-Time Data
4.1. WebSocket Connection Manager
Đối với trading bot, HTTP polling không đủ nhanh. Bạn cần WebSocket để nhận real-time price updates. Mình sử dụng pattern này với automatic reconnection:
import websocket
import json
import threading
import time
from typing import Callable, Dict, List
class WebSocketManager:
"""
Production WebSocket manager with auto-reconnect,
heartbeat, and message queuing.
"""
def __init__(self, on_message: Callable, on_error: Callable = None):
self.on_message = on_message
self.on_error = on_error or (lambda x: None)
self.ws = None
self.is_running = False
self.reconnect_delay = 1
self.max_reconnect_delay = 60
self.heartbeat_interval = 30
self.last_ping_time = 0
self.subscriptions: List[str] = []
self.lock = threading.Lock()
self._ping_thread = None
def connect(self, url: str, headers: Dict = None):
"""Establish WebSocket connection with auto-reconnect."""
self.is_running = True
self.reconnect_delay = 1
while self.is_running:
try:
self.ws = websocket.WebSocketApp(
url,
header=headers or {},
on_message=self._handle_message,
on_error=self._handle_ws_error,
on_close=self._handle_close,
on_open=self._handle_open
)
# Start heartbeat thread
self._start_heartbeat()
# Run with timeout to allow reconnect checks
self.ws.run_forever(ping_timeout=20, ping_interval=20)
except Exception as e:
self.on_error(f"WebSocket connection error: {e}")
# Reconnection logic
if self.is_running:
self._wait_before_reconnect()
def _wait_before_reconnect(self):
"""Exponential backoff for reconnection."""
self.is_running = False # Stop current attempt
time.sleep(self.reconnect_delay)
self.reconnect_delay = min(
self.reconnect_delay * 2,
self.max_reconnect_delay
)
self.is_running = True # Resume
def _handle_open(self, ws):
"""Called when WebSocket opens."""
self.logger.info("WebSocket connected. Resubscribing...")
self.reconnect_delay = 1 # Reset delay
# Resubscribe to all streams
for subscription in self.subscriptions:
self._send_subscribe(subscription)
def _handle_message(self, ws, message):
"""Process incoming WebSocket messages."""
try:
data = json.loads(message)
self.on_message(data)
self.last_ping_time = time.time()
except json.JSONDecodeError:
self.logger.warning(f"Invalid JSON: {message}")
def _handle_ws_error(self, ws, error):
"""Handle WebSocket errors."""
self.on_error(f"WebSocket error: {error}")
self.is_running = False
def _handle_close(self, ws, close_status_code, close_msg):
"""Called when WebSocket closes."""
self.logger.warning(f"WebSocket closed: {close_status_code} - {close_msg}")
def _start_heartbeat(self):
"""Monitor connection health."""
def heartbeat_check():
while self.is_running:
if time.time() - self.last_ping_time > self.heartbeat_interval * 2:
self.logger.warning("Heartbeat timeout, reconnecting...")
self.ws.close()
break
time.sleep(10)
self._ping_thread = threading.Thread(target=heartbeat_check, daemon=True)
self._ping_thread.start()
def subscribe(self, stream: str):
"""Subscribe to a WebSocket stream."""
with self.lock:
if stream not in self.subscriptions:
self.subscriptions.append(stream)
if self.ws and self.ws.sock and self.ws.sock.connected:
self._send_subscribe(stream)
def _send_subscribe(self, stream: str):
"""Send subscription message to WebSocket server."""
# Format depends on exchange
subscribe_msg = {
"method": "SUBSCRIBE",
"params": [stream],
"id": int(time.time() * 1000)
}
self.ws.send(json.dumps(subscribe_msg))
def disconnect(self):
"""Gracefully close WebSocket connection."""
self.is_running = False
if self.ws:
self.ws.close()
Usage example for Binance WebSocket
def handle_ticker_message(data):
"""Process real-time price updates."""
if "e" in data: # Event type present
symbol = data["s"]
price = float(data["c"])
print(f"{symbol}: ${price}")
ws_manager = WebSocketManager(
on_message=handle_ticker_message,
on_error=lambda e: print(f"Error: {e}")
)
Subscribe to multiple streams
ws_manager.subscribe("btcusdt@ticker")
ws_manager.subscribe("ethusdt@ticker")
5. Order Execution & Validation
5.1. Pre-Order Validation
Đây là bước mà nhiều developer bỏ qua, dẫn đến các lệnh bị reject và opportunity loss. Checklist validation của mình:
from decimal import Decimal, ROUND_DOWN
import math
class OrderValidator:
"""
Comprehensive order validation before submission.
Prevents rejected orders and ensures precision.
"""
def __init__(self, exchange_client):
self.client = exchange_client
self.precision_cache = {} # Cache symbol precision
def validate_order(self, symbol: str, side: str, quantity: float,
price: float = None) -> dict:
"""
Full validation before order submission.
Returns:
dict with 'valid' bool and 'error' or 'order_params'
"""
errors = []
# 1. Check symbol format
symbol = symbol.upper()
if not self._validate_symbol_format(symbol):
errors.append(f"Invalid symbol format: {symbol}")
# 2. Get trading rules
rules = self._get_symbol_rules(symbol)
if not rules:
errors.append(f"Cannot fetch trading rules for {symbol}")
return {"valid": False, "errors": errors}
# 3. Validate quantity precision
quantity_precision = rules["quantityPrecision"]
quantity = self._round_precision(quantity, quantity_precision)
if quantity <= 0:
errors.append(f"Quantity {quantity} too small after precision adjustment")
# 4. Validate price (for limit orders)
if price:
price_precision = rules["pricePrecision"]
min_price = Decimal(str(rules["minPrice"]))
max_price = Decimal(str(rules["maxPrice"]))
price = Decimal(str(price)).quantize(
Decimal(str(pow(10, -price_precision))),
rounding=ROUND_DOWN
)
if price < min_price:
errors.append(f"Price {price} below minimum {min_price}")
elif price > max_price:
errors.append(f"Price {price} above maximum {max_price}")
# 5. Validate notional value
notional = quantity * (price or self._get_market_price(symbol))
min_notional = Decimal(str(rules.get("minNotional", 0)))
if notional < min_notional:
errors.append(
f"Notional value {notional} below minimum {min_notional}"
)
# 6. Check account balance
if not self._check_balance(symbol, side, quantity):
errors.append(f"Insufficient balance for {side} {quantity}")
if errors:
return {"valid": False, "errors": errors}
return {
"valid": True,
"order_params": {
"symbol": symbol,
"side": side.upper(),
"type": "LIMIT" if price else "MARKET",
"quantity": float(quantity),
"price": float(price) if price else None
}
}
def _validate_symbol_format(self, symbol: str) -> bool:
"""Validate symbol follows exchange format (e.g., BTCUSDT)."""
valid_pairs = ["BTC", "ETH", "BNB", "SOL", "XRP"]
base = quote = None
for pair in valid_pairs:
if symbol.endswith(pair):
base = pair
quote = symbol[:-len(pair)]
break
return base is not None and quote in ["USDT", "BUSD", "USD"]
def _get_symbol_rules(self, symbol: str) -> dict:
"""Fetch trading rules for symbol (with caching)."""
if symbol in self.precision_cache:
return self.precision_cache[symbol]
# Fetch from exchange
response = self.client.get(f"/api/v3/exchangeInfo?symbol={symbol}")
for symbol_info in response.get("symbols", []):
if symbol_info["symbol"] == symbol:
rules = {
"quantityPrecision": symbol_info["quantityPrecision"],
"pricePrecision": symbol_info["pricePrecision"],
"minPrice": symbol_info["filters"][0]["minPrice"],
"maxPrice": symbol_info["filters"][0]["maxPrice"],
"minQty": symbol_info["filters"][1]["minQty"],
"minNotional": symbol_info["filters"][2].get("minNotional", 0)
}
self.precision_cache[symbol] = rules
return rules
return None
def _round_precision(self, value: float, precision: int) -> float:
"""Round value to specified decimal precision."""
multiplier = pow(10, precision)
return math.floor(value * multiplier) / multiplier
def _check_balance(self, symbol: str, side: str, quantity: float) -> bool:
"""Verify sufficient account balance."""
quote = symbol[-4:] # USDT, BUSD, etc.
base = symbol[:-4]
asset = base if side.upper() == "BUY" else quote
required = quantity if side.upper() == "BUY" else quantity * self._get_market_price(symbol)
balance = self.client.get_balance(asset)
return balance >= required
def _get_market_price(self, symbol: str) -> float:
"""Get current market price for symbol."""
ticker = self.client.get_ticker(symbol)
return float(ticker["lastPrice"])
Usage
validator = OrderValidator(binance_client)
Validate before placing order
result = validator.validate_order(
symbol="BTCUSDT",
side="BUY",
quantity=0.001,
price=42000.50
)
if result["valid"]:
print(f"Order params validated: {result['order_params']}")
else:
print(f"Validation errors: {result['errors']}")
6. Monitoring & Alerting
6.1. Health Check Dashboard
Một bot không có monitoring giống như xe không có đồng hồ — bạn không biết mình đang đi đâu cho đến khi quá muộn. Mình sử dụng comprehensive health monitoring:
import time
import psutil
from dataclasses import dataclass, field
from typing import Dict, List
from collections import deque
import threading
@dataclass
class HealthMetrics:
"""Real-time health metrics for trading bot."""
timestamp: float
api_latency_ms: float
error_rate_percent: float
orders_pending: int
orders_filled: int
orders_failed: int
balance_usd: float
cpu_percent: float
memory_percent: float
class TradingBotMonitor:
"""
Comprehensive monitoring for trading bot health.
Tracks metrics, detects anomalies, and triggers alerts.
"""
def __init__(self, alert_threshold: dict = None):
self.metrics_history: deque = deque(maxlen=1000)
self.alert_threshold = alert_threshold or {
"max_latency_ms": 500,
"max_error_rate": 5.0,
"max_pending_orders": 10,
"memory_threshold": 85.0
}
self.start_time = time.time()
self.lock = threading.Lock()
# Metrics
self.total_requests = 0
self.total_errors = 0
self.latencies = deque(maxlen=100)
self.order_stats = {"pending": 0, "filled": 0, "failed": 0}
def record_request(self, latency_ms: float, success: bool):
"""Record API request metrics."""
with self.lock:
self.total_requests += 1
self.latencies.append(latency_ms)
if not success:
self.total_errors += 1
self._check_alerts()
def record_order(self, status: str):
"""Record order status changes."""
with self.lock:
if status == "pending":
self.order_stats["pending"] += 1
elif status == "filled":
self.order_stats["filled"] += 1
self.order_stats["pending"] -= 1
elif status == "failed":
self.order_stats["failed"] += 1
self.order_stats["pending"] -= 1
def get_current_health(self) -> HealthMetrics:
"""Get current system health metrics."""
avg_latency = sum(self.latencies) / len(self.latencies) if self.latencies else 0
error_rate = (self.total_errors / self.total_requests * 100) if self.total_requests > 0 else 0
return HealthMetrics(
timestamp=time.time(),
api_latency_ms=avg_latency,
error_rate_percent=error_rate,
orders_pending=self.order_stats["pending"],
orders_filled=self.order_stats["filled"],
orders_failed=self.order_stats["failed"],
balance_usd=self._get_account_balance(),
cpu_percent=psutil.cpu_percent(interval=0.1),
memory_percent=psutil.virtual_memory().percent
)
def get_health_report(self) -> dict:
"""Generate comprehensive health report."""
health = self.get_current_health()
uptime_hours = (time.time() - self.start_time) / 3600
return {
"status": self._determine_status(health),
"uptime_hours": round(uptime_hours, 2),
"api_latency_ms": round(health.api_latency_ms, 2),
"error_rate_%": round(health.error_rate_percent, 2),
"orders": {
"filled": health.orders_filled,
"failed": health.orders_failed,
"pending": health.orders_pending,
"success_rate_%": round(
health.orders_filled / max(1, health.orders_filled + health.orders_failed) * 100, 2
)
},
"system": {
"cpu_%": round(health.cpu_percent, 1),
"memory_%": round(health.memory_percent, 1)
},
"recommendations": self._generate_recommendations(health)
}
def _determine_status(self, health: HealthMetrics) -> str:
"""Determine overall bot status."""
if health.error_rate_percent > self.alert_threshold["max_error_rate"]:
return "CRITICAL"
if health.api_latency_ms > self.alert_threshold["max_latency_ms"]:
return "DEGRADED"
if health.memory_percent > self.alert_threshold["memory_threshold"]:
return "WARNING"
return "HEALTHY"
def _check_alerts(self):
"""Check if any metric