Tôi vẫn nhớ rõ cái đêm tháng 3 năm 2024, hệ thống giao dịch của mình đang chạy ngon lành thì bất ngờ nhận được ConnectionError: timeout after 30000ms từ API của một nhà cung cấp lớn. Thị trường biến động mạnh, tín hiệu mua xuất hiện nhưng bot không thể gọi AI phân tích được. Kết quả? Mất 47.3 đô la lãi tiềm năng chỉ trong 3 phút. Sau sự cố đó, tôi quyết định xây dựng một hệ thống AI hedge fund với khả năng chịu lỗi cao, sử dụng HolySheep AI làm relay layer — và bài viết này sẽ chia sẻ toàn bộ kiến trúc, code, và bài học xương máu của tôi.
Tại Sao Cần API Relay Cho AI Hedge Fund?
Trong thị trường tài chính, độ trễ dưới 100ms là ranh giới giữa lợi nhuận và thua lỗ. Khi sử dụng trực tiếp API của OpenAI hay Anthropic, bạn đối mặt với:
- Latency không kiểm soát được: Trung bình 800-2000ms cho mỗi request
- Rate limit không dự đoán được: Peak hours có thể bị rejected
- Chi phí không tối ưu: GPT-4o đắt gấp 19 lần so với DeepSeek V3.2
- Single point of failure: Một provider down = toàn bộ hệ thống dừng
HolySheep API relay giải quyết tất cả: với WeChat/Alipay thanh toán, tỷ giá chỉ ¥1=$1, tiết kiệm 85%+, độ trễ dưới 50ms với infrastructure tại châu Á.
Kiến Trúc Hệ Thống AI Hedge Fund
Tổng Quan Architecture
Hệ thống gồm 4 layers chính:
- Data Layer: WebSocket streaming market data
- AI Analysis Layer: HolySheep relay với multi-provider fallback
- Decision Layer: Rule-based + ML hybrid trading engine
- Execution Layer: Broker API integration (Binance, Interactive Brokers)
Mã Nguồn Data Ingestion Layer
import websocket
import json
import asyncio
from typing import List, Dict
from dataclasses import dataclass, field
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class MarketData:
symbol: str
price: float
volume: float
timestamp: datetime = field(default_factory=datetime.now)
bid: float = 0.0
ask: float = 0.0
spread: float = 0.0
def calculate_spread(self) -> float:
if self.bid > 0 and self.ask > 0:
self.spread = (self.ask - self.bid) / self.bid * 100
return self.spread
class MarketDataStream:
"""
Real-time market data ingestion với WebSocket.
Kết nối đến multiple exchanges để đảm bảo redundancy.
"""
def __init__(self, symbols: List[str]):
self.symbols = symbols
self.data_buffer: Dict[str, MarketData] = {}
self.callbacks: List[callable] = []
self.ws_connections: List[websocket.WebSocketApp] = []
self.is_running = False
self.last_message_time: Dict[str, datetime] = {}
self.max_latency_ms = 100 # Alert nếu latency > 100ms
async def connect(self, exchange: str = "binance"):
"""Kết nối WebSocket đến exchange."""
if exchange == "binance":
streams = "/".join([f"{s}@trade" for s in self.symbols])
ws_url = f"wss://stream.binance.com:9443/stream?streams={streams}"
else:
raise ValueError(f"Unsupported exchange: {exchange}")
ws = websocket.WebSocketApp(
ws_url,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
self.ws_connections.append(ws)
logger.info(f"Connected to {exchange} for symbols: {self.symbols}")
def _on_message(self, ws, message):
try:
data = json.loads(message)
if 'data' in data:
trade = data['data']
symbol = trade['s']
market_data = MarketData(
symbol=symbol,
price=float(trade['p']),
volume=float(trade['q']),
timestamp=datetime.fromtimestamp(trade['T'] / 1000),
bid=float(trade.get('b', 0)),
ask=float(trade.get('a', 0))
)
market_data.calculate_spread()
self.data_buffer[symbol] = market_data
self.last_message_time[symbol] = datetime.now()
# Notify all callbacks (AI analysis layer)
for callback in self.callbacks:
asyncio.create_task(self._safe_callback(callback, market_data))
except Exception as e:
logger.error(f"Message parse error: {e}")
async def _safe_callback(self, callback, data):
"""Đảm bảo callback không crash main loop."""
try:
await callback(data)
except Exception as e:
logger.error(f"Callback error: {e}")
def _on_error(self, ws, error):
logger.error(f"WebSocket error: {error}")
# Auto-reconnect logic
asyncio.create_task(self._reconnect(ws))
async def _reconnect(self, ws, max_retries=5):
"""Auto-reconnect với exponential backoff."""
for attempt in range(max_retries):
try:
logger.info(f"Reconnection attempt {attempt + 1}/{max_retries}")
await asyncio.sleep(2 ** attempt)
# Implement reconnection here
break
except Exception as e:
logger.error(f"Reconnect failed: {e}")
def _on_close(self, ws, close_status_code, close_msg):
logger.warning(f"Connection closed: {close_status_code} - {close_msg}")
def _on_open(self, ws):
logger.info("WebSocket connection established")
def register_callback(self, callback: callable):
"""Đăng ký callback để xử lý market data."""
self.callbacks.append(callback)
async def start(self):
"""Bắt đầu streaming data."""
self.is_running = True
for ws in self.ws_connections:
asyncio.create_task(
asyncio.to_thread(ws.run_forever, ping_interval=30)
)
async def health_check(self):
"""Monitor connection health và latency."""
while self.is_running:
await asyncio.sleep(10)
for symbol, last_time in self.last_message_time.items():
latency = (datetime.now() - last_time).total_seconds() * 1000
if latency > self.max_latency_ms:
logger.warning(
f"High latency detected for {symbol}: {latency:.2f}ms"
)
Usage example
async def main():
stream = MarketDataStream(symbols=["BTCUSDT", "ETHUSDT", "SOLUSDT"])
await stream.connect("binance")
stream.register_callback(lambda data: logger.info(f"Got: {data}"))
await stream.start()
await stream.health_check()
if __name__ == "__main__":
asyncio.run(main())HolySheep API Relay Layer — Core Implementation
Đây là trái tim của hệ thống. Tôi đã xây dựng một relay layer thông minh sử dụng HolySheep AI với các tính năng:
- Automatic failover: Nếu DeepSeek fails → tự động chuyển sang Gemini
- Cost optimization: Dùng model rẻ nhất cho từng task
- Response caching: Tránh gọi lại API cho cùng một query
- Rate limit handling: Exponential backoff tự động
import aiohttp
import asyncio
import hashlib
import json
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime, timedelta
import logging
from collections import OrderedDict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AIModel(Enum):
"""Các model được hỗ trợ với pricing 2026."""
GPT_4_1 = "gpt-4.1"
CLAUDE_SONNET_4_5 = "claude-sonnet-4.5"
GEMINI_2_5_FLASH = "gemini-2.5-flash"
DEEPSEEK_V3_2 = "deepseek-v3.2"
@dataclass
class ModelPricing:
"""Pricing structure cho từng model (USD per million tokens)."""
input_cost: float
output_cost: float
latency_p50_ms: float
max_tokens: int
MODEL_PRICING = {
AIModel.GPT_4_1: ModelPricing(8.0, 8.0, 45, 128000),
AIModel.CLAUDE_SONNET_4_5: ModelPricing(15.0, 15.0, 52, 200000),
AIModel.GEMINI_2_5_FLASH: ModelPricing(2.50, 2.50, 38, 1048576),
AIModel.DEEPSEEK_V3_2: ModelPricing(0.42, 0.42, 42, 640000),
}
@dataclass
class APIResponse:
content: str
model: AIModel
latency_ms: float
tokens_used: int
cost_usd: float
cached: bool = False
class LRUCache:
"""Simple LRU cache để tránh duplicate API calls."""
def __init__(self, max_size=1000, ttl_seconds=300):
self.cache = OrderedDict()
self.max_size = max_size
self.ttl = timedelta(seconds=ttl_seconds)
self.expiry: Dict[str, datetime] = {}
def _make_key(self, prompt: str, model: AIModel) -> str:
content = f"{model.value}:{prompt}"
return hashlib.sha256(content.encode()).hexdigest()
def get(self, prompt: str, model: AIModel) -> Optional[str]:
key = self._make_key(prompt, model)
if key in self.cache:
if datetime.now() < self.expiry[key]:
self.cache.move_to_end(key)
return self.cache[key]
else:
del self.cache[key]
del self.expiry[key]
return None
def set(self, prompt: str, model: AIModel, response: str):
key = self._make_key(prompt, model)
if len(self.cache) >= self.max_size:
oldest = next(iter(self.cache))
del self.cache[oldest]
del self.expiry[oldest]
self.cache[key] = response
self.expiry[key] = datetime.now() + self.ttl
class HolySheepAIRelay:
"""
HolySheep API Relay với smart routing, caching và failover.
base_url: https://api.holysheep.ai/v1
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.cache = LRUCache(max_size=2000, ttl_seconds=600)
self.session: Optional[aiohttp.ClientSession] = None
self.request_stats: Dict[AIModel, Dict] = {
model: {"success": 0, "failed": 0, "avg_latency": 0}
for model in AIModel
}
async def _get_session(self) -> aiohttp.ClientSession:
if self.session is None or self.session.closed:
timeout = aiohttp.ClientTimeout(total=30, connect=5)
self.session = aiohttp.ClientSession(timeout=timeout)
return self.session
def _estimate_tokens(self, text: str) -> int:
"""Rough estimation: ~4 characters per token for English."""
return len(text) // 4
def _calculate_cost(self, model: AIModel, input_tokens: int, output_tokens: int) -> float:
pricing = MODEL_PRICING[model]
input_cost = (input_tokens / 1_000_000) * pricing.input_cost
output_cost = (output_tokens / 1_000_000) * pricing.output_cost
return round(input_cost + output_cost, 6)
async def _call_model(
self,
model: AIModel,
prompt: str,
max_retries: int = 3
) -> Optional[APIResponse]:
"""Gọi một model cụ thể với retry logic."""
for attempt in range(max_retries):
try:
session = await self._get_session()
start_time = time.time()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Convert model name cho HolySheep format
model_name = model.value
payload = {
"model": model_name,
"messages": [
{"role": "user", "content": prompt}
],
"max_tokens": MODEL_PRICING[model].max_tokens // 2
}
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
latency_ms = (time.time() - start_time) * 1000
if response.status == 200:
data = await response.json()
content = data["choices"][0]["message"]["content"]
# Estimate tokens
input_tokens = self._estimate_tokens(prompt)
output_tokens = self._estimate_tokens(content)
total_tokens = input_tokens + output_tokens
cost = self._calculate_cost(model, input_tokens, output_tokens)
self.request_stats[model]["success"] += 1
self.request_stats[model]["avg_latency"] = (
self.request_stats[model]["avg_latency"] * 0.9 +
latency_ms * 0.1
)
return APIResponse(
content=content,
model=model,
latency_ms=latency_ms,
tokens_used=total_tokens,
cost_usd=cost,
cached=False
)
elif response.status == 429:
# Rate limited - exponential backoff
wait_time = (2 ** attempt) * 1.5
logger.warning(f"Rate limited, waiting {wait_time}s")
await asyncio.sleep(wait_time)
elif response.status == 401:
logger.error("Invalid API key!")
raise PermissionError("Invalid HolySheep API key")
else:
error_text = await response.text()
logger.error(f"API error {response.status}: {error_text}")
except asyncio.TimeoutError:
logger.warning(f"Timeout for {model.value}, attempt {attempt + 1}")
except aiohttp.ClientError as e:
logger.error(f"Client error: {e}")
self.request_stats[model]["failed"] += 1
return None
async def analyze_market(
self,
market_data: Dict[str, Any],
preferred_model: AIModel = AIModel.DEEPSEEK_V3_2,
enable_fallback: bool = True
) -> Optional[APIResponse]:
"""
Phân tích market data với smart model selection.
Strategy:
1. Check cache first
2. Try preferred (cheapest) model
3. Fallback to more expensive models if needed
"""
# Build prompt từ market data
prompt = self._build_analysis_prompt(market_data)
# Check cache
cached = self.cache.get(prompt, preferred_model)
if cached:
return APIResponse(
content=cached,
model=preferred_model,
latency_ms=0,
tokens_used=0,
cost_usd=0,
cached=True
)
# Model priority order (cheapest first)
models_to_try = [preferred_model]
if enable_fallback:
if preferred_model == AIModel.DEEPSEEK_V3_2:
models_to_try.extend([AIModel.GEMINI_2_5_FLASH, AIModel.CLAUDE_SONNET_4_5])
elif preferred_model == AIModel.GEMINI_2_5_FLASH:
models_to_try.extend([AIModel.CLAUDE_SONNET_4_5])
for model in models_to_try:
logger.info(f"Trying model: {model.value}")
response = await self._call_model(model, prompt)
if response:
# Cache successful response
self.cache.set(prompt, model, response.content)
return response
# Small delay between models
await asyncio.sleep(0.5)
logger.error("All models failed!")
return None
def _build_analysis_prompt(self, market_data: Dict[str, Any]) -> str:
"""Build prompt từ market data cho AI phân tích."""
symbols_info = []
for symbol, data in market_data.items():
if isinstance(data, dict):
info = f"{symbol}: Price=${data.get('price', 0):.2f}, "
info += f"Volume={data.get('volume', 0):.2f}, "
info += f"Spread={data.get('spread', 0):.4f}%"
symbols_info.append(info)
else:
symbols_info.append(f"{symbol}: {data}")
prompt = f"""Analyze the following market data and provide trading signals:
Data:
{chr(10).join(symbols_info)}
Provide a brief analysis with:
1. Market sentiment (Bullish/Bearish/Neutral)
2. Key support and resistance levels
3. Recommended action (BUY/SELL/HOLD)
4. Risk assessment
Keep response concise for fast processing."""
return prompt
async def generate_trading_signal(
self,
symbol: str,
price: float,
volume: float,
indicators: Dict[str, float]
) -> Optional[Dict[str, Any]]:
"""
Generate trading signal sử dụng AI analysis.
Trả về structured signal cho execution layer.
"""
prompt = f"""Analyze {symbol} for trading decision:
Price: ${price:.2f}
Volume: {volume:.2f}
Indicators: {json.dumps(indicators, indent=2)}
Respond ONLY with valid JSON (no markdown):
{{
"signal": "BUY|SELL|HOLD",
"confidence": 0.0-1.0,
"stop_loss": price_level,
"take_profit": price_level,
"position_size": percentage_of_capital,
"reasoning": "brief explanation"
}}"""
response = await self.analyze_market(
{symbol: {"price": price, "volume": volume, "indicators": indicators}},
preferred_model=AIModel.DEEPSEEK_V3_2
)
if response:
try:
# Parse JSON từ response
import re
json_match = re.search(r'\{[^{}]*\}', response.content, re.DOTALL)
if json_match:
return json.loads(json_match.group())
except json.JSONDecodeError:
logger.error(f"Failed to parse signal JSON: {response.content}")
return None
async def get_cost_summary(self) -> Dict[str, Any]:
"""Get cost summary từ request stats."""
total_cost = 0
total_requests = 0
summary = {"models": {}}
for model, stats in self.request_stats.items():
success = stats["success"]
failed = stats["failed"]
total = success + failed
if total > 0:
pricing = MODEL_PRICING[model]
# Estimate cost per request (avg 1000 tokens)
estimated_cost = total * (2000 / 1_000_000) * pricing.input_cost
total_cost += estimated_cost
total_requests += total
summary["models"][model.value] = {
"requests": total,
"success_rate": success / total * 100,
"avg_latency_ms": round(stats["avg_latency"], 2),
"estimated_cost_usd": round(estimated_cost, 4)
}
summary["total"] = {
"requests": total_requests,
"estimated_cost_usd": round(total_cost, 4)
}
return summary
async def close(self):
"""Cleanup connections."""
if self.session and not self.session.closed:
await self.session.close()
============ INTEGRATION EXAMPLE ============
async def example_trading_loop():
"""Example integration với market data stream."""
from dataclasses import dataclass
# Initialize HolySheep relay
api_key = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng key thực tế
ai_relay = HolySheepAIRelay(api_key)
# Simulated market data (thay bằng real data stream)
market_data = {
"BTCUSDT": {
"price": 67432.50,
"volume": 15234.67,
"spread": 0.015,
"indicators": {
"rsi": 58.3,
"macd": 125.45,
"ma_50": 65200.00,
"ma_200": 62100.00
}
}
}
try:
# Generate trading signal
signal = await ai_relay.generate_trading_signal(
symbol="BTCUSDT",
price=market_data["BTCUSDT"]["price"],
volume=market_data["BTCUSDT"]["volume"],
indicators=market_data["BTCUSDT"]["indicators"]
)
if signal:
logger.info(f"Trading Signal: {json.dumps(signal, indent=2)}")
# Get cost summary
summary = await ai_relay.get_cost_summary()
logger.info(f"Cost Summary: {json.dumps(summary, indent=2)}")
finally:
await ai_relay.close()
if __name__ == "__main__":
asyncio.run(example_trading_loop())Trading Engine Với Risk Management
Đây là phần execution engine hoàn chỉnh với position sizing, stop-loss, và portfolio balancing:
import asyncio
from decimal import Decimal
from typing import Dict, Optional
from dataclasses import dataclass
from enum import Enum
import logging
from datetime import datetime, timedelta
logger = logging.getLogger(__name__)
class OrderType(Enum):
MARKET = "market"
LIMIT = "limit"
STOP_LOSS = "stop_loss"
TAKE_PROFIT = "take_profit"
@dataclass
class Position:
symbol: str
quantity: Decimal
entry_price: Decimal
current_price: Decimal
unrealized_pnl: Decimal
timestamp: datetime
@dataclass
class TradingSignal:
symbol: str
action: str # BUY, SELL, HOLD
confidence: float
position_size: float # 0.0 - 1.0 (% of capital)
stop_loss_pct: float
take_profit_pct: float
reasoning: str
class RiskManager:
"""
Risk management system với:
- Position sizing (Kelly Criterion variant)
- Max drawdown protection
- Portfolio diversification
- Correlation-based exposure limits
"""
def __init__(
self,
max_portfolio_risk: float = 0.02, # Max 2% portfolio per trade
max_position_size: float = 0.15, # Max 15% capital per position
max_correlation_exposure: float = 0.30,
max_daily_drawdown: float = 0.05 # Stop trading if -5% daily
):
self.max_portfolio_risk = max_portfolio_risk
self.max_position_size = max_position_size
self.max_correlation_exposure = max_correlation_exposure
self.max_daily_drawdown = max_daily_drawdown
self.portfolio_value = Decimal("100000") # Initial capital
self.positions: Dict[str, Position] = {}
self.daily_pnl = Decimal("0")
self.daily_trades = 0
self.max_drawdown = Decimal("0")
def calculate_position_size(
self,
signal: TradingSignal,
current_price: Decimal
) -> Optional[Decimal]:
"""
Calculate optimal position size dựa trên Kelly Criterion.
"""
# Max amount willing to risk
risk_amount = self.portfolio_value * Decimal(str(self.max_portfolio_risk))
# Adjusted by confidence
confidence_adjusted_risk = risk_amount * Decimal(str(signal.confidence))
# Calculate position size based on stop loss distance
stop_loss_price = current_price * (1 - Decimal(str(signal.stop_loss_pct)))
risk_per_share = current_price - stop_loss_price
if risk_per_share == 0:
return None
raw_position_size = confidence_adjusted_risk / risk_per_share
# Apply max position size limit
max_size = self.portfolio_value * Decimal(str(self.max_position_size))
position_value = raw_position_size * current_price
if position_value > max_size:
raw_position_size = max_size / current_price
logger.warning(
f"Position size capped by max limit: {raw_position_size}"
)
return raw_position_size
def check_drawdown(self) -> bool:
"""Check if daily drawdown limit exceeded."""
if self.daily_pnl < -self.portfolio_value * Decimal(str(self.max_daily_drawdown)):
logger.critical(
f"Daily drawdown limit exceeded: {self.daily_pnl / self.portfolio_value * 100:.2f}%"
)
return False
return True
def get_portfolio_exposure(self) -> Dict[str, Decimal]:
"""Get current portfolio exposure breakdown."""
total_value = self.portfolio_value + sum(
p.unrealized_pnl for p in self.positions.values()
)
exposure = {}
for symbol, position in self.positions.items():
position_value = position.quantity * position.current_price
exposure[symbol] = position_value / total_value
return exposure
def update_position(
self,
symbol: str,
quantity: Decimal,
price: Decimal,
action: str
):
"""Update position after trade execution."""
if action == "BUY":
cost = quantity * price
if symbol in self.positions:
old_pos = self.positions[symbol]
new_quantity = old_pos.quantity + quantity
new_entry = (
old_pos.entry_price * old_pos.quantity + price * quantity
) / new_quantity
self.positions[symbol] = Position(
symbol=symbol,
quantity=new_quantity,
entry_price=new_entry,
current_price=price,
unrealized_pnl=Decimal("0"),
timestamp=datetime.now()
)
else:
self.positions[symbol] = Position(
symbol=symbol,
quantity=quantity,
entry_price=price,
current_price=price,
unrealized_pnl=Decimal("0"),
timestamp=datetime.now()
)
elif action == "SELL":
if symbol in self.positions:
old_pos = self.positions[symbol]
if quantity >= old_pos.quantity:
# Close entire position
pnl = (price - old_pos.entry_price) * old_pos.quantity
self.daily_pnl += pnl
self.portfolio_value += pnl
del self.positions[symbol]
logger.info(f"Closed position {symbol}, PnL: {pnl}")
else:
# Partial close
pnl = (price - old_pos.entry_price) * quantity
old_pos.quantity -= quantity
self.daily_pnl += pnl
self.portfolio_value += pnl
def update_prices(self, prices: Dict[str, Decimal]):
"""Update current prices và recalculate PnL."""
for symbol, price in prices.items():
if symbol in self.positions:
pos = self.positions[symbol]
pos.current_price = price
pos.unrealized_pnl = (price - pos.entry_price) * pos.quantity
def get_status_report(self) -> Dict:
"""Get current portfolio status."""
total_pnl = sum(p.unrealized_pnl for p in self.positions.values())
exposure = self.get_portfolio_exposure()
return {
"portfolio_value": float(self.portfolio_value),
"total_unrealized_pnl": float(total_pnl),
"daily_pnl": float(self.daily_pnl),
"open_positions": len(self.positions),
"exposure": {k: float(v) for k, v in exposure.items()},
"trading_allowed": self.check_drawdown()
}
class TradingEngine:
"""
Main trading engine orchestration.
Kết hợp AI signals với risk management và execution.
"""
def __init__(
self,
api_key: str,
risk_manager: RiskManager
):
from your_module import HolySheepAIRelay
self.ai_relay = HolySheepAIRelay(api_key)
self.risk_manager = risk_manager
self.is_running = False
self.trade_history = []
async def process_signal(self, signal: TradingSignal, current_price: Decimal):
"""Process a trading signal với full risk checks."""
logger.info(f"Processing signal: {signal.symbol} - {signal.action}")
if signal.action == "HOLD":
logger.info(f"Holding {signal.symbol}, confidence too low")
return
# Calculate position size
position_size = self.risk_manager.calculate_position_size(
signal, current_price
)
if position_size is None or position_size == 0:
logger.warning("Position size calculated as 0, skipping")
return
# Execute order (mock - replace với real broker API)
order = await self._execute_order(
symbol=signal.symbol,
action=signal.action,
quantity=position_size,
order_type=OrderType.MARKET
)
# Update risk manager
self.risk_manager.update_position(
symbol=signal.symbol,
quantity=position_size,
price=Decimal(str(order["execution_price"])),
action=signal.action
)
# Log trade
self.trade_history.append({
"timestamp": datetime.now(),
"symbol": signal.symbol,
"action": signal.action,
"quantity": float(position_size),
"price": order["execution_price"],
"signal_confidence": signal.confidence,
"reasoning": signal.reasoning
})
logger.info(f"Trade executed: {order}")
async def _execute_order(
self,
symbol: str,
action: str,
quantity: Decimal,
order_type: OrderType
) -> Dict:
"""
Mock order execution.
Thay thế bằng real broker API integration:
- Binance: python-binance
- Interactive Brokers: ib_insync
- Alpaca: alpaca-trade-api
"""
# Simulate execution
execution_price = 67432.50 # Mock price
return {
"order_id": f"ORD-{datetime.now().timestamp()}",
"symbol": symbol,
"action": action,
"quantity": float(quantity),
"execution_price": execution_price,
"status": "filled",
"timestamp": datetime.now().isoformat()
}
async def run(self, market_stream):
"""Main trading loop."""
self.is_running = True
logger.info("Trading engine started")
while self.is_running:
try:
# Check risk limits
if not self.risk_manager.check_drawdown():
logger.critical("Risk limit exceeded, stopping trading")
break
# Get latest market data
market_data = market_stream.data_buffer
if market_data:
# Update risk manager prices
prices = {
symbol: Decimal(str(data.price))
for symbol, data in market_data.items()
}
self.risk_manager.update_prices(prices)
# Generate AI signal
for symbol, data in market_data.items():
signal = await self.ai_relay.generate_trading_signal(
symbol=symbol,
price=data.price,
volume=data.volume,
indicators={"rsi": 58.3} # Add real indicators
)
if signal:
await self.process_signal(
TradingSignal(
symbol=symbol,
action=signal.get("signal", "HOLD"),
confidence=signal.get("confidence", 0),
position_size=signal.get("position_size", 0.1),
stop_loss_pct=0.02,
take_profit_pct=0.05,
reasoning=signal.get("reasoning