""" Fossati AI Bot - Gestão de Risco e Capital Meta diária, drawdown máximo, sizing de posição, compounding """ import json import logging from datetime import datetime, date from typing import Optional from pathlib import Path from dataclasses import dataclass, asdict log = logging.getLogger(__name__) @dataclass class DailyStats: date: str starting_balance: float current_balance: float daily_target: float # USDT absoluto da meta daily_target_pct: float # % da meta realized_pnl: float unrealized_pnl: float total_trades: int winning_trades: int losing_trades: int max_drawdown_today: float target_reached: bool trading_halted: bool halt_reason: str class RiskManager: def __init__(self, config): self.cfg = config self.stats_file = Path("daily_stats.json") self.trades_file = Path(config.TRADE_LOG_FILE) self._load_or_init() def _load_or_init(self): today = str(date.today()) if self.stats_file.exists(): with open(self.stats_file) as f: data = json.load(f) if data.get("date") == today: self.stats = DailyStats(**data) log.info(f"Stats do dia carregados: saldo={self.stats.current_balance:.2f}") return # Novo dia: aplica compounding se habilitado prev_balance = self._get_previous_balance() balance = prev_balance if self.cfg.COMPOUNDING and prev_balance else self.cfg.INITIAL_BALANCE target_pct = (self.cfg.CONSERVATIVE_TARGET if self.cfg.CONSERVATIVE_MODE else self.cfg.DAILY_PROFIT_TARGET) target_usdt = balance * target_pct self.stats = DailyStats( date=today, starting_balance=balance, current_balance=balance, daily_target=target_usdt, daily_target_pct=target_pct, realized_pnl=0.0, unrealized_pnl=0.0, total_trades=0, winning_trades=0, losing_trades=0, max_drawdown_today=0.0, target_reached=False, trading_halted=False, halt_reason="", ) self._save() log.info(f"Novo dia iniciado | Saldo: ${balance:.2f} | Meta: ${target_usdt:.2f} (+{target_pct*100:.0f}%)") def _get_previous_balance(self) -> Optional[float]: if self.stats_file.exists(): with open(self.stats_file) as f: data = json.load(f) return data.get("current_balance") return None def _save(self): with open(self.stats_file, "w") as f: json.dump(asdict(self.stats), f, indent=2) # ─── VERIFICAÇÕES ANTES DE ENTRAR ─────────────────────── def can_trade(self, open_positions: int = 0) -> tuple[bool, str]: """Verifica se o bot pode abrir uma nova operação.""" if self.stats.trading_halted: return False, f"Trading pausado: {self.stats.halt_reason}" if self.stats.target_reached: return False, "Meta diária atingida. Aguardando próximo dia." if open_positions >= self.cfg.MAX_OPEN_POSITIONS: return False, f"Máximo de {self.cfg.MAX_OPEN_POSITIONS} posições simultâneas atingido" drawdown = self._current_drawdown() if drawdown >= self.cfg.MAX_DAILY_DRAWDOWN: self._halt(f"Drawdown máximo de {self.cfg.MAX_DAILY_DRAWDOWN*100:.0f}% atingido") return False, self.stats.halt_reason return True, "OK" def calculate_position_size(self, entry: float, stop_loss: float, leverage: int, signal_direction: str) -> dict: """Calcula o tamanho da posição baseado no risco definido.""" balance = self.stats.current_balance risk_pct = (self.cfg.CONSERVATIVE_RISK if self.cfg.CONSERVATIVE_MODE else self.cfg.MAX_RISK_PER_TRADE) risk_usdt = balance * risk_pct # Distância do SL em % if signal_direction == "bullish": sl_distance_pct = abs(entry - stop_loss) / entry else: sl_distance_pct = abs(stop_loss - entry) / entry if sl_distance_pct == 0: return {"error": "SL distance zero"} # Tamanho da posição em USDT (com alavancagem) position_usdt = (risk_usdt / sl_distance_pct) position_usdt_leveraged = position_usdt / leverage # Margin necessária margin_required = position_usdt / leverage # Quantidade de contratos quantity = position_usdt / entry # Verificações min_margin = balance * 0.05 # Mínimo de 5% do saldo como margem max_margin = balance * 0.50 # Máximo de 50% do saldo como margem margin_required = max(min_margin, min(margin_required, max_margin)) position_usdt = margin_required * leverage quantity = position_usdt / entry return { "entry": entry, "stop_loss": stop_loss, "leverage": leverage, "risk_usdt": round(risk_usdt, 2), "risk_pct": round(risk_pct * 100, 1), "sl_distance_pct": round(sl_distance_pct * 100, 3), "position_size_usdt": round(position_usdt, 2), "margin_required": round(margin_required, 2), "quantity": quantity, } # ─── ATUALIZAÇÃO DE P&L ───────────────────────────────── def register_trade_open(self, symbol: str, signal): self.stats.total_trades += 1 self._save() log.info(f"Trade aberto registrado: {symbol} ({signal.direction.value})") def register_trade_close(self, symbol: str, pnl: float): self.stats.realized_pnl += pnl self.stats.current_balance += pnl if pnl > 0: self.stats.winning_trades += 1 else: self.stats.losing_trades += 1 # Atualiza drawdown dd = self._current_drawdown() if dd > self.stats.max_drawdown_today: self.stats.max_drawdown_today = dd # Verifica meta profit_pct = (self.stats.current_balance - self.stats.starting_balance) / self.stats.starting_balance if profit_pct >= self.stats.daily_target_pct: self.stats.target_reached = True log.info(f"🎯 META DIÁRIA ATINGIDA! P&L: +${self.stats.realized_pnl:.2f} (+{profit_pct*100:.1f}%)") # Verifica drawdown if dd >= self.cfg.MAX_DAILY_DRAWDOWN: self._halt(f"Drawdown {dd*100:.1f}% atingiu limite de {self.cfg.MAX_DAILY_DRAWDOWN*100:.0f}%") self._save() log.info(f"Trade fechado: {symbol} PnL={pnl:+.2f} | Saldo: ${self.stats.current_balance:.2f}") def _current_drawdown(self) -> float: if self.stats.starting_balance == 0: return 0.0 dd = (self.stats.starting_balance - self.stats.current_balance) / self.stats.starting_balance return max(0.0, dd) def _halt(self, reason: str): self.stats.trading_halted = True self.stats.halt_reason = reason self._save() log.warning(f"⛔ TRADING PAUSADO: {reason}") def get_summary(self) -> str: s = self.stats profit_pct = (s.current_balance - s.starting_balance) / s.starting_balance * 100 wr = (s.winning_trades / s.total_trades * 100) if s.total_trades > 0 else 0 remaining = max(0, s.daily_target - s.realized_pnl) return ( f"📊 RESUMO DO DIA — {s.date}\n" f"💰 Saldo: ${s.current_balance:.2f} ({profit_pct:+.1f}%)\n" f"🎯 Meta: ${s.daily_target:.2f} | Faltam: ${remaining:.2f}\n" f"📈 P&L Realizado: ${s.realized_pnl:+.2f}\n" f"🏆 Win Rate: {wr:.0f}% ({s.winning_trades}W/{s.losing_trades}L)\n" f"📉 Max DD Hoje: {s.max_drawdown_today*100:.1f}%\n" f"{'✅ META ATINGIDA' if s.target_reached else '🔄 Em andamento'}" )