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:

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ụ