""" Fossati AI Bot - Motor de Decisão por IA (Claude API) Analisa o contexto SMC e gera sinal de trade com confiança """ import json import logging import anthropic from typing import Optional, Tuple from dataclasses import dataclass from smc_analyzer import SMCAnalysis, Direction log = logging.getLogger(__name__) @dataclass class TradeSignal: symbol: str direction: Direction entry: float stop_loss: float take_profit: float take_profit_2: Optional[float] # TP2 para parcial take_profit_3: Optional[float] # TP3 agressivo leverage: int confidence: float # 0-1 score: int # 0-12 confluências confluences: list reasoning: str risk_reward: float position_size_pct: float # % do saldo a alocar class AIEngine: def __init__(self, config): self.cfg = config self.client = anthropic.Anthropic(api_key=config.ANTHROPIC_API_KEY) def generate_signal( self, analysis_15m: SMCAnalysis, analysis_1h: Optional[SMCAnalysis], analysis_4h: Optional[SMCAnalysis], score: int, confluences: list, signal_direction: Direction, volume_ratio: float, rsi_15m: float, current_balance: float, ) -> Optional[TradeSignal]: if score < self.cfg.MIN_CONFLUENCE_SCORE: log.info(f"{analysis_15m.symbol}: score {score}/12 insuficiente (mín {self.cfg.MIN_CONFLUENCE_SCORE})") return None context = self._build_context( analysis_15m, analysis_1h, analysis_4h, score, confluences, signal_direction, volume_ratio, rsi_15m, current_balance ) try: response = self.client.messages.create( model=self.cfg.AI_MODEL, max_tokens=self.cfg.AI_MAX_TOKENS, system=self._system_prompt(), messages=[{"role": "user", "content": context}] ) raw = response.content[0].text signal_data = self._parse_response(raw) if not signal_data: return None # Valida e constrói o TradeSignal entry = signal_data.get("entry", analysis_15m.current_price) sl = signal_data.get("stop_loss") tp1 = signal_data.get("take_profit_1") tp2 = signal_data.get("take_profit_2") tp3 = signal_data.get("take_profit_3") if not sl or not tp1: log.warning(f"{analysis_15m.symbol}: IA não retornou SL ou TP válidos") return None # Calcula R:R if signal_direction == Direction.BULLISH: rr = (tp1 - entry) / (entry - sl) if entry > sl else 0 else: rr = (entry - tp1) / (sl - entry) if sl > entry else 0 if rr < self.cfg.MIN_RR_RATIO: log.info(f"{analysis_15m.symbol}: R:R {rr:.2f} abaixo do mínimo {self.cfg.MIN_RR_RATIO}") return None # Alavancagem baseada no score leverage = self.cfg.LEVERAGE_MAP.get(score, self.cfg.DEFAULT_LEVERAGE) leverage = min(leverage, self.cfg.MAX_LEVERAGE) # Tamanho da posição (% do saldo pelo risco definido) risk_pct = self.cfg.CONSERVATIVE_RISK if self.cfg.CONSERVATIVE_MODE else self.cfg.MAX_RISK_PER_TRADE confidence = signal_data.get("confidence", 0.7) return TradeSignal( symbol=analysis_15m.symbol, direction=signal_direction, entry=float(entry), stop_loss=float(sl), take_profit=float(tp1), take_profit_2=float(tp2) if tp2 else None, take_profit_3=float(tp3) if tp3 else None, leverage=leverage, confidence=float(confidence), score=score, confluences=confluences, reasoning=signal_data.get("reasoning", ""), risk_reward=round(rr, 2), position_size_pct=risk_pct, ) except Exception as e: log.error(f"Erro na IA para {analysis_15m.symbol}: {e}") return None def _system_prompt(self) -> str: return """Você é o motor de decisão do Fossati AI Trading Bot, especializado em Smart Money Concepts (SMC) e ICT Methodology aplicados a futuros perpétuos de criptomoedas na Binance. Seu papel é analisar o contexto de mercado fornecido e gerar um sinal de trade preciso no formato JSON. Metodologia aplicada: - Método Inteligente Fossati SMC PRO Elite - Entrada baseada em Order Blocks (OB) de qualidade - Stop Loss: abaixo/acima do OB com margem de segurança de 0.1% - Take Profit 1: próxima zona de liquidez (R:R mínimo 2:1) - Take Profit 2: próximo OB oposto ou nível de estrutura (R:R 3-4:1) - Take Profit 3: nível de liquidez HTF (R:R 5:1+) - Alavancagem proporcional ao score de confluências SEMPRE responda APENAS com JSON válido no formato especificado, sem texto adicional.""" def _build_context( self, a15: SMCAnalysis, a1h: Optional[SMCAnalysis], a4h: Optional[SMCAnalysis], score: int, confluences: list, direction: Direction, volume_ratio: float, rsi: float, balance: float, ) -> str: ob = a15.nearest_bullish_ob if direction == Direction.BULLISH else a15.nearest_bearish_ob context_data = { "symbol": a15.symbol, "sinal_proposto": direction.value, "preco_atual": a15.current_price, "saldo_atual_usdt": round(balance, 2), "score_confluencias": f"{score}/12", "confluencias": confluences, "timeframe_15m": { "tendencia": a15.structure.trend.value, "fase": a15.structure.current_phase, "bos": a15.structure.last_bos, "choch": a15.structure.last_choch, "in_discount": a15.in_discount, "in_premium": a15.in_premium, "pd_level": round(a15.premium_discount_level, 4), "ob_mais_proximo": { "high": round(ob.high, 4), "low": round(ob.low, 4), "forca": round(ob.strength, 2), } if ob else None, "fvgs_bull": len(a15.bullish_fvg), "fvgs_bear": len(a15.bearish_fvg), "liquidez_niveis": [ {"tipo": l.level_type, "preco": round(l.price, 4)} for l in a15.liquidity_levels[:5] ], "rsi_14": round(rsi, 1), "volume_ratio": round(volume_ratio, 2), }, "timeframe_1h": { "tendencia": a1h.structure.trend.value, "fase": a1h.structure.current_phase, } if a1h else None, "timeframe_4h": { "tendencia": a4h.structure.trend.value, "fase": a4h.structure.current_phase, } if a4h else None, } return f"""Analise o setup abaixo e gere o sinal de trade: {json.dumps(context_data, indent=2, ensure_ascii=False)} Responda APENAS com JSON no formato: {{ "aprovar": true ou false, "motivo_rejeicao": "string ou null", "entry": 0.0, "stop_loss": 0.0, "take_profit_1": 0.0, "take_profit_2": 0.0, "take_profit_3": 0.0, "confidence": 0.0, "reasoning": "string curta explicando a entrada" }} Regras: - Se score < 9/12, aprovar = false - SL deve estar do outro lado do OB com 0.1% de margem - TP1: R:R mínimo 2:1 - TP2: R:R mínimo 3:1 - TP3: R:R mínimo 5:1 - confidence: 0.0 a 1.0 proporcional ao score e alinhamento de tendências""" def _parse_response(self, raw: str) -> Optional[dict]: try: raw = raw.strip() if raw.startswith("```"): lines = raw.split("\n") raw = "\n".join(lines[1:-1] if lines[-1] == "```" else lines[1:]) data = json.loads(raw) if not data.get("aprovar", False): log.info(f"IA rejeitou o setup: {data.get('motivo_rejeicao', 'sem motivo')}") return None return data except json.JSONDecodeError as e: log.error(f"Erro ao parsear resposta da IA: {e}\nRaw: {raw[:200]}") return None