""" Fossati AI Bot — Orquestrador Principal Método Inteligente Fossati · SMC PRO Elite · IA Automatizada """ import time import logging import pandas as pd import numpy as np from datetime import datetime import pytz import ta import config from smc_analyzer import SMCAnalyzer, SMCAnalysis, Direction from ai_engine import AIEngine from risk_manager import RiskManager from executor import BinanceExecutor from notifier import TelegramNotifier import scan_log # ─── LOGGING ──────────────────────────────────────────────── import sys logging.basicConfig( level=getattr(logging, config.LOG_LEVEL), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[ logging.FileHandler(config.LOG_FILE, encoding="utf-8"), logging.StreamHandler(stream=open(sys.stdout.fileno(), mode="w", encoding="utf-8", closefd=False)), ], ) log = logging.getLogger("fossati.main") def klines_to_df(klines: list) -> pd.DataFrame: if not klines: return pd.DataFrame() df = pd.DataFrame(klines, columns=[ "timestamp", "open", "high", "low", "close", "volume", "close_time", "quote_volume", "trades", "taker_buy_base", "taker_buy_quote", "ignore", ]) for col in ["open", "high", "low", "close", "volume"]: df[col] = df[col].astype(float) df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms") return df.reset_index(drop=True) def get_rsi(df: pd.DataFrame, period: int = 14) -> float: if len(df) < period + 1: return 50.0 try: rsi = ta.momentum.RSIIndicator(df["close"], window=period).rsi() return float(rsi.iloc[-1]) except Exception: return 50.0 def get_volume_ratio(df: pd.DataFrame, period: int = 20) -> float: if len(df) < period: return 1.0 avg = df["volume"].iloc[-period-1:-1].mean() cur = df["volume"].iloc[-1] return float(cur / avg) if avg > 0 else 1.0 def is_trading_hour() -> bool: if not config.TRADING_HOURS_ENABLED: return True tz = pytz.timezone(config.TIMEZONE) now = datetime.now(tz) return config.TRADING_START_HOUR <= now.hour < config.TRADING_END_HOUR def scan_symbol( symbol: str, executor: BinanceExecutor, analyzer: SMCAnalyzer, ai_engine: AIEngine, risk_mgr: RiskManager, notifier: TelegramNotifier, open_positions: list, ) -> bool: """ Escaneia um símbolo, analisa e executa se houver sinal. Retorna True se uma ordem foi executada. """ log.debug(f"Escaneando {symbol}...") # ── Coleta de dados em 3 timeframes klines_15m = executor.get_candles(symbol, "15m", config.CANDLE_LIMIT) klines_1h = executor.get_candles(symbol, "1h", config.CANDLE_LIMIT) klines_4h = executor.get_candles(symbol, "4h", config.CANDLE_LIMIT) df_15m = klines_to_df(klines_15m) df_1h = klines_to_df(klines_1h) df_4h = klines_to_df(klines_4h) if df_15m.empty: log.warning(f"{symbol}: sem dados 15m") scan_log.record(symbol, status="no_data", reason="sem candles 15m") return False # ── Análise SMC analysis_15m = analyzer.analyze(df_15m, symbol, "15m") analysis_1h = analyzer.analyze(df_1h, symbol, "1h") if not df_1h.empty else None analysis_4h = analyzer.analyze(df_4h, symbol, "4h") if not df_4h.empty else None if not analysis_15m: scan_log.record(symbol, status="no_data", reason="análise 15m falhou") return False # ── Indicadores auxiliares rsi_15m = get_rsi(df_15m) vol_ratio = get_volume_ratio(df_15m) # ── Determina direção do sinal trend_15 = analysis_15m.structure.trend trend_1h = analysis_1h.structure.trend if analysis_1h else Direction.NEUTRAL trend_4h = analysis_4h.structure.trend if analysis_4h else Direction.NEUTRAL # HTF concordante htf_trend = trend_4h if trend_4h != Direction.NEUTRAL else trend_1h # Define direção proposta signal_direction = None # Bullish: tendência de alta + OB de compra + preço em desconto if (trend_15 == Direction.BULLISH and analysis_15m.nearest_bullish_ob and analysis_15m.in_discount and rsi_15m < 60): signal_direction = Direction.BULLISH # Bearish: tendência de baixa + OB de venda + preço em prêmio elif (trend_15 == Direction.BEARISH and analysis_15m.nearest_bearish_ob and analysis_15m.in_premium and rsi_15m > 40): signal_direction = Direction.BEARISH if not signal_direction: # Diagnóstico: indica QUAL gate falhou e calcula score informativo # da direção mais plausível para alimentar o painel probe_dir = (Direction.BULLISH if trend_15 == Direction.BULLISH else Direction.BEARISH if trend_15 == Direction.BEARISH else (Direction.BULLISH if htf_trend == Direction.BULLISH else Direction.BEARISH if htf_trend == Direction.BEARISH else None)) probe_score = None if probe_dir is not None: try: probe_score, _ = analyzer.score_setup( analysis_15m, probe_dir, volume_ratio=vol_ratio, rsi=rsi_15m, htf_trend=htf_trend, ) except Exception as e: log.debug(f"{symbol}: probe score falhou: {e}") # Detalha qual gate barrou a direção missing = [] if trend_15 == Direction.NEUTRAL: missing.append("trend 15m neutro") if trend_15 == Direction.BULLISH: if not analysis_15m.nearest_bullish_ob: missing.append("sem OB bull") if not analysis_15m.in_discount: missing.append("fora do desconto") if rsi_15m >= 60: missing.append(f"RSI {rsi_15m:.1f}≥60") elif trend_15 == Direction.BEARISH: if not analysis_15m.nearest_bearish_ob: missing.append("sem OB bear") if not analysis_15m.in_premium: missing.append("fora do prêmio") if rsi_15m <= 40: missing.append(f"RSI {rsi_15m:.1f}≤40") reason = "falta: " + ", ".join(missing) if missing else "sem direção" log.debug(f"{symbol}: sem sinal — {reason} (probe {probe_score}/12)") scan_log.record( symbol, status="no_signal", direction=probe_dir.value if probe_dir else None, score=probe_score, price=analysis_15m.current_price, rsi=rsi_15m, volume_ratio=vol_ratio, trend=trend_15.value, htf_trend=htf_trend.value, in_discount=analysis_15m.in_discount, in_premium=analysis_15m.in_premium, reason=reason, ) return False # ── Score de confluências (checklist Fossati) score, confluences = analyzer.score_setup( analysis_15m, signal_direction, volume_ratio=vol_ratio, rsi=rsi_15m, htf_trend=htf_trend, ) log.info(f"{symbol}: score {score}/12 ({signal_direction.value})") if score < config.MIN_CONFLUENCE_SCORE: scan_log.record( symbol, status="low_score", direction=signal_direction.value, score=score, price=analysis_15m.current_price, rsi=rsi_15m, volume_ratio=vol_ratio, trend=trend_15.value, htf_trend=htf_trend.value, in_discount=analysis_15m.in_discount, in_premium=analysis_15m.in_premium, reason=f"score {score}/12 < mín {config.MIN_CONFLUENCE_SCORE}", ) return False # Score atende o mínimo scan_log.record( symbol, status="ok", direction=signal_direction.value, score=score, price=analysis_15m.current_price, rsi=rsi_15m, volume_ratio=vol_ratio, trend=trend_15.value, htf_trend=htf_trend.value, in_discount=analysis_15m.in_discount, in_premium=analysis_15m.in_premium, reason=f"score {score}/12 ≥ mín, avaliando entrada", ) # ── Verificação de risco can, reason = risk_mgr.can_trade(open_positions=len(open_positions)) if not can: log.info(f"Trade bloqueado pelo risk manager: {reason}") return False # ── Geração do sinal pela IA signal = ai_engine.generate_signal( analysis_15m=analysis_15m, analysis_1h=analysis_1h, analysis_4h=analysis_4h, score=score, confluences=confluences, signal_direction=signal_direction, volume_ratio=vol_ratio, rsi_15m=rsi_15m, current_balance=risk_mgr.stats.current_balance, ) if not signal: return False # ── Sizing sizing = risk_mgr.calculate_position_size( entry=signal.entry, stop_loss=signal.stop_loss, leverage=signal.leverage, signal_direction=signal.direction.value, ) if "error" in sizing: log.error(f"{symbol}: erro no sizing: {sizing['error']}") return False # ── Execução log.info( f"🚀 EXECUTANDO: {symbol} {signal.direction.value.upper()} | " f"E:{signal.entry:.4f} SL:{signal.stop_loss:.4f} TP:{signal.take_profit:.4f} | " f"{signal.leverage}x | Score:{score}/12" ) result = executor.execute_signal(signal, sizing) if not result: return False # ── Registra e notifica risk_mgr.register_trade_open(symbol, signal) notifier.send_signal(signal, sizing) return True def monitor_positions( executor: BinanceExecutor, risk_mgr: RiskManager, notifier: TelegramNotifier, known_positions: dict, ) -> dict: """Monitora posições abertas e registra fechamentos.""" current = {p["symbol"]: p for p in executor.get_open_positions()} # Detecta posições que fecharam for sym, pos in known_positions.items(): if sym not in current: entry_price = float(pos.get("entryPrice", 0)) unreal_pnl = float(pos.get("unRealizedProfit", 0)) direction = "bullish" if float(pos.get("positionAmt", 0)) > 0 else "bearish" current_p = float(pos.get("markPrice", entry_price)) # PnL estimado (a Binance liquida e retorna o real) pnl = unreal_pnl risk_mgr.register_trade_close(sym, pnl) notifier.send_trade_closed( symbol=sym, direction=direction, pnl=pnl, entry=entry_price, exit_price=current_p, ) # Verifica meta if risk_mgr.stats.target_reached: notifier.send_target_reached( balance=risk_mgr.stats.current_balance, pnl=risk_mgr.stats.realized_pnl, pct=risk_mgr.stats.daily_target_pct * 100, ) return current def main(): log.info("=" * 60) log.info(" FOSSATI AI TRADING BOT — INICIANDO") log.info(" Método Inteligente Fossati · SMC PRO Elite") log.info("=" * 60) # Inicializa módulos analyzer = SMCAnalyzer(config) ai_engine = AIEngine(config) risk_mgr = RiskManager(config) executor = BinanceExecutor(config) notifier = TelegramNotifier(config) # Saldo real da Binance real_balance = executor.get_account_balance() if real_balance > 0: # Se o saldo real difere do arquivo (depósito/saque), reseta a base do dia if abs(real_balance - risk_mgr.stats.starting_balance) > 0.01: log.info( f"Sincronizando com Binance: arquivo=${risk_mgr.stats.starting_balance:.2f} " f"→ real=${real_balance:.2f} (resetando base do dia)" ) risk_mgr.stats.starting_balance = real_balance risk_mgr.stats.daily_target = real_balance * risk_mgr.stats.daily_target_pct risk_mgr.stats.max_drawdown_today = 0.0 if risk_mgr.stats.trading_halted and "rawdown" in risk_mgr.stats.halt_reason: risk_mgr.stats.trading_halted = False risk_mgr.stats.halt_reason = "" risk_mgr.stats.current_balance = real_balance risk_mgr._save() log.info(f"Saldo real da Binance: ${real_balance:.2f} | Meta: ${risk_mgr.stats.daily_target:.2f}") # Notifica início notifier.send_startup( balance=risk_mgr.stats.current_balance, daily_target=risk_mgr.stats.daily_target, symbols=config.SYMBOLS, ) known_positions = {} last_summary = datetime.now() log.info(f"Monitorando {len(config.SYMBOLS)} tokens a cada {config.SCAN_INTERVAL_SECONDS}s") while True: try: now = datetime.now() # ── Resumo diário a cada hora if (now - last_summary).seconds >= 3600: notifier.send_daily_summary(risk_mgr) last_summary = now # ── Monitora posições abertas (detecta fechamentos) known_positions = monitor_positions(executor, risk_mgr, notifier, known_positions) open_positions = list(known_positions.values()) # ── Sincroniza saldo real da Binance (detecta depósitos/saques) try: live_balance = executor.get_account_balance() if live_balance > 0 and abs(live_balance - risk_mgr.stats.current_balance) > 0.01: log.info(f"Saldo Binance atualizado: ${risk_mgr.stats.current_balance:.2f} → ${live_balance:.2f}") risk_mgr.stats.current_balance = live_balance # Se o saldo aumentou (depósito) e estava em halt por drawdown, reseta if risk_mgr.stats.trading_halted and "rawdown" in risk_mgr.stats.halt_reason: risk_mgr.stats.starting_balance = live_balance risk_mgr.stats.daily_target = live_balance * risk_mgr.stats.daily_target_pct risk_mgr.stats.trading_halted = False risk_mgr.stats.halt_reason = "" risk_mgr.stats.max_drawdown_today = 0.0 log.info(f"Trading retomado · novo saldo base ${live_balance:.2f} · meta ${risk_mgr.stats.daily_target:.2f}") risk_mgr._save() except Exception as e: log.debug(f"Falha ao sincronizar saldo: {e}") # ── Verifica se pode operar can_trade, reason = risk_mgr.can_trade(len(open_positions)) if not can_trade: log.info(f"[Aguardando] {reason}") time.sleep(config.SCAN_INTERVAL_SECONDS) continue # ── Verifica horário de trading if not is_trading_hour(): log.info("Fora do horário de operação. Aguardando...") time.sleep(300) continue # ── Varre tokens em busca de setups scan_log.start_cycle() for symbol in config.SYMBOLS: # Não abre nova posição no mesmo símbolo if symbol in known_positions: continue executed = scan_symbol( symbol=symbol, executor=executor, analyzer=analyzer, ai_engine=ai_engine, risk_mgr=risk_mgr, notifier=notifier, open_positions=open_positions, ) if executed: # Recarrega posições abertas após nova entrada known_positions = {p["symbol"]: p for p in executor.get_open_positions()} open_positions = list(known_positions.values()) # Pequena pausa entre símbolos para não saturar a API time.sleep(1) log.info(f"Ciclo completo. Posições abertas: {len(open_positions)}. " f"Aguardando {config.SCAN_INTERVAL_SECONDS}s...") time.sleep(config.SCAN_INTERVAL_SECONDS) except KeyboardInterrupt: log.info("Bot interrompido pelo usuário.") notifier.send_daily_summary(risk_mgr) break except Exception as e: log.error(f"Erro no loop principal: {e}", exc_info=True) notifier.send_error(str(e)) time.sleep(30) if __name__ == "__main__": main()