#!/usr/bin/env python3 # ═══════════════════════════════════════════════════════════════ # bot.py — Motor Principal do Trading Bot BTC/USDT # Uso: python bot.py # Para rodar em background: nohup python bot.py & # ═══════════════════════════════════════════════════════════════ import ccxt import pandas as pd import numpy as np import time import logging import csv import os import signal import sys from datetime import datetime from typing import Optional import config as cfg from strategy import AnaliseTecnica, Sinal from risk_manager import RiskManager from notifier import Notifier # ── LOGGING ─────────────────────────────────────────────────── logging.basicConfig( level = getattr(logging, cfg.LOG_LEVEL), format = "%(asctime)s [%(name)s] %(levelname)s: %(message)s", datefmt = "%d/%m/%Y %H:%M:%S", handlers = [ logging.FileHandler(cfg.LOG_FILE, encoding="utf-8"), logging.StreamHandler(sys.stdout), ] ) logger = logging.getLogger("BotBTC") # ── SEPARADORES VISUAIS ─────────────────────────────────────── SEP = "─" * 60 SEP2 = "═" * 60 class BTCBot: """Bot de trading automático para BTC/USDT na Binance.""" def __init__(self): self.modo = "PAPER" if cfg.MODO_PAPER else "REAL" self.risk = RiskManager() self.notifier = Notifier() self.exchange = self._conectar_exchange() self.executando = True self.ciclos = 0 self._ultimo_relatorio = datetime.now().hour # Trap de saída limpa signal.signal(signal.SIGINT, self._encerrar) signal.signal(signal.SIGTERM, self._encerrar) logger.info(SEP2) logger.info(f" 🤖 BOT BTC/USDT — MODO {self.modo}") logger.info(f" 📊 Par: {cfg.SYMBOL} | TF: {cfg.TIMEFRAME}") logger.info(f" 💼 Capital: ${cfg.CAPITAL_TOTAL:,.2f} USDT") logger.info(f" ⚙️ Risco/trade: {cfg.RISCO_POR_TRADE*100:.1f}%") logger.info(SEP2) self.notifier.inicio(self.modo) def _conectar_exchange(self) -> ccxt.Exchange: params = { "apiKey": cfg.API_KEY, "secret": cfg.API_SECRET, "enableRateLimit": True, "options": {"defaultType": "spot"}, } ex = getattr(ccxt, cfg.EXCHANGE)(params) if cfg.TESTNET and hasattr(ex, "set_sandbox_mode"): ex.set_sandbox_mode(True) logger.info("🧪 Modo TESTNET ativado") return ex def buscar_candles(self) -> Optional[pd.DataFrame]: """Busca candles OHLCV da exchange.""" try: ohlcv = self.exchange.fetch_ohlcv( cfg.SYMBOL, cfg.TIMEFRAME, limit=cfg.CANDLES_LIMIT ) df = pd.DataFrame( ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"] ) df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms") df = df.set_index("timestamp") return df except ccxt.NetworkError as e: logger.error(f"❌ Erro de rede: {e}") except ccxt.ExchangeError as e: logger.error(f"❌ Erro da exchange: {e}") except Exception as e: logger.error(f"❌ Erro inesperado: {e}") return None def buscar_saldo(self) -> Optional[dict]: """Busca saldo real da conta.""" try: balance = self.exchange.fetch_balance() return { "USDT": balance["USDT"]["free"], "BTC": balance["BTC"]["free"], } except Exception as e: logger.error(f"❌ Erro ao buscar saldo: {e}") return None def executar_compra(self, sinal: Sinal) -> bool: """Executa ordem de compra.""" pode, motivo = self.risk.pode_operar() if not pode: logger.info(f"⏭️ Trade bloqueado: {motivo}") return False preco = sinal.preco qtd = self.risk.calcular_tamanho(preco, sinal.stop) if qtd <= 0: logger.warning("⚠️ Quantidade calculada inválida") return False logger.info(SEP) logger.info(f" 🟢 SINAL DE COMPRA — Score: {sinal.score} | {sinal.motivo}") logger.info(f" 💰 Preço: ${preco:,.2f} | Qtd: {qtd:.5f} BTC") logger.info(f" 🛑 Stop: ${sinal.stop:,.2f} | A1: ${sinal.alvo1:,.2f} | A2: ${sinal.alvo2:,.2f}") logger.info(f" ⚖️ R:R = {sinal.rr}") logger.info(SEP) if self.modo == "REAL": try: order = self.exchange.create_order( symbol = cfg.SYMBOL, type = "market", side = "buy", amount = qtd, ) preco_real = order.get("average") or order.get("price") or preco qtd_real = order.get("filled") or qtd logger.info(f"✅ Ordem executada — ID: {order['id']} @ ${preco_real:,.2f}") preco = preco_real qtd = qtd_real except Exception as e: logger.error(f"❌ Falha na ordem de compra: {e}") self.notifier.erro(f"Falha na compra: {e}") return False else: logger.info("📝 [PAPER] Compra simulada") self.risk.registrar_entrada(preco, qtd, sinal.stop, sinal.alvo1, sinal.alvo2) self.notifier.compra(preco, qtd, sinal.stop, sinal.alvo1, sinal.alvo2, sinal.score) self._salvar_trade("COMPRA", preco, qtd, 0, sinal.score, sinal.motivo) return True def executar_venda(self, preco: float, motivo: str) -> bool: """Executa ordem de venda.""" if not self.risk.posicao_aberta: return False qtd = self.risk.entrada_qtd logger.info(SEP) logger.info(f" 🔴 SAÍDA [{motivo}] @ ${preco:,.2f}") logger.info(SEP) if self.modo == "REAL": try: order = self.exchange.create_order( symbol = cfg.SYMBOL, type = "market", side = "sell", amount = qtd, ) preco_real = order.get("average") or order.get("price") or preco qtd_real = order.get("filled") or qtd preco = preco_real qtd = qtd_real except Exception as e: logger.error(f"❌ Falha na ordem de venda: {e}") self.notifier.erro(f"Falha na venda: {e}") return False else: logger.info("📝 [PAPER] Venda simulada") pnl, pnl_pct = self.risk.registrar_saida(preco, qtd, motivo) self.notifier.venda(preco, qtd, motivo, pnl, pnl_pct, self.risk.capital) self._salvar_trade("VENDA", preco, qtd, pnl, 0, motivo) return True def _monitorar_posicao(self, preco_atual: float): """Verifica stop/targets da posição aberta.""" gatilho = self.risk.atualizar_trailing(preco_atual) if gatilho == "STOP": self.executar_venda(preco_atual, "STOP LOSS") elif gatilho == "ALVO1": self.notifier.alvo1(preco_atual, self.risk.stop_atual) elif gatilho == "ALVO2": self.executar_venda(preco_atual, "ALVO 2 ✨") def _salvar_trade(self, tipo: str, preco: float, qtd: float, pnl: float, score: int, motivo: str): if not cfg.SALVAR_TRADES: return arquivo = "trades.csv" novo = not os.path.exists(arquivo) with open(arquivo, "a", newline="", encoding="utf-8") as f: w = csv.writer(f) if novo: w.writerow(["data_hora", "tipo", "preco", "quantidade", "pnl", "score", "motivo", "capital", "modo"]) w.writerow([ datetime.now().strftime("%d/%m/%Y %H:%M:%S"), tipo, round(preco, 2), round(qtd, 6), round(pnl, 2), score, motivo, round(self.risk.capital, 2), self.modo, ]) def _relatorio_periodico(self): """Envia relatório diário via Telegram às 23h.""" hora = datetime.now().hour if hora == 23 and self._ultimo_relatorio != hora: self._ultimo_relatorio = hora status = self.risk.status() self.notifier.relatorio_diario(status) logger.info(f"📊 Relatório enviado: {status}") def _log_ciclo(self, df: pd.DataFrame, sinal: Sinal): """Loga resumo do ciclo atual.""" u = df.iloc[-1] pos = "" if self.risk.posicao_aberta: pnl_pos = (sinal.preco - self.risk.entrada_preco) / self.risk.entrada_preco * 100 pos = f" | POS: {pnl_pos:+.2f}% | Stop: ${self.risk.stop_atual:,.0f}" logger.info( f"[#{self.ciclos:04d}] ${sinal.preco:,.2f} | " f"RSI={sinal.rsi:.1f} | MACD={sinal.macd_hist:.0f} | " f"Score={sinal.score:+d} | {sinal.acao} | " f"Tend={sinal.tendencia} | Cap=${self.risk.capital:.0f}{pos}" ) def rodar(self): """Loop principal do bot.""" logger.info("▶️ Bot iniciado. Ctrl+C para encerrar.\n") while self.executando: try: self.ciclos += 1 # 1. Busca candles df = self.buscar_candles() if df is None or len(df) < 50: logger.warning("⚠️ Dados insuficientes, aguardando...") time.sleep(cfg.LOOP_INTERVAL) continue # 2. Análise técnica analise = AnaliseTecnica(df) sinal = analise.gerar_sinal() # 3. Log self._log_ciclo(df, sinal) if sinal.alertas: for a in sinal.alertas: logger.warning(f"⚠️ {a}") # 4. Monitora posição aberta if self.risk.posicao_aberta: self._monitorar_posicao(sinal.preco) # 5. Executa sinal se adequado elif sinal.acao == "COMPRA" and sinal.rr >= 1.5: self.executar_compra(sinal) # 6. Relatório diário self._relatorio_periodico() # 7. Aguarda próximo ciclo time.sleep(cfg.LOOP_INTERVAL) except KeyboardInterrupt: break except Exception as e: logger.error(f"❌ Erro no ciclo principal: {e}", exc_info=True) self.notifier.erro(str(e)) time.sleep(cfg.LOOP_INTERVAL * 2) def _encerrar(self, *args): logger.info("\n🛑 Encerrando bot...") status = self.risk.status() logger.info( f"📊 Resumo final:\n" f" Capital: ${status['capital']:,.2f}\n" f" P&L Total: {'+'if status['pnl_total']>=0 else ''}{status['pnl_total']:.2f}\n" f" Win Rate: {status['win_rate']}%\n" f" Wins: {status['wins']} | Losses: {status['losses']}" ) self.notifier.relatorio_diario(status) self.executando = False sys.exit(0) # ── ENTRY POINT ─────────────────────────────────────────────── if __name__ == "__main__": bot = BTCBot() bot.rodar()