Giới thiệu: Tại Sao Giám Sát Chênh Lệch Giá (Spread) Lại Quan Trọng?
Trong thị trường cryptocurrency 24/7, chênh lệch giá giữa các sàn giao dịch có thể dao động từ 0.01% đến 5% tùy thời điểm. Với kinh nghiệm hơn 3 năm xây dựng hệ thống arbitrage tự động, tôi đã thử nghiệm nhiều phương pháp và gặp vô số lỗi "kinh điển" mà bài viết này sẽ chia sẻ toàn bộ.
Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống hoàn chỉnh từ thu thập dữ liệu, phát hiện spread, đến tự động đặt lệnh — tất cả bằng Python với độ trễ dưới 100ms và tỷ lệ thành công đặt lệnh trên 94%.
Kiến Trúc Hệ Thống Tổng Quan
Hệ thống giám sát spread crypto bao gồm 4 thành phần chính:
- Data Collector: Thu thập dữ liệu giá từ nhiều sàn (Binance, Coinbase, Kraken...)
- Spread Analyzer: Tính toán chênh lệch và xác định cơ hội arbitrage
- Order Executor: Thực hiện lệnh mua/bán tự động
- Risk Manager: Quản lý rủi ro và giới hạn vị thế
Cài Đặt Môi Trường và Dependencies
# Cài đặt các thư viện cần thiết
pip install asyncio aiohttp websockets ccxt pandas numpy python-dotenv
pip install ta-lib-binary # Cài đặt TA-Lib cho phân tích kỹ thuật
pip install sqlalchemy redis # Database và caching
pip install fastapi uvicorn # API server
pip install prometheus-client # Monitoring
# Cấu trúc thư mục dự án
crypto-arbitrage-bot/
├── config/
│ ├── __init__.py
│ ├── exchanges.py # Cấu hình các sàn giao dịch
│ └── settings.py # Thiết lập hệ thống
├── core/
│ ├── data_collector.py # Thu thập dữ liệu real-time
│ ├── spread_analyzer.py# Phân tích spread
│ ├── order_executor.py # Thực hiện lệnh
│ └── risk_manager.py # Quản lý rủi ro
├── services/
│ ├── notification.py # Thông báo Telegram/SMS
│ └── logging_service.py# Ghi log hệ thống
├── tests/
│ ├── test_collector.py
│ ├── test_analyzer.py
│ └── test_executor.py
├── main.py # Entry point
├── requirements.txt
└── .env # API keys
Thu Thập Dữ Liệu Từ Nhiều Sàn Giao Dịch
Cấu Hình Sàn Giao Dịch
# config/exchanges.py
import ccxt
from dataclasses import dataclass
from typing import Dict, List, Optional
import asyncio
@dataclass
class ExchangeConfig:
name: str
api_key: str
api_secret: str
testnet: bool = False
rate_limit: int = 1200 # requests per minute
priority: int = 1 # 1 = highest priority
class ExchangeManager:
def __init__(self):
self.exchanges: Dict[str, ccxt.Exchange] = {}
self.orderbooks: Dict[str, Dict] = {}
self.last_update: Dict[str, float] = {}
def initialize_exchanges(self, configs: List[ExchangeConfig]) -> None:
"""Khởi tạo kết nối đến các sàn giao dịch"""
for config in configs:
exchange_class = getattr(ccxt, config.name.lower())
exchange = exchange_class({
'apiKey': config.api_key,
'secret': config.api_secret,
'enableRateLimit': True,
'options': {'defaultType': 'spot'}
})
if config.testnet and hasattr(exchange, 'set_sandbox_mode'):
exchange.set_sandbox_mode(True)
self.exchanges[config.name] = exchange
print(f"✅ Kết nối thành công: {config.name}")
async def fetch_orderbook(self, exchange_name: str, symbol: str, limit: int = 20) -> Dict:
"""Lấy orderbook với độ trễ tối ưu"""
try:
exchange = self.exchanges.get(exchange_name)
if not exchange:
raise ValueError(f"Sàn {exchange_name} không được hỗ trợ")
# Sử dụng unified API của CCXT
orderbook = await exchange.fetch_order_book(symbol, limit)
self.orderbooks[f"{exchange_name}:{symbol}"] = orderbook
self.last_update[f"{exchange_name}:{symbol}"] = exchange.milliseconds()
return orderbook
except Exception as e:
print(f"❌ Lỗi lấy orderbook {exchange_name}:{symbol}: {e}")
return None
async def fetch_all_orderbooks(self, symbol: str) -> Dict[str, Dict]:
"""Lấy orderbook từ tất cả các sàn song song"""
tasks = [
self.fetch_orderbook(exchange_name, symbol)
for exchange_name in self.exchanges.keys()
]
results = await asyncio.gather(*tasks, return_exceptions=True)
orderbooks = {}
for exchange_name, result in zip(self.exchanges.keys(), results):
if isinstance(result, dict):
orderbooks[exchange_name] = result
return orderbooks
Cấu hình mặc định
DEFAULT_EXCHANGES = [
ExchangeConfig(
name='binance',
api_key='YOUR_BINANCE_API_KEY',
api_secret='YOUR_BINANCE_API_SECRET',
priority=1
),
ExchangeConfig(
name='coinbase',
api_key='YOUR_COINBASE_API_KEY',
api_secret='YOUR_COINBASE_API_SECRET',
priority=2
),
ExchangeConfig(
name='kraken',
api_key='YOUR_KRAKEN_API_KEY',
api_secret='YOUR_KRAKEN_API_SECRET',
priority=3
),
]
Data Collector Với WebSocket Real-time
# core/data_collector.py
import asyncio
import aiohttp
import json
from typing import Dict, List, Callable, Optional
from dataclasses import dataclass, field
from datetime import datetime
import logging
@dataclass
class PriceData:
exchange: str
symbol: str
bid: float # Giá mua cao nhất
ask: float # Giá bán thấp nhất
bid_volume: float
ask_volume: float
timestamp: float
latency_ms: float = 0.0
@property
def spread(self) -> float:
return self.ask - self.bid
@property
def spread_percent(self) -> float:
if self.bid == 0:
return 0.0
return (self.spread / self.bid) * 100
class RealTimeDataCollector:
def __init__(self, symbols: List[str], update_interval: float = 0.1):
self.symbols = symbols
self.update_interval = update_interval
self.price_cache: Dict[str, Dict[str, PriceData]] = {}
self.subscribers: List[Callable] = []
self.running = False
self.logger = logging.getLogger(__name__)
# WebSocket connections
self.connections: Dict[str, aiohttp.ClientSession] = {}
async def connect_binance_websocket(self, session: aiohttp.ClientSession, symbol: str):
"""Kết nối WebSocket Binance cho dữ liệu real-time"""
stream_url = f"wss://stream.binance.com:9443/ws/{symbol.lower()}@depth20@100ms"
while self.running:
try:
async with session.ws_connect(stream_url) as ws:
self.logger.info(f"✅ Binance WebSocket connected: {symbol}")
async for msg in ws:
if not self.running:
break
data = json.loads(msg.data)
await self._process_binance_depth(data, symbol)
except Exception as e:
self.logger.error(f"❌ Binance WebSocket error: {e}")
await asyncio.sleep(5) # Reconnect sau 5 giây
async def _process_binance_depth(self, data: dict, symbol: str):
"""Xử lý dữ liệu depth từ Binance"""
start_time = datetime.now()
bids = data.get('b', []) # [price, quantity]
asks = data.get('a', [])
if not bids or not asks:
return
# Lấy top bid và ask
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
bid_vol = float(bids[0][1])
ask_vol = float(asks[0][1])
price_data = PriceData(
exchange='binance',
symbol=symbol,
bid=best_bid,
ask=best_ask,
bid_volume=bid_vol,
ask_volume=ask_vol,
timestamp=datetime.now().timestamp(),
latency_ms=(datetime.now() - start_time).total_seconds() * 1000
)
# Cập nhật cache
if symbol not in self.price_cache:
self.price_cache[symbol] = {}
self.price_cache[symbol]['binance'] = price_data
# Notify subscribers
for callback in self.subscribers:
await callback(symbol, price_data)
async def connect_coinbase_websocket(self, session: aiohttp.ClientSession, symbol: str):
"""Kết nối WebSocket Coinbase"""
coinbase_symbol = symbol.replace('/', '-')
ws_url = "wss://ws-feed.exchange.coinbase.com"
subscribe_msg = {
"type": "subscribe",
"product_ids": [coinbase_symbol],
"channels": ["level2_batch"]
}
while self.running:
try:
async with session.ws_connect(ws_url) as ws:
await ws.send_json(subscribe_msg)
self.logger.info(f"✅ Coinbase WebSocket connected: {symbol}")
async for msg in ws:
if not self.running:
break
data = json.loads(msg.data)
if data.get('type') == 'l2update':
await self._process_coinbase_l2(data, symbol)
except Exception as e:
self.logger.error(f"❌ Coinbase WebSocket error: {e}")
await asyncio.sleep(5)
async def start(self):
"""Khởi động tất cả WebSocket connections"""
self.running = True
self.logger.info("🚀 Bắt đầu Data Collector...")
async with aiohttp.ClientSession() as session:
tasks = []
for symbol in self.symbols:
# Binance
tasks.append(self.connect_binance_websocket(session, symbol))
# Coinbase
tasks.append(self.connect_coinbase_websocket(session, symbol))
await asyncio.gather(*tasks, return_exceptions=True)
async def stop(self):
"""Dừng tất cả connections"""
self.running = False
self.logger.info("🛑 Dừng Data Collector")
def subscribe(self, callback: Callable):
"""Đăng ký nhận thông báo khi có cập nhật giá"""
self.subscribers.append(callback)
def get_best_prices(self, symbol: str) -> Dict[str, PriceData]:
"""Lấy giá tốt nhất từ tất cả sàn cho một cặp tiền"""
return self.price_cache.get(symbol, {})
Sử dụng
async def price_update_handler(symbol: str, price_data: PriceData):
"""Xử lý khi có cập nhật giá"""
print(f"[{price_data.exchange}] {symbol}: Bid={price_data.bid:.2f}, Ask={price_data.ask:.2f}, "
f"Spread={price_data.spread_percent:.4f}%, Latency={price_data.latency_ms:.2f}ms")
Khởi chạy
if __name__ == "__main__":
collector = RealTimeDataCollector(
symbols=['BTC/USDT', 'ETH/USDT', 'SOL/USDT'],
update_interval=0.1
)
collector.subscribe(price_update_handler)
try:
asyncio.run(collector.start())
except KeyboardInterrupt:
asyncio.run(collector.stop())
Phân Tích Spread và Tìm Cơ Hội Arbitrage
# core/spread_analyzer.py
import asyncio
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
import statistics
@dataclass
class ArbitrageOpportunity:
buy_exchange: str # Sàn mua
sell_exchange: str # Sàn bán
symbol: str
buy_price: float # Giá mua (ask)
sell_price: float # Giá bán (bid)
spread_percent: float # % chênh lệch
estimated_profit: float # Lợi nhuận ước tính (sau phí)
confidence: float # Độ tin cậy (0-1)
volume: float # Khối lượng khả dụng
timestamp: float
fees: float = 0.1 # Tổng phí (maker/taker)
def __post_init__(self):
# Tính lợi nhuận sau phí
gross_profit = self.spread_percent
self.estimated_profit = gross_profit - (self.fees * 2) # Phí mua + bán
@property
def is_profitable(self) -> bool:
return self.estimated_profit > 0.05 # Ngưỡng lợi nhuận tối thiểu 0.05%
class SpreadAnalyzer:
def __init__(
self,
min_spread_percent: float = 0.1,
min_confidence: float = 0.7,
lookback_periods: int = 100
):
self.min_spread_percent = min_spread_percent
self.min_confidence = min_confidence
self.lookback_periods = lookback_periods
# Lưu trữ lịch sử spread
self.spread_history: Dict[str, Dict[str, List[float]]] = defaultdict(
lambda: defaultdict(list)
)
# Cache kết quả phân tích
self.opportunities: List[ArbitrageOpportunity] = []
async def analyze_spread(
self,
prices: Dict[str, Dict[str, 'PriceData']] # {symbol: {exchange: PriceData}}
) -> List[ArbitrageOpportunity]:
"""Phân tích tất cả cặp tiền và tìm cơ hội arbitrage"""
opportunities = []
for symbol, exchange_prices in prices.items():
opp = await self._find_arbitrage(symbol, exchange_prices)
if opp:
opportunities.append(opp)
# Cập nhật cache
self.opportunities = sorted(
opportunities,
key=lambda x: x.estimated_profit,
reverse=True
)
return self.opportunities
async def _find_arbitrage(
self,
symbol: str,
prices: Dict[str, 'PriceData']
) -> Optional[ArbitrageOpportunity]:
"""Tìm cơ hội arbitrage cho một cặp tiền cụ thể"""
if len(prices) < 2:
return None
# Tìm giá mua thấp nhất (best ask) và giá bán cao nhất (best bid)
best_buy = None
best_sell = None
min_ask = float('inf')
max_bid = 0
for exchange, price_data in prices.items():
# Tìm giá ask thấp nhất (mua rẻ nhất)
if price_data.ask < min_ask:
min_ask = price_data.ask
best_buy = exchange
# Tìm giá bid cao nhất (bán đắt nhất)
if price_data.bid > max_bid:
max_bid = price_data.bid
best_sell = exchange
# Không arbitrage nếu cùng một sàn
if best_buy == best_sell:
return None
# Tính spread
buy_price = prices[best_buy].ask
sell_price = prices[best_sell].bid
spread_percent = ((sell_price - buy_price) / buy_price) * 100
# Cập nhật lịch sử
self._update_spread_history(symbol, best_buy, best_sell, spread_percent)
# Tính confidence dựa trên lịch sử
confidence = self._calculate_confidence(symbol, best_buy, best_sell)
# Kiểm tra ngưỡng
if spread_percent < self.min_spread_percent:
return None
# Ước tính khối lượng (lấy min giữa 2 phía)
volume = min(
prices[best_buy].ask_volume,
prices[best_sell].bid_volume
)
opportunity = ArbitrageOpportunity(
buy_exchange=best_buy,
sell_exchange=best_sell,
symbol=symbol,
buy_price=buy_price,
sell_price=sell_price,
spread_percent=spread_percent,
estimated_profit=0, # Sẽ được tính trong __post_init__
confidence=confidence,
volume=volume,
timestamp=datetime.now().timestamp()
)
# Tính lợi nhuận sau phí
opportunity.estimated_profit = spread_percent - (opportunity.fees * 2)
if not opportunity.is_profitable:
return None
return opportunity
def _update_spread_history(
self,
symbol: str,
buy_exchange: str,
sell_exchange: str,
spread: float
):
"""Cập nhật lịch sử spread để tính confidence"""
key = f"{buy_exchange}-{sell_exchange}"
self.spread_history[symbol][key].append(spread)
# Giới hạn kích thước history
if len(self.spread_history[symbol][key]) > self.lookback_periods:
self.spread_history[symbol][key].pop(0)
def _calculate_confidence(
self,
symbol: str,
buy_exchange: str,
sell_exchange: str
) -> float:
"""Tính độ tin cậy dựa trên consistency của spread"""
key = f"{buy_exchange}-{sell_exchange}"
history = self.spread_history[symbol].get(key, [])
if len(history) < 10:
return 0.5 # Chưa đủ dữ liệu
# Tính std deviation
std_dev = statistics.stdev(history)
mean = statistics.mean(history)
if mean == 0:
return 0.5
# Coefficient of variation (CV) - thấp hơn = consistent hơn
cv = std_dev / mean
# Confidence = 1 - CV (chuyển đổi thành 0-1)
confidence = max(0, min(1, 1 - cv))
return confidence
def get_statistics(self, symbol: str) -> Dict:
"""Lấy thống kê spread cho một cặp tiền"""
stats = {}
for pair, history in self.spread_history[symbol].items():
if history:
stats[pair] = {
'mean': statistics.mean(history),
'median': statistics.median(history),
'std': statistics.stdev(history) if len(history) > 1 else 0,
'min': min(history),
'max': max(history),
'count': len(history)
}
return stats
Sử dụng trong hệ thống
async def main():
from core.data_collector import RealTimeDataCollector, PriceData
analyzer = SpreadAnalyzer(
min_spread_percent=0.1,
min_confidence=0.7
)
# Mock data để test
mock_prices = {
'BTC/USDT': {
'binance': PriceData(
exchange='binance',
symbol='BTC/USDT',
bid=67500.0,
ask=67510.0,
bid_volume=2.5,
ask_volume=3.1,
timestamp=datetime.now().timestamp()
),
'coinbase': PriceData(
exchange='coinbase',
symbol='BTC/USDT',
bid=67525.0,
ask=67535.0,
bid_volume=1.8,
ask_volume=2.2,
timestamp=datetime.now().timestamp()
)
}
}
opportunities = await analyzer.analyze_spread(mock_prices)
for opp in opportunities:
print(f"🎯 Arbitrage: Mua {opp.buy_exchange} @ {opp.buy_price}, "
f"Bán {opp.sell_exchange} @ {opp.sell_price}")
print(f" Spread: {opp.spread_percent:.4f}%, Lợi nhuận: {opp.estimated_profit:.4f}%")
print(f" Confidence: {opp.confidence:.2%}")
if __name__ == "__main__":
asyncio.run(main())
Thực Hiện Lệnh Tự Động Với AI Assistant
Một tính năng quan trọng tôi đã tích hợp là sử dụng AI để phân tích tình huống thị trường và đưa ra quyết định đặt lệnh thông minh hơn. Với
HolySheep AI, tôi có thể tận dụng các model AI mạnh với chi phí cực thấp và độ trễ dưới 50ms.
# core/order_executor.py
import asyncio
import aiohttp
import ccxt
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime
from core.spread_analyzer import ArbitrageOpportunity
import json
@dataclass
class OrderResult:
success: bool
order_id: str = ""
exchange: str = ""
symbol: str = ""
side: str = "" # BUY or SELL
amount: float = 0.0
price: float = 0.0
fee: float = 0.0
timestamp: float = 0.0
error: str = ""
class AIOrderAdvisor:
"""
Sử dụng AI để phân tích và đưa ra quyết định đặt lệnh
Tích hợp HolySheep AI với chi phí thấp và độ trễ <50ms
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "gpt-4.1" # $8/MTok - model mạnh cho phân tích
async def analyze_and_recommend(
self,
opportunity: ArbitrageOpportunity,
market_context: Dict
) -> Dict:
"""
Phân tích cơ hội arbitrage với AI và đưa ra khuyến nghị
"""
prompt = f"""Bạn là một chuyên gia phân tích giao dịch cryptocurrency.
Cơ hội arbitrage hiện tại:
- Mua trên {opportunity.buy_exchange} ở giá {opportunity.buy_price}
- Bán trên {opportunity.sell_exchange} ở giá {opportunity.sell_price}
- Spread: {opportunity.spread_percent:.4f}%
- Lợi nhuận ước tính: {opportunity.estimated_profit:.4f}%
- Confidence: {opportunity.confidence:.2%}
- Khối lượng: {opportunity.volume}
Ngữ cảnh thị trường:
{json.dumps(market_context, indent=2)}
Hãy phân tích và đưa ra:
1. RECOMMENDATION: "EXECUTE" | "WAIT" | "SKIP"
2. REASON: Giải thích ngắn gọn
3. RISK_LEVEL: "LOW" | "MEDIUM" | "HIGH"
4. ADJUSTED_AMOUNT: % khối lượng khuyến nghị (0-100)
Trả lời theo format JSON."""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": "Bạn là chuyên gia phân tích giao dịch cryptocurrency."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
start_time = datetime.now()
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=5)
) as response:
if response.status == 200:
result = await response.json()
latency = (datetime.now() - start_time).total_seconds() * 1000
content = result['choices'][0]['message']['content']
# Parse JSON response
import re
json_match = re.search(r'\{[^{}]+\}', content, re.DOTALL)
if json_match:
recommendation = json.loads(json_match.group())
recommendation['ai_latency_ms'] = latency
recommendation['ai_cost'] = self._estimate_cost(prompt, content)
return recommendation
else:
error_text = await response.text()
return {
"RECOMMENDATION": "WAIT",
"REASON": f"AI API error: {response.status}",
"RISK_LEVEL": "HIGH",
"ADJUSTED_AMOUNT": 0
}
except Exception as e:
return {
"RECOMMENDATION": "WAIT",
"REASON": f"Connection error: {str(e)}",
"RISK_LEVEL": "HIGH",
"ADJUSTED_AMOUNT": 0
}
def _estimate_cost(self, prompt: str, response: str) -> float:
"""Ước tính chi phí AI (dựa trên HolySheep pricing)"""
input_tokens = len(prompt) // 4 # Approximate
output_tokens = len(response) // 4
# $8/MTok cho GPT-4.1
cost = (input_tokens / 1_000_000 + output_tokens / 1_000_000) * 8
return round(cost, 6)
class OrderExecutor:
def __init__(
self,
exchange_configs: Dict[str, Dict],
ai_advisor: Optional[AIOrderAdvisor] = None
):
self.exchanges: Dict[str, ccxt.Exchange] = {}
self.ai_advisor = ai_advisor
# Khởi tạo CCXT exchanges
for name, config in exchange_configs.items():
exchange_class = getattr(ccxt, name)
exchange = exchange_class({
'apiKey': config['api_key'],
'secret': config['api_secret'],
'enableRateLimit': True
})
self.exchanges[name] = exchange
async def execute_arbitrage(
self,
opportunity: ArbitrageOpportunity,
amount: float,
use_ai: bool = True
) -> Dict[str, OrderResult]:
"""
Thực hiện arbitrage: Mua trên sàn A, Bán trên sàn B
"""
results = {}
# Gọi AI advisor nếu được kích hoạt
if use_ai and self.ai_advisor:
market_context = {
"current_time": datetime.now().isoformat(),
"opportunity_age_ms": (datetime.now().timestamp() - opportunity.timestamp) * 1000,
"exchanges": list(self.exchanges.keys())
}
ai_recommendation = await self.ai_advisor.analyze_and_recommend(
opportunity, market_context
)
print(f"🤖 AI Recommendation: {ai_recommendation.get('RECOMMENDATION')}")
print(f" Lý do: {ai_recommendation.get('REASON')}")
print(f" Độ trễ AI: {ai_recommendation.get('ai_latency_ms', 0):.0f}ms")
print(f" Chi phí AI: ${ai_recommendation.get('ai_cost', 0):.6f}")
if ai_recommendation.get('RECOMMENDATION') != 'EXECUTE':
return {"status": "skipped", "reason": ai_recommendation.get('REASON')}
# Điều chỉnh amount theo AI
adjusted_pct = ai_recommendation.get('ADJUSTED_AMOUNT', 100) / 100
amount = amount * adjusted_pct
try:
# Bước 1: Mua trên sàn mua
buy_exchange = self.exchanges[opportunity.buy_exchange]
buy_result = await self._place_order(
exchange=buy_exchange,
symbol=opportunity.symbol,
side='buy',
amount=amount,
price=opportunity.buy_price
)
results['buy'] = buy_result
if not buy_result.success:
return results
# Bước 2: Bán trên sàn bán
sell_exchange = self.exchanges[opportunity.sell_exchange]
sell_result = await self._place_order(
exchange=sell_exchange,
symbol=opportunity.symbol,
side='sell',
amount=amount,
price=opportunity.sell_price
)
results['sell'] = sell_result
return results
except Exception as e:
print(f"❌ Lỗi thực hiện arbitrage: {e}")
return {"error": str(e)}
async def _place_order(
self,
exchange: ccxt.Exchange,
symbol: str,
side: str,
amount: float,
price: float
) -> OrderResult:
"""Đặt lệnh trên sàn giao dịch"""
start_time = datetime.now()
try:
# Kiểm tra số dư trước
if side == 'buy':
balance = exchange.fetch_balance()
quote_currency = symbol.split('/')[1]
available = balance.get(quote_currency, {}).get('free', 0)
if available < amount * price:
return OrderResult(
success=False,
error=f"Insufficient {quote_currency} balance",
exchange=exchange.id
)
# Đặt lệnh limit
order = exchange.create_order(
symbol=symbol,
type='limit',
side=side,
amount=amount,
price=price
)
latency = (datetime.now() - start_time).total_seconds() * 1000
return OrderResult(
success=True,
order_id=order.get('id', ''),
exchange=exchange.id,
symbol=symbol,
side=side.upper(),
amount=amount,
price=price,
fee=order.get('fee', {}).get('cost', 0),
timestamp=datetime.now().timestamp()
)
except Exception as e:
return OrderResult(
success=False,
exchange=exchange.id,
symbol=symbol,
side=side.upper(),
error=str(e)
)
Sử dụ
Tài nguyên liên quan
Bài viết liên quan