""" Fossati AI Bot - Executor de Ordens (Binance Futures) Market orders + OCO (SL + TP) + Trailing Stop """ import logging import math from typing import Optional from binance.client import Client from binance.exceptions import BinanceAPIException from ai_engine import TradeSignal, Direction log = logging.getLogger(__name__) class BinanceExecutor: def __init__(self, config): self.cfg = config self.client = Client(config.BINANCE_API_KEY, config.BINANCE_API_SECRET) self._exchange_info = {} self._load_exchange_info() def _load_exchange_info(self): try: info = self.client.futures_exchange_info() for sym in info["symbols"]: self._exchange_info[sym["symbol"]] = sym log.info(f"Exchange info carregado: {len(self._exchange_info)} símbolos") except Exception as e: log.error(f"Erro ao carregar exchange info: {e}") def execute_signal(self, signal: TradeSignal, sizing: dict) -> Optional[dict]: """Executa o sinal: define alavancagem, entra no trade, coloca SL e TP.""" symbol = signal.symbol try: # 1. Define alavancagem self._set_leverage(symbol, signal.leverage) # 2. Define margem isolada self.client.futures_change_margin_type(symbol=symbol, marginType="ISOLATED") # 3. Calcula quantidade com precisão correta quantity = self._adjust_quantity(symbol, sizing["quantity"]) if quantity <= 0: log.error(f"{symbol}: quantidade inválida após ajuste") return None # 4. Entra na posição (MARKET) side = "BUY" if signal.direction == Direction.BULLISH else "SELL" entry_order = self.client.futures_create_order( symbol=symbol, side=side, type="MARKET", quantity=quantity, ) log.info(f"✅ Entrada executada: {symbol} {side} {quantity} @ mercado") # 5. Coloca Stop Loss sl_order = self._place_stop_loss(signal, quantity) # 6. Coloca Take Profits (1, 2 e 3 parciais) tp_orders = self._place_take_profits(signal, quantity) # 7. Trailing Stop (opcional) if self.cfg.TRAILING_STOP: self._place_trailing_stop(signal, quantity) return { "entry_order": entry_order, "sl_order": sl_order, "tp_orders": tp_orders, "symbol": symbol, "direction": signal.direction.value, "quantity": quantity, "entry_price": signal.entry, "stop_loss": signal.stop_loss, "take_profit": signal.take_profit, "leverage": signal.leverage, } except BinanceAPIException as e: log.error(f"Binance API error ao executar {symbol}: {e.status_code} - {e.message}") return None except Exception as e: log.error(f"Erro ao executar {symbol}: {e}") return None def _set_leverage(self, symbol: str, leverage: int): try: self.client.futures_change_leverage(symbol=symbol, leverage=leverage) log.info(f"{symbol}: alavancagem definida em {leverage}x") except BinanceAPIException as e: log.warning(f"{symbol}: não foi possível definir alavancagem {leverage}x — {e.message}") def _place_stop_loss(self, signal: TradeSignal, quantity: float) -> Optional[dict]: symbol = signal.symbol sl_price = self._adjust_price(symbol, signal.stop_loss) try: if signal.direction == Direction.BULLISH: order = self.client.futures_create_order( symbol=symbol, side="SELL", type="STOP_MARKET", stopPrice=sl_price, quantity=quantity, reduceOnly=True, timeInForce="GTE_GTC", ) else: order = self.client.futures_create_order( symbol=symbol, side="BUY", type="STOP_MARKET", stopPrice=sl_price, quantity=quantity, reduceOnly=True, timeInForce="GTE_GTC", ) log.info(f"{symbol}: SL colocado em {sl_price}") return order except BinanceAPIException as e: log.error(f"{symbol}: erro ao colocar SL: {e.message}") return None def _place_take_profits(self, signal: TradeSignal, quantity: float) -> list: orders = [] symbol = signal.symbol targets = [ (signal.take_profit, 0.50), # TP1: 50% da posição (signal.take_profit_2, 0.30), # TP2: 30% da posição (signal.take_profit_3, 0.20), # TP3: 20% da posição ] for tp_price, pct in targets: if tp_price is None: continue tp_qty = self._adjust_quantity(symbol, quantity * pct) tp_price = self._adjust_price(symbol, tp_price) try: if signal.direction == Direction.BULLISH: order = self.client.futures_create_order( symbol=symbol, side="SELL", type="TAKE_PROFIT_MARKET", stopPrice=tp_price, quantity=tp_qty, reduceOnly=True, timeInForce="GTE_GTC", ) else: order = self.client.futures_create_order( symbol=symbol, side="BUY", type="TAKE_PROFIT_MARKET", stopPrice=tp_price, quantity=tp_qty, reduceOnly=True, timeInForce="GTE_GTC", ) orders.append(order) log.info(f"{symbol}: TP colocado em {tp_price} ({pct*100:.0f}% da posição)") except BinanceAPIException as e: log.error(f"{symbol}: erro ao colocar TP {tp_price}: {e.message}") return orders def _place_trailing_stop(self, signal: TradeSignal, quantity: float): symbol = signal.symbol callback_rate = self.cfg.TRAILING_CALLBACK try: if signal.direction == Direction.BULLISH: self.client.futures_create_order( symbol=symbol, side="SELL", type="TRAILING_STOP_MARKET", callbackRate=callback_rate, quantity=quantity, reduceOnly=True, activationPrice=signal.take_profit, ) else: self.client.futures_create_order( symbol=symbol, side="BUY", type="TRAILING_STOP_MARKET", callbackRate=callback_rate, quantity=quantity, reduceOnly=True, activationPrice=signal.take_profit, ) log.info(f"{symbol}: Trailing Stop de {callback_rate}% ativado em TP1") except BinanceAPIException as e: log.warning(f"{symbol}: não foi possível colocar trailing stop — {e.message}") def cancel_all_orders(self, symbol: str): try: self.client.futures_cancel_all_open_orders(symbol=symbol) log.info(f"{symbol}: todas as ordens abertas canceladas") except Exception as e: log.error(f"{symbol}: erro ao cancelar ordens: {e}") def get_open_positions(self) -> list: try: positions = self.client.futures_position_information() return [p for p in positions if float(p["positionAmt"]) != 0] except Exception as e: log.error(f"Erro ao buscar posições: {e}") return [] def get_account_balance(self) -> float: try: info = self.client.futures_account() for asset in info["assets"]: if asset["asset"] == "USDT": return float(asset["walletBalance"]) return 0.0 except Exception as e: log.error(f"Erro ao buscar saldo: {e}") return 0.0 def get_candles(self, symbol: str, interval: str, limit: int = 200) -> list: try: return self.client.futures_klines(symbol=symbol, interval=interval, limit=limit) except BinanceAPIException as e: log.error(f"Erro ao buscar candles {symbol} {interval}: {e.message}") return [] def _adjust_quantity(self, symbol: str, quantity: float) -> float: info = self._exchange_info.get(symbol, {}) for f in info.get("filters", []): if f["filterType"] == "LOT_SIZE": step = float(f["stepSize"]) precision = len(str(step).rstrip("0").split(".")[-1]) if "." in str(step) else 0 quantity = math.floor(quantity / step) * step return round(quantity, precision) return round(quantity, 3) def _adjust_price(self, symbol: str, price: float) -> float: info = self._exchange_info.get(symbol, {}) for f in info.get("filters", []): if f["filterType"] == "PRICE_FILTER": tick = float(f["tickSize"]) precision = len(str(tick).rstrip("0").split(".")[-1]) if "." in str(tick) else 0 price = round(round(price / tick) * tick, precision) return price return round(price, 4)