Thị trường crypto derivatives giao dịch 24/7, và năm 2025, khối lượng hợp đồng tương lai trên Bybit đã vượt 280 tỷ USD mỗi ngày. Tôi đã chứng kiến rất nhiều đội ngũ quant Việt Nam xây dựng bot giao dịch chỉ để rồi thất bại vì thiếu một framework xử lý dữ liệu thông minh. Bài viết này sẽ hướng dẫn bạn xây dựng LangChain multi-Agent quantitative framework với dữ liệu Bybit real-time, tích hợp AI analysis hoàn toàn qua HolySheep AI — nền tảng API AI tiết kiệm 85%+ chi phí so với các provider phương Tây.
Tại sao cần Multi-Agent Framework cho Quantitative Trading?
Traditional quant bot chỉ có 1 pipeline đơn luồng: fetch data → calculate indicators → execute trade. Cách tiếp cận này có 3 vấn đề nghiêm trọng:
- Latency chết người: Khi market volatility cao, 1 giây trễ có thể mất 2-5% spread
- Thông tin rời rạc: Bot không hiểu correlation giữa các indicators hoặc news sentiment
- Không có fallback: Khi API fails, toàn bộ hệ thống dừng
Multi-Agent architecture giải quyết bằng cách tách biệt concerns:
- Data Agent: Fetch và preprocess dữ liệu từ Bybit WebSocket
- Analysis Agent: Chạy technical analysis và pattern recognition
- Risk Agent: Kiểm tra position size, drawdown limits
- Execution Agent: Quản lý order execution và retry logic
- Reporting Agent: Tạo daily P&L reports và alerts
Kiến trúc tổng thể
┌─────────────────────────────────────────────────────────────────┐
│ LangChain Multi-Agent Framework │
├─────────────────────────────────────────────────────────────────┤
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Data │ │ Analysis │ │ Risk │ │Execution │ │
│ │ Agent │──│ Agent │──│ Agent │──│ Agent │ │
│ │ │ │ │ │ │ │ │ │
│ │ Bybit WS │ │HolySheep │ │ Position │ │ Order │ │
│ │ Parser │ │ AI API │ │ Check │ │ Manager │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Shared State Manager (Redis/DB) │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘Cài đặt môi trường và dependencies
# Requirements - Python 3.10+
pip install langchain langchain-core langgraph
pip install bybit-api websockets pandas numpy
pip install redis asyncpg # Shared state
pip install python-dotenv asyncio-lock
Project structure
bybit-quant-agents/
├── agents/
│ ├── __init__.py
│ ├── data_agent.py
│ ├── analysis_agent.py
│ ├── risk_agent.py
│ └── execution_agent.py
├── config/
│ └── settings.py
├── core/
│ ├── state_manager.py
│ └── llm_client.py
├── main.py
└── requirements.txtCấu hình HolySheep AI Client
Toàn bộ LLM calls sẽ đi qua HolySheep AI với chi phí chỉ từ $0.42/1M tokens (DeepSeek V3.2) — rẻ hơn 95% so với GPT-4o ($8/1M tokens). Đăng ký và lấy API key ngay để bắt đầu.
# core/llm_client.py
import os
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
from dotenv import load_dotenv
load_dotenv()
=== QUAN TRỌNG: Sử dụng HolySheep AI thay vì OpenAI ===
Giá HolySheep 2026: DeepSeek V3.2 chỉ $0.42/1M tokens (tiết kiệm 95%+)
So sánh: GPT-4.1 $8, Claude Sonnet 4.5 $15, Gemini 2.5 Flash $2.50
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class HolySheepLLM:
"""Wrapper cho HolySheep AI - tương thích OpenAI SDK"""
def __init__(self, model: str = "deepseek-v3.2"):
self.model = model
self.client = ChatOpenAI(
model=model,
openai_api_key=HOLYSHEEP_API_KEY,
openai_api_base=HOLYSHEEP_BASE_URL,
max_tokens=2048,
temperature=0.3 # Low temp cho quantitative analysis
)
def analyze_market(self, market_data: dict, context: str) -> str:
"""Phân tích market data và đưa ra trading signals"""
system_prompt = """Bạn là một quantitative analyst chuyên nghiệp.
Phân tích dữ liệu thị trường và đưa ra:
1. Trend direction (bullish/bearish/neutral)
2. Key support/resistance levels
3. Entry points với risk/reward ratio
4. Position sizing recommendation
Chỉ phản hồi bằng JSON format."""
user_prompt = f"""
Market Data:
{market_data}
Context: {context}
Trả lời bằng JSON."""
response = self.client([
SystemMessage(content=system_prompt),
HumanMessage(content=user_prompt)
])
return response.content
def generate_trading_plan(self, signals: list, portfolio: dict) -> dict:
"""Tạo trading plan từ multiple signals"""
prompt = f"""
Signals từ các indicators:
{signals}
Current Portfolio:
{portfolio}
Tạo consolidated trading plan với:
- Final recommendation (BUY/SELL/HOLD)
- Position size (% of portfolio)
- Stop loss và take profit levels
- Risk assessment
Trả lời bằng JSON."""
response = self.client([HumanMessage(content=prompt)])
return response.content
Singleton instance
llm = HolySheepLLM(model="deepseek-v3.2")Data Agent - Kết nối Bybit WebSocket
# agents/data_agent.py
import asyncio
import json
from datetime import datetime
from typing import Dict, List, Optional
from bybit_api import BybitWebSocket
import pandas as pd
class BybitDataAgent:
"""Agent xử lý real-time data từ Bybit"""
def __init__(self, state_manager):
self.ws = None
self.state_manager = state_manager
self.subscriptions = [
"orderbook.50.BTCUSDT",
"publicTrade.BTCUSDT",
"ticker.BTCUSDT"
]
self.orderbook_cache = {}
self.trade_cache = []
self.max_trades = 1000
async def connect(self):
"""Khởi tạo WebSocket connection"""
self.ws = BybitWebSocket(
testnet=False,
channel_types=["orderbook", "trade", "ticker"]
)
for sub in self.subscriptions:
self.ws.subscribe(sub)
await asyncio.create_task(self._listen())
async def _listen(self):
"""Listen và xử lý incoming messages"""
while True:
try:
data = await self.ws.fetch()
await self._process_message(data)
except Exception as e:
print(f"WebSocket error: {e}")
await asyncio.sleep(5) # Reconnect sau 5s
async def _process_message(self, data: dict):
"""Parse và store data theo type"""
topic = data.get("topic", "")
if "orderbook" in topic:
await self._handle_orderbook(data)
elif "publicTrade" in topic:
await self._handle_trade(data)
elif "ticker" in topic:
await self._handle_ticker(data)
# Update shared state
await self.state_manager.update("latest_data", {
"orderbook": self.orderbook_cache,
"recent_trades": self.trade_cache[-100:],
"timestamp": datetime.now().isoformat()
})
async def _handle_orderbook(self, data: dict):
"""Xử lý orderbook updates - critical cho liquidity analysis"""
payload = data.get("data", {})
self.orderbook_cache = {
"bids": [(float(p), float(q)) for p, q in payload.get("b", [])],
"asks": [(float(p), float(q)) for p, q in payload.get("a", [])],
"timestamp": data.get("ts", 0)
}
# Tính spread
if self.orderbook_cache["bids"] and self.orderbook_cache["asks"]:
best_bid = self.orderbook_cache["bids"][0][0]
best_ask = self.orderbook_cache["asks"][0][0]
spread_pct = (best_ask - best_bid) / best_bid * 100
await self.state_manager.update("spread", spread_pct)
async def _handle_trade(self, data: dict):
"""Xử lý trade data - phát hiện large trades/whale activity"""
trades = data.get("data", [])
for trade in trades:
trade_info = {
"price": float(trade["p"]),
"volume": float(trade["v"]),
"side": trade["S"], # Buy/Sell
"timestamp": int(trade["T"]),
"value_usd": float(trade["p"]) * float(trade["v"])
}
self.trade_cache.append(trade_info)
# Detect whale trades (> $100k)
if trade_info["value_usd"] > 100_000:
await self._emit_whale_alert(trade_info)
# Giới hạn cache size
if len(self.trade_cache) > self.max_trades:
self.trade_cache = self.trade_cache[-self.max_trades:]
async def _handle_ticker(self, data: dict):
"""Xử lý 24h ticker data"""
ticker = data.get("data", {})
await self.state_manager.update("ticker", {
"last_price": float(ticker.get("last", 0)),
"high_24h": float(ticker.get("high_24h", 0)),
"low_24h": float(ticker.get("low_24h", 0)),
"volume_24h": float(ticker.get("volume_24h", 0)),
"turnover_24h": float(ticker.get("turnover_24h", 0)),
"funding_rate": float(ticker.get("funding_rate", 0))
})
async def _emit_whale_alert(self, trade: dict):
"""Emit event khi phát hiện whale activity"""
print(f"🐋 WHALE ALERT: ${trade['value_usd']:,.0f} {trade['side']} @ ${trade['price']:,}")
await self.state_manager.emit("whale_alert", trade)
def get_market_summary(self) -> dict:
"""Trả về market data summary cho Analysis Agent"""
return {
"orderbook": self.orderbook_cache,
"recent_trades": self.trade_cache[-50:],
"trade_stats": self._calculate_trade_stats()
}
def _calculate_trade_stats(self) -> dict:
"""Tính trade statistics"""
if not self.trade_cache:
return {}
recent = [t for t in self.trade_cache if t["timestamp"] > (datetime.now().timestamp() * 1000 - 300000)]
buy_volume = sum(t["value_usd"] for t in recent if t["side"] == "Buy")
sell_volume = sum(t["value_usd"] for t in recent if t["side"] == "Sell")
return {
"buy_ratio": buy_volume / (buy_volume + sell_volume) if (buy_volume + sell_volume) > 0 else 0.5,
"total_volume_5m": buy_volume + sell_volume,
"trade_count_5m": len(recent)
}Analysis Agent - AI-Powered Market Analysis
# agents/analysis_agent.py
import pandas as pd
import numpy as np
from typing import Dict, List
from datetime import datetime, timedelta
from core.llm_client import llm
class AnalysisAgent:
"""Agent phân tích market sử dụng HolySheep AI"""
def __init__(self, state_manager):
self.state_manager = state_manager
self.indicators_cache = {}
async def analyze(self) -> dict:
"""Chạy full analysis pipeline"""
# 1. Fetch dữ liệu từ state
latest_data = await self.state_manager.get("latest_data")
ticker = await self.state_manager.get("ticker")
# 2. Calculate technical indicators
indicators = await self._calculate_indicators(latest_data)
# 3. AI-powered pattern recognition
ai_analysis = await self._ai_pattern_analysis(indicators, ticker)
# 4. Generate signals
signals = self._generate_signals(indicators, ai_analysis)
# 5. Store results
await self.state_manager.update("analysis", {
"indicators": indicators,
"ai_analysis": ai_analysis,
"signals": signals,
"timestamp": datetime.now().isoformat()
})
return signals
async def _calculate_indicators(self, data: dict) -> dict:
"""Tính toán technical indicators từ raw data"""
trades = data.get("recent_trades", [])
if len(trades) < 20:
return {}
df = pd.DataFrame(trades)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df = df.set_index('timestamp').sort_index()
# Price-based indicators
prices = df['price']
# Moving Averages
ma_7 = prices.rolling('7T').mean().iloc[-1]
ma_25 = prices.rolling('25T').mean().iloc[-1]
ma_99 = prices.rolling('99T').mean().iloc[-1]
# RSI (14-period)
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
rsi_current = rsi.iloc[-1] if not rsi.empty else 50
# Bollinger Bands
bb_20 = prices.rolling(20)
bb_std = bb_20.std().iloc[-1]
bb_mid = bb_20.mean().iloc[-1]
bb_upper = bb_mid + (bb_std * 2)
bb_lower = bb_mid - (bb_std * 2)
# VWAP approximation
vwap = (df['price'] * df['volume']).sum() / df['volume'].sum()
return {
"price": prices.iloc[-1],
"ma_7": ma_7,
"ma_25": ma_25,
"ma_99": ma_99,
"rsi": rsi_current,
"bb_upper": bb_upper,
"bb_mid": bb_mid,
"bb_lower": bb_lower,
"vwap": vwap,
"volatility": bb_std / bb_mid if bb_mid > 0 else 0,
"trade_stats": data.get("trade_stats", {})
}
async def _ai_pattern_analysis(self, indicators: dict, ticker: dict) -> str:
"""Sử dụng HolySheep AI để phân tích patterns"""
market_data = {
"current_price": indicators.get("price", 0),
"ma_7": indicators.get("ma_7", 0),
"ma_25": indicators.get("ma_25", 0),
"ma_99": indicators.get("ma_99", 0),
"rsi": indicators.get("rsi", 50),
"bb_position": self._calculate_bb_position(indicators),
"funding_rate": ticker.get("funding_rate", 0),
"24h_high": ticker.get("high_24h", 0),
"24h_low": ticker.get("low_24h", 0),
"volume_24h": ticker.get("volume_24h", 0),
"buy_ratio": indicators.get("trade_stats", {}).get("buy_ratio", 0.5)
}
context = f"""
Timeframe: 5-minute
Funding Rate: {market_data['funding_rate']:.4f}%
24h Change: {((market_data['current_price'] - market_data['24h_low']) / market_data['24h_low'] * 100):.2f}%
"""
try:
analysis = llm.analyze_market(market_data, context)
return analysis
except Exception as e:
print(f"AI Analysis Error: {e}")
return '{"error": "AI analysis unavailable"}'
def _calculate_bb_position(self, indicators: dict) -> float:
"""Tính vị trí giá trong Bollinger Bands (0-100)"""
price = indicators.get("price", 0)
upper = indicators.get("bb_upper", 0)
lower = indicators.get("bb_lower", 0)
if upper == lower:
return 50
return (price - lower) / (upper - lower) * 100
def _generate_signals(self, indicators: dict, ai_analysis: str) -> List[dict]:
"""Tổng hợp signals từ multiple sources"""
signals = []
price = indicators.get("price", 0)
# Signal 1: MA Crossover
ma_7 = indicators.get("ma_7", 0)
ma_25 = indicators.get("ma_25", 0)
ma_99 = indicators.get("ma_99", 0)
if ma_7 > ma_25 > ma_99:
signals.append({
"type": "MA_CROSSOVER",
"direction": "BUY",
"strength": 0.8,
"reason": "Golden cross - MA7 > MA25 > MA99"
})
elif ma_7 < ma_25 < ma_99:
signals.append({
"type": "MA_CROSSOVER",
"direction": "SELL",
"strength": 0.8,
"reason": "Death cross - MA7 < MA25 < MA99"
})
# Signal 2: RSI
rsi = indicators.get("rsi", 50)
if rsi < 30:
signals.append({
"type": "RSI_OVERSOLD",
"direction": "BUY",
"strength": 0.7,
"reason": f"RSI oversold at {rsi:.1f}"
})
elif rsi > 70:
signals.append({
"type": "RSI_OVERBOUGHT",
"direction": "SELL",
"strength": 0.7,
"reason": f"RSI overbought at {rsi:.1f}"
})
# Signal 3: Bollinger Band Squeeze
bb_pos = self._calculate_bb_position(indicators)
volatility = indicators.get("volatility", 0)
if bb_pos < 20:
signals.append({
"type": "BB_LOWER_TOUCH",
"direction": "BUY",
"strength": 0.6,
"reason": f"Price at lower BB ({bb_pos:.1f}%)"
})
elif bb_pos > 80:
signals.append({
"type": "BB_UPPER_TOUCH",
"direction": "SELL",
"strength": 0.6,
"reason": f"Price at upper BB ({bb_pos:.1f}%)"
})
return signalsRisk Agent - Quản lý rủi ro chặt chẽ
# agents/risk_agent.py
from typing import Dict, Optional
from datetime import datetime
import asyncio
class RiskAgent:
"""Agent quản lý rủi ro - bắt buộc trước mọi execution"""
def __init__(self, state_manager):
self.state_manager = state_manager
self.max_position_pct = 0.1 # 10% max position
self.max_daily_loss_pct = 0.05 # 5% max daily loss
self.max_leverage = 10
self.max_drawdown_pct = 0.15 # 15% max drawdown
async def check_risk(self, proposed_trade: dict) -> Dict:
"""Kiểm tra trade proposal against risk rules"""
checks = []
passed = True
# 1. Position size check
position_check = await self._check_position_size(proposed_trade)
checks.append(position_check)
if not position_check["passed"]:
passed = False
# 2. Daily loss check
daily_loss_check = await self._check_daily_loss()
checks.append(daily_loss_check)
if not daily_loss_check["passed"]:
passed = False
# 3. Drawdown check
drawdown_check = await self._check_drawdown()
checks.append(drawdown_check)
if not drawdown_check["passed"]:
passed = False
# 4. Leverage check
leverage_check = await self._check_leverage(proposed_trade)
checks.append(leverage_check)
if not leverage_check["passed"]:
passed = False
# 5. Correlation check (prevent over-concentration)
correlation_check = await self._check_correlation(proposed_trade)
checks.append(correlation_check)
return {
"approved": passed,
"checks": checks,
"rejection_reasons": [c["reason"] for c in checks if not c["passed"]]
}
async def _check_position_size(self, trade: dict) -> dict:
"""Kiểm tra position size không vượt limit"""
portfolio = await self.state_manager.get("portfolio", {})
current_position_value = portfolio.get("position_value", 0)
total_value = portfolio.get("total_value", 100_000)
proposed_size = trade.get("size", 0)
proposed_value = proposed_size * trade.get("price", 0)
new_total_position_value = current_position_value + proposed_value
position_pct = new_total_position_value / total_value
if position_pct > self.max_position_pct:
return {
"check": "POSITION_SIZE",
"passed": False,
"reason": f"Position size {position_pct:.1%} exceeds max {self.max_position_pct:.1%}",
"current": position_pct,
"limit": self.max_position_pct
}
return {
"check": "POSITION_SIZE",
"passed": True,
"position_pct": position_pct
}
async def _check_daily_loss(self) -> dict:
"""Kiểm tra daily P&L không vượt limit"""
daily_pnl = await self.state_manager.get("daily_pnl", 0)
portfolio = await self.state_manager.get("portfolio", {})
total_value = portfolio.get("total_value", 100_000)
daily_loss_pct = abs(daily_pnl) / total_value if daily_pnl < 0 else 0
if daily_pnl < 0 and daily_loss_pct > self.max_daily_loss_pct:
return {
"check": "DAILY_LOSS",
"passed": False,
"reason": f"Daily loss {daily_loss_pct:.1%} exceeds max {self.max_daily_loss_pct:.1%}",
"current_loss": daily_loss_pct
}
return {
"check": "DAILY_LOSS",
"passed": True,
"daily_loss_pct": daily_loss_pct
}
async def _check_drawdown(self) -> dict:
"""Kiểm tra max drawdown"""
peak_value = await self.state_manager.get("peak_value", 100_000)
current_value = await self.state_manager.get("portfolio", {}).get("total_value", 100_000)
drawdown = (peak_value - current_value) / peak_value if peak_value > 0 else 0
if drawdown > self.max_drawdown_pct:
return {
"check": "DRAWDOWN",
"passed": False,
"reason": f"Drawdown {drawdown:.1%} exceeds max {self.max_drawdown_pct:.1%}",
"current_drawdown": drawdown
}
return {
"check": "DRAWDOWN",
"passed": True,
"current_drawdown": drawdown
}
async def _check_leverage(self, trade: dict) -> dict:
"""Kiểm tra leverage không vượt limit"""
leverage = trade.get("leverage", 1)
if leverage > self.max_leverage:
return {
"check": "LEVERAGE",
"passed": False,
"reason": f"Leverage {leverage}x exceeds max {self.max_leverage}x",
"requested": leverage,
"max": self.max_leverage
}
return {
"check": "LEVERAGE",
"passed": True,
"leverage": leverage
}
async def _check_correlation(self, trade: dict) -> dict:
"""Prevent over-concentration in correlated assets"""
# Simplified check - in production, use correlation matrix
portfolio = await self.state_manager.get("portfolio", {})
positions = portfolio.get("positions", {})
symbol = trade.get("symbol", "")
current_size = positions.get(symbol, {}).get("size", 0)
# For BTCUSDT, check if already have large ETH position
# In real implementation, use correlation coefficients
return {
"check": "CORRELATION",
"passed": True,
"correlation_risk": "LOW"
}
async def calculate_position_size(self, entry_price: float, stop_loss: float,
risk_pct: float = 0.02) -> dict:
"""Kelly Criterion based position sizing"""
portfolio = await self.state_manager.get("portfolio", {})
total_value = portfolio.get("total_value", 100_000)
risk_amount = total_value * risk_pct
price_risk = abs(entry_price - stop_loss)
if price_risk == 0:
return {"size": 0, "reason": "Invalid stop loss"}
raw_size = risk_amount / price_risk
# Apply position size limit
max_size_value = total_value * self.max_position_pct
max_size = max_size_value / entry_price
final_size = min(raw_size, max_size)
return {
"size": final_size,
"entry_price": entry_price,
"stop_loss": stop_loss,
"risk_amount": risk_amount,
"risk_pct": risk_amount / total_value,
"leverage_recommended": final_size * entry_price / total_value
}Execution Agent - Order Management với Retry Logic
# agents/execution_agent.py
import asyncio
from typing import Dict, Optional
from datetime import datetime
from bybit_api import BybitRestClient
from core.llm_client import llm
class ExecutionAgent:
"""Agent quản lý order execution với retry và fallback"""
def __init__(self, state_manager):
self.state_manager = state_manager
self.bybit = BybitRestClient(
key=os.getenv("BYBIT_API_KEY"),
secret=os.getenv("BYBIT_API_SECRET"),
testnet=False
)
self.max_retries = 3
self.retry_delay = 1 # seconds
self.order_cache = {}
async def execute_trade(self, trade_plan: dict) -> Dict:
"""Execute trade với comprehensive error handling"""
symbol = trade_plan.get("symbol", "BTCUSDT")
direction = trade_plan.get("direction") # BUY or SELL
size = trade_plan.get("size", 0)
leverage = trade_plan.get("leverage", 1)
if size <= 0:
return {"status": "REJECTED", "reason": "Invalid size"}
# Set leverage trước
await self._set_leverage(symbol, leverage)
# Prepare order
order_params = {
"symbol": symbol,
"side": direction,
"order_type": "Market",
"qty": size,
"time_in_force": "GTC"
}
# Execute với retry
result = await self._execute_with_retry(order_params)
# Update portfolio state
if result["status"] == "FILLED":
await self._update_portfolio(trade_plan, result)
await self._emit_execution_report(result)
return result
async def _execute_with_retry(self, order_params: dict) -> Dict:
"""Execute order với exponential backoff retry"""
for attempt in range(self.max_retries):
try:
# Submit order
response = await self.bybit.place_order(**order_params)
if response.get("ret_code") == 0:
order_id = response["result"]["order_id"]
# Wait for fill
fill_result = await self._wait_for_fill(order_id)
if fill_result["status"] == "FILLED":
return {
"status": "FILLED",
"order_id": order_id,
"fill_price": fill_result["avg_price"],
"fill_qty": fill_result["qty"],
"attempts": attempt + 1
}
else:
# Cancel và retry
await self.bybit.cancel_order(order_id)
continue
elif response.get("ret_code") == 10001:
# Rate limit - wait longer
await asyncio.sleep(self.retry_delay * (2 ** attempt) + 5)
continue
else:
return {
"status": "REJECTED",
"reason": response.get("ret_msg"),
"ret_code": response.get("ret_code")
}
except Exception as e:
print(f"Execution attempt {attempt + 1} failed: {e}")
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delay * (2 ** attempt))
else:
return {
"status": "FAILED",
"reason": str(e),
"attempts": attempt + 1
}
return {
"status": "FAILED",
"reason": "Max retries exceeded"
}
async def _wait_for_fill(self, order_id: str, timeout: int = 30) -> Dict:
"""Chờ order fill với polling"""
start_time = datetime.now()
while (datetime.now() - start_time).seconds < timeout:
try:
order_status = await self.bybit.get_order(order_id)
if order_status["result"]["order_status"] == "Filled":
return {
"status": "FILLED",
"avg_price": order_status["result"]["avg_price"],
"qty": order_status["result"]["qty"]
}
elif order_status["result"]["order_status"] in ["Cancelled", "Rejected"]:
return {
"status": order_status["result"]["order_status"]
}
await asyncio.sleep(0.5)
except Exception as e:
print(f"Fill check error: {e}")
await asyncio.sleep(1)
return {"status": "PENDING", "timeout": True}
async def _set_leverage(self, symbol: str, leverage: int):
"""Set position leverage"""
try:
await self.bybit.set_leverage(symbol=symbol, leverage=leverage)
except Exception as e:
print(f"Leverage setting warning: {e}")
async def _update_portfolio(self, trade_plan: dict, fill_result: dict):
"""Cập nhật portfolio state sau khi fill"""
portfolio = await self.state_manager.get("portfolio", {})
positions = portfolio.get("positions", {})
symbol = trade_plan["symbol"]
direction = trade_plan["direction"]
fill_price = fill_result["fill_price"]
fill_qty = fill_result["fill_qty"]