#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ REXMULTITRADE - Robô de Trading Híbrido Combina múltiplas moedas (MULTITRADE) com análise de padrões de compra/venda (REX) """ import json import logging import logging.handlers import time import os import sys import hashlib import hmac import urllib.request import urllib.error import urllib.parse import socket from datetime import datetime, timedelta from collections import deque import statistics # Configurar logging rotativo (20MB max, 3 backups - como REX) log_file = os.path.join(os.path.dirname(__file__), 'logs', 'log_trades.log') os.makedirs(os.path.dirname(log_file), exist_ok=True) handler = logging.handlers.RotatingFileHandler( log_file, maxBytes=20*1024*1024, backupCount=3 ) formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) handler.setFormatter(formatter) logger = logging.getLogger('rexmultitrade') logger.setLevel(logging.INFO) logger.addHandler(handler) # Também log no console console = logging.StreamHandler(sys.stdout) console.setFormatter(formatter) logger.addHandler(console) class BinanceAPI: """Wrapper da API Binance com sincronização de timestamp""" def __init__(self, api_key, api_secret, testnet=False): self.api_key = api_key self.api_secret = api_secret self.testnet = testnet if testnet: self.base_url = "https://testnet.binance.vision/api/v3" else: self.base_url = "https://api.binance.com/api/v3" self.server_time_offset = 0 def _fazer_requisicao(self, metodo, endpoint, params=None, assinado=False): """Faz requisição HTTP para Binance""" try: url = f"{self.base_url}{endpoint}" headers = {"X-MBX-APIKEY": self.api_key} if assinado and params: # Adicionar timestamp sincronizado params['timestamp'] = int((time.time() + self.server_time_offset) * 1000) # Gerar assinatura query_string = urllib.parse.urlencode(params) signature = hmac.new( self.api_secret.encode(), query_string.encode(), hashlib.sha256 ).hexdigest() url += f"?{query_string}&signature={signature}" elif params: url += "?" + urllib.parse.urlencode(params) req = urllib.request.Request(url, headers=headers, method=metodo) with urllib.request.urlopen(req, timeout=10) as response: return json.loads(response.read().decode()) except (urllib.error.URLError, socket.timeout) as e: logger.error(f"Erro na requisição Binance: {e}") return None def sincronizar_timestamp(self): """Sincroniza timestamp com servidor Binance (resolve erro -1021)""" try: response = self._fazer_requisicao('GET', '/time') if response and 'serverTime' in response: server_time_ms = response['serverTime'] local_time_ms = int(time.time() * 1000) self.server_time_offset = (server_time_ms - local_time_ms) / 1000 logger.info(f"Timestamp sincronizado. Offset: {self.server_time_offset:.3f}s") return True except Exception as e: logger.error(f"Erro ao sincronizar timestamp: {e}") return False def obter_candles(self, symbol, interval, limit=500): """Obtém candles históricos""" params = { 'symbol': symbol, 'interval': interval, 'limit': limit } return self._fazer_requisicao('GET', '/klines', params) def obter_ticker(self, symbol): """Obtém ticker atual""" params = {'symbol': symbol} return self._fazer_requisicao('GET', '/ticker/24hr', params) def obter_saldo(self, asset): """Obtém saldo de um ativo""" try: response = self._fazer_requisicao( 'GET', '/account', assinado=True ) if response and 'balances' in response: for balance in response['balances']: if balance['asset'] == asset: return float(balance['free']) return 0 except Exception as e: logger.error(f"Erro ao obter saldo: {e}") return 0 def place_market_order(self, symbol, side, quantity, max_retries=3): """Coloca order market""" for attempt in range(max_retries): try: params = { 'symbol': symbol, 'side': side, 'type': 'MARKET', 'quantity': quantity } response = self._fazer_requisicao( 'POST', '/order', params, assinado=True ) if response and 'orderId' in response: logger.info(f"Order colocada: {symbol} {side} {quantity}") return response else: logger.warning(f"Resposta inválida da order: {response}") except Exception as e: logger.error(f"Erro ao colocar order (tentativa {attempt+1}): {e}") if attempt < max_retries - 1: time.sleep(1) return None class AnalisadorMercado: """Analisa mercado com indicadores e padrões horários""" def __init__(self, binance_api, config): self.binance = binance_api self.config = config self.historico_30dias = {} # Cache de 30 dias por moeda def calcular_rsi(self, closes, period=14): """Calcula RSI usando ganhos e perdas médias""" if len(closes) < period + 1: return 50 ganhos = [] perdas = [] for i in range(1, len(closes)): mudanca = closes[i] - closes[i-1] if mudanca > 0: ganhos.append(mudanca) perdas.append(0) else: ganhos.append(0) perdas.append(abs(mudanca)) ganho_medio = statistics.mean(ganhos[-period:]) perda_media = statistics.mean(perdas[-period:]) if perda_media == 0: return 100 if ganho_medio > 0 else 50 rs = ganho_medio / perda_media rsi = 100 - (100 / (1 + rs)) return rsi def calcular_atr(self, highs, lows, closes, period=14): """Calcula Average True Range""" if len(highs) < period: return 0 trs = [] for i in range(1, len(highs)): tr = max( highs[i] - lows[i], abs(highs[i] - closes[i-1]), abs(lows[i] - closes[i-1]) ) trs.append(tr) if not trs: return 0 return statistics.mean(trs[-period:]) def obter_candles_preparados(self, symbol, interval='3m', limit=500): """Obtém candles e retorna com closes, highs, lows""" candles = self.binance.obter_candles(symbol, interval, limit) if not candles: return None closes = [float(c[4]) for c in candles] highs = [float(c[2]) for c in candles] lows = [float(c[3]) for c in candles] volumes = [float(c[7]) for c in candles] timestamps = [int(c[0]) / 1000 for c in candles] return { 'closes': closes, 'highs': highs, 'lows': lows, 'volumes': volumes, 'timestamps': timestamps, 'candles': candles } def calcular_triple_rsi(self, closes): """Calcula RSI em 3 períodos (5, 14, 21) - estratégia do MULTITRADE""" rsi_5 = self.calcular_rsi(closes, 5) rsi_14 = self.calcular_rsi(closes, 14) rsi_21 = self.calcular_rsi(closes, 21) rsi_media = (rsi_14 + rsi_21) / 2 return { 'rsi_5': rsi_5, 'rsi_14': rsi_14, 'rsi_21': rsi_21, 'rsi_media': rsi_media } def verificar_volume_alto(self, volumes): """Verifica se volume está acima da MA20""" if len(volumes) < 20: return True volume_ma20 = statistics.mean(volumes[-20:]) volume_atual = volumes[-1] return volume_atual >= volume_ma20 * 0.8 def obter_melhores_precos_30dias(self, symbol): """ Analisa últimos 30 dias em 6 períodos de 4 horas Retorna melhor hora de compra e venda (REX pattern matching) """ # Obter 240 candles de 1h = 10 dias * 24h # Para 30 dias, obteremos 720 candles de 1h candles = self.binance.obter_candles(symbol, '1h', 720) if not candles or len(candles) < 100: logger.warning(f"Histórico insuficiente para {symbol}") return None # Agrupar em períodos de 4 horas (6 períodos por dia) periodos = {} # 0-4h, 4-8h, 8-12h, 12-16h, 16-20h, 20-24h for candle in candles: timestamp = int(candle[0]) / 1000 dt = datetime.fromtimestamp(timestamp) hora = dt.hour periodo_idx = hora // 4 # 0-5 if periodo_idx not in periodos: periodos[periodo_idx] = { 'closes': [], 'lows': [], 'highs': [], 'count': 0 } periodos[periodo_idx]['closes'].append(float(candle[4])) periodos[periodo_idx]['lows'].append(float(candle[3])) periodos[periodo_idx]['highs'].append(float(candle[2])) periodos[periodo_idx]['count'] += 1 # Calcular estatísticas por período resultados = {} for idx in range(6): if idx not in periodos: continue p = periodos[idx] if len(p['lows']) == 0: continue resultados[idx] = { 'min_low': min(p['lows']), 'max_high': max(p['highs']), 'avg_close': statistics.mean(p['closes']), 'confianca': min(100, (p['count'] / (len(candles) / 6)) * 100), 'label_horas': f"{idx*4:02d}h-{(idx+1)*4:02d}h" } # Encontrar melhor hora de compra (menor preço médio) if resultados: melhor_compra = min(resultados.items(), key=lambda x: x[1]['min_low']) melhor_venda = max(resultados.items(), key=lambda x: x[1]['max_high']) return { 'melhor_compra': { 'periodo': melhor_compra[0], 'label': melhor_compra[1]['label_horas'], 'preco_min': melhor_compra[1]['min_low'], 'confianca': melhor_compra[1]['confianca'] }, 'melhor_venda': { 'periodo': melhor_venda[0], 'label': melhor_venda[1]['label_horas'], 'preco_max': melhor_venda[1]['max_high'], 'confianca': melhor_venda[1]['confianca'] } } return None def esta_no_melhor_horario_compra(self, analise_30dias): """Verifica se está no melhor horário de compra""" if not analise_30dias: return True hora_atual = datetime.now().hour periodo_atual = hora_atual // 4 periodo_ideal = analise_30dias['melhor_compra']['periodo'] # Permitir 1 período antes ou depois return abs(periodo_atual - periodo_ideal) <= 1 def esta_no_melhor_horario_venda(self, analise_30dias): """Verifica se está no melhor horário de venda""" if not analise_30dias: return True hora_atual = datetime.now().hour periodo_atual = hora_atual // 4 periodo_ideal = analise_30dias['melhor_venda']['periodo'] # Permitir 1 período antes ou depois return abs(periodo_atual - periodo_ideal) <= 1 def calcular_ganho_medio_diario(self, symbol): """ Calcula o percentual médio de ganho DIÁRIO da moeda nos últimos 30 dias Usa candles diários (1d) Retorna: percentual médio de ganho para usar como Take Profit dinâmico """ try: # Obter 30 candles diários (30 dias) candles = self.binance.obter_candles(symbol, '1d', 30) if not candles or len(candles) < 10: logger.warning(f"Histórico insuficiente para cálculo de ganho diário: {symbol}") return 0.5 # Padrão: 0.5% se não tiver dados ganhos_diarios = [] # Calcular ganho diário (close - open) / open * 100 for candle in candles: open_price = float(candle[1]) close_price = float(candle[4]) if open_price > 0: ganho_pct = ((close_price - open_price) / open_price) * 100 ganhos_diarios.append(ganho_pct) if not ganhos_diarios: return 0.5 # Calcular média dos ganhos ganho_medio = statistics.mean(ganhos_diarios) # Usar valor absoluto (mesmo que dia tenha caído, usamos magnitude) ganho_medio_abs = abs(ganho_medio) # Garantir mínimo de 0.3% e máximo de 3% (para não ser irreal) ganho_dinamico = max(0.3, min(ganho_medio_abs, 3.0)) logger.info(f"{symbol}: Ganho médio diário últimos 30 dias: {ganho_medio:.2f}% → Take Profit: {ganho_dinamico:.2f}%") return ganho_dinamico except Exception as e: logger.error(f"Erro ao calcular ganho diário de {symbol}: {e}") return 0.5 # Padrão class EstrategiaTrading: """Estratégia de trading (Triple RSI + Padrões Horários)""" def __init__(self, analisador, config): self.analisador = analisador self.config = config def dever_comprar(self, symbol, dados_candles, analise_padroes): """ Verifica se deve comprar: 1. Triple RSI baixo 2. Confirmações (volume, momentum) 3. Está no melhor horário de compra """ rsi = self.analisador.calcular_triple_rsi(dados_candles['closes']) rsi_min = self.config['indicadores']['rsi_minimo_compra'] # Condição 1: Triple RSI baixo if not (rsi['rsi_5'] <= rsi_min and rsi['rsi_14'] <= rsi_min and rsi['rsi_media'] <= rsi_min): return False, "RSI não está baixo" # Condição 2: Confirmação por volume if not self.analisador.verificar_volume_alto(dados_candles['volumes']): return False, "Volume baixo" # Condição 3: Melhor horário de compra if not self.analisador.esta_no_melhor_horario_compra(analise_padroes): return False, "Fora do melhor horário de compra" logger.info(f"{symbol}: SINAL DE COMPRA - RSI:{rsi['rsi_5']:.1f}/{rsi['rsi_14']:.1f}/{rsi['rsi_21']:.1f}") return True, "Todas as condições atendidas" def dever_vender(self, symbol, dados_candles, analise_padroes, preco_entrada, preco_atual, ganho_medio_diario=0.5, trailing_stop_ativo=False, trailing_stop_preco=0): """ Verifica se deve vender: 1. Triple RSI alto OU atingiu ganho médio dinâmico 2. Está no melhor horário de venda 3. Ou ativou trailing stop ganho_medio_diario: percentual de ganho médio diário histórico (use como Take Profit dinâmico) """ rsi = self.analisador.calcular_triple_rsi(dados_candles['closes']) rsi_max = self.config['indicadores']['rsi_maximo_venda'] # Calcular ganho atual ganho_pct = ((preco_atual - preco_entrada) / preco_entrada) * 100 # Verificar trailing stop if trailing_stop_ativo and preco_atual <= trailing_stop_preco: logger.info(f"{symbol}: VENDA - Trailing Stop ativado (Ganho: {ganho_pct:.2f}%)") return True, "Trailing stop ativado" # Condição 1: Atingiu Take Profit dinâmico (ganho médio diário)? if ganho_pct >= ganho_medio_diario: # Verificar se está no melhor horário de venda if self.analisador.esta_no_melhor_horario_venda(analise_padroes): logger.info(f"{symbol}: VENDA - Atingiu Take Profit dinâmico ({ganho_pct:.2f}% >= {ganho_medio_diario:.2f}%)") return True, f"Take Profit atingido ({ganho_pct:.2f}%)" # Condição 2: Triple RSI alto? if rsi['rsi_5'] >= rsi_max and rsi['rsi_14'] >= rsi_max: # Verificar se está no melhor horário de venda if self.analisador.esta_no_melhor_horario_venda(analise_padroes): logger.info(f"{symbol}: VENDA - RSI alto (Ganho: {ganho_pct:.2f}%)") return True, "RSI alto no melhor horário" return False, f"Aguardando: Ganho {ganho_pct:.2f}% (meta: {ganho_medio_diario:.2f}%)" def calcular_trailing_stop(self, preco_atual, preco_max_atingido): """Calcula trailing stop dinâmico""" trailing_inicial = self.config['trailing_stop_inicial'] trailing_posterior = self.config['trailing_stop_posterior'] # Se recém entrou, usar trailing inicial (28%) if preco_max_atingido == preco_atual: return preco_atual * (1 - trailing_inicial) # Caso contrário, usar trailing posterior (0.07%) return preco_max_atingido * (1 - trailing_posterior) class RoboTrader: """Gerenciador principal do robô""" def __init__(self, config_file): self.config_file = config_file self.config = self._carregar_config() self.estado_file = os.path.join(os.path.dirname(config_file), 'estado_robos.json') self.estado = self._carregar_estado() # Determinar ambiente (testnet ou produção) ambiente = self.config.get('ambiente', 'produção').lower() testnet = ambiente == 'testnet' if testnet: logger.info("🧪 MODO TESTNET - Nenhuma moeda real será tradada!") else: logger.warning("⚠️ MODO PRODUÇÃO - Dinheiro real será tradado!") # ✓ NOVO: Selecionar credenciais corretas baseado no ambiente (padrão REX) if testnet: api_key = self.config.get('binance_api_key_testnet', '').strip() api_secret = self.config.get('binance_secret_key_testnet', '').strip() if not api_key or not api_secret or 'TESTNET' in api_key.upper(): logger.error("❌ Configure suas credenciais Testnet!") logger.error(" Edite 'binance_api_key_testnet' e 'binance_secret_key_testnet' em configuracoes.json") logger.error(" Obtenha em: https://testnet.binance.vision (Dashboard -> API Management)") sys.exit(1) else: api_key = self.config.get('binance_api_key', '').strip() api_secret = self.config.get('binance_secret_key', '').strip() if not api_key or not api_secret or 'COLE' in api_key.upper(): logger.error("❌ Configure suas credenciais Binance Produção!") logger.error(" Edite 'binance_api_key' e 'binance_secret_key' em configuracoes.json") logger.error(" Obtenha em: https://www.binance.com/pt/account/api-management") sys.exit(1) # Inicializar APIs com credenciais corretas self.binance = BinanceAPI(api_key, api_secret, testnet=testnet) # Sincronizar timestamp if not self.binance.sincronizar_timestamp(): logger.error("Falha ao sincronizar timestamp com Binance") sys.exit(1) # Inicializar analisador e estratégia self.analisador = AnalisadorMercado(self.binance, self.config) self.estrategia = EstrategiaTrading(self.analisador, self.config) # Cache de análise de 30 dias self.cache_analise_30dias = {} # Cache de ganho médio diário (atualiza 1x por dia) self.cache_ganho_medio_diario = {} self.timestamp_cache = 0 self.timestamp_cache_ganho = 0 def _carregar_config(self): """Carrega configurações do JSON""" try: with open(self.config_file, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.error(f"Erro ao carregar configuração: {e}") sys.exit(1) def _carregar_estado(self): """Carrega estado dos robôs""" try: if os.path.exists(self.estado_file): with open(self.estado_file, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.error(f"Erro ao carregar estado: {e}") return {'robos': {}} def _salvar_estado(self): """Salva estado dos robôs""" try: self.estado['timestamp_sincronizacao'] = int(time.time()) with open(self.estado_file, 'w', encoding='utf-8') as f: json.dump(self.estado, f, indent=2) except Exception as e: logger.error(f"Erro ao salvar estado: {e}") def registrar_trade(self, symbol, side, quantity, price, tipo_trade): """Registra trade em SQLite via HTTP API""" try: payload = { 'symbol': symbol, 'side': side, 'quantity': quantity, 'price': price, 'tipo': tipo_trade, 'timestamp': int(time.time()) } # Chamar api.php local api_url = 'http://localhost/projetos/rexmultitrade/api.php' data = urllib.parse.urlencode(payload).encode('utf-8') req = urllib.request.Request(api_url, data=data, method='POST') with urllib.request.urlopen(req, timeout=5) as response: result = json.loads(response.read().decode()) if result.get('sucesso'): logger.info(f"Trade registrado em BD: {symbol} {side} {quantity} @ {price}") else: logger.warning(f"Falha ao registrar trade: {result.get('erro')}") except Exception as e: logger.warning(f"Não foi possível registrar trade em BD: {e}") def processar_moeda(self, moeda_config): """Processa uma moeda individual""" symbol = moeda_config['symbol'] par = moeda_config['par'] if not moeda_config.get('ativo', True): return logger.debug(f"Processando {par}...") # Obter candles dados = self.analisador.obter_candles_preparados(symbol, '3m') if not dados: logger.warning(f"Não conseguiu obter candles para {symbol}") return # Análise de 30 dias (cache com atualização a cada 1 hora) agora = time.time() if (symbol not in self.cache_analise_30dias or agora - self.timestamp_cache > 3600): # 1 hora self.cache_analise_30dias[symbol] = self.analisador.obter_melhores_precos_30dias(symbol) self.timestamp_cache = agora analise_30dias = self.cache_analise_30dias.get(symbol) # Calcular ganho médio diário (cache 1x por dia = 86400 segundos) if (symbol not in self.cache_ganho_medio_diario or agora - self.timestamp_cache_ganho > 86400): # 1 dia self.cache_ganho_medio_diario[symbol] = self.analisador.calcular_ganho_medio_diario(symbol) self.timestamp_cache_ganho = agora ganho_medio_diario = self.cache_ganho_medio_diario.get(symbol, 0.5) # Inicializar estado da moeda se não existir if symbol not in self.estado['robos']: self.estado['robos'][symbol] = { 'capital_inicial': moeda_config['capital'], 'capital_atual': moeda_config['capital'], 'quantidade': 0, 'preco_entrada': 0, 'trades_realizados': 0, 'ganho_total': 0, 'ganho_percentual': 0, 'em_posicao': False, 'last_compra_timestamp': 0, 'last_venda_timestamp': 0, 'trailing_stop_ativo': False, 'trailing_stop_preco': 0, 'ganho_medio_diario': ganho_medio_diario } estado_moeda = self.estado['robos'][symbol] # Atualizar ganho médio no estado estado_moeda['ganho_medio_diario'] = ganho_medio_diario preco_atual = dados['closes'][-1] # COMPRA if not estado_moeda['em_posicao']: deve_comprar, razao = self.estrategia.dever_comprar( par, dados, analise_30dias ) if deve_comprar: # Calcular quantidade com 98% do capital (agressividade) capital_use = estado_moeda['capital_atual'] * moeda_config['agressividade'] quantidade = capital_use / preco_atual # Verificar quantidade mínima if quantidade * preco_atual < moeda_config['capital_min']: logger.warning(f"{par}: Capital insuficiente para compra") return # Executar compra order = self.binance.place_market_order(symbol, 'BUY', quantidade) if order: estado_moeda['quantidade'] = quantidade estado_moeda['preco_entrada'] = preco_atual estado_moeda['em_posicao'] = True estado_moeda['last_compra_timestamp'] = int(time.time()) estado_moeda['trailing_stop_ativo'] = False estado_moeda['trailing_stop_preco'] = 0 self.registrar_trade(symbol, 'BUY', quantidade, preco_atual, 'compra_padrao') logger.info(f"{par}: COMPRA EXECUTADA - {quantidade:.8f} @ {preco_atual}") # VENDA else: # Atualizar trailing stop if preco_atual > estado_moeda['preco_entrada']: trailing_stop = self.estrategia.calcular_trailing_stop( preco_atual, estado_moeda['trailing_stop_preco'] or estado_moeda['preco_entrada'] ) estado_moeda['trailing_stop_preco'] = trailing_stop estado_moeda['trailing_stop_ativo'] = True deve_vender, razao = self.estrategia.dever_vender( par, dados, analise_30dias, estado_moeda['preco_entrada'], preco_atual, ganho_medio_diario, estado_moeda['trailing_stop_ativo'], estado_moeda['trailing_stop_preco'] ) if deve_vender: # Executar venda order = self.binance.place_market_order(symbol, 'SELL', estado_moeda['quantidade']) if order: # Calcular ganho ganho_valor = (preco_atual - estado_moeda['preco_entrada']) * estado_moeda['quantidade'] ganho_pct = ((preco_atual - estado_moeda['preco_entrada']) / estado_moeda['preco_entrada']) * 100 # Atualizar estado estado_moeda['capital_atual'] += ganho_valor estado_moeda['ganho_total'] += ganho_valor estado_moeda['ganho_percentual'] = ((estado_moeda['capital_atual'] - estado_moeda['capital_inicial']) / estado_moeda['capital_inicial']) * 100 estado_moeda['trades_realizados'] += 1 estado_moeda['em_posicao'] = False estado_moeda['last_venda_timestamp'] = int(time.time()) estado_moeda['trailing_stop_ativo'] = False self.registrar_trade(symbol, 'SELL', estado_moeda['quantidade'], preco_atual, 'venda_padrao') logger.info(f"{par}: VENDA EXECUTADA - {estado_moeda['quantidade']:.8f} @ {preco_atual} | Ganho: {ganho_pct:.2f}%") logger.info(f"{par}: Capital: {estado_moeda['capital_atual']:.2f} USDT | Ganho Total: {estado_moeda['ganho_percentual']:.2f}%") def executar_ciclo(self): """Executa um ciclo de trading para todas as moedas""" logger.debug("=" * 80) logger.debug(f"Iniciando ciclo de trading: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") for moeda in self.config['moedas']: try: self.processar_moeda(moeda) except Exception as e: logger.error(f"Erro ao processar {moeda['par']}: {e}", exc_info=True) self._salvar_estado() logger.debug("Ciclo concluído") def executar(self): """Loop principal do robô""" logger.info("=" * 80) logger.info("REXMULTITRADE iniciado") logger.info(f"Moedas: {', '.join([m['par'] for m in self.config['moedas']])}") logger.info(f"Timeframe: {self.config['timeframe']}") logger.info(f"Ciclo: {self.config['ciclo_segundos']}s") logger.info("=" * 80) try: while True: try: self.executar_ciclo() except Exception as e: logger.error(f"Erro no ciclo: {e}", exc_info=True) # Aguardar próximo ciclo time.sleep(self.config['ciclo_segundos']) except KeyboardInterrupt: logger.info("Robô interrompido pelo usuário") except Exception as e: logger.error(f"Erro fatal: {e}", exc_info=True) finally: self._salvar_estado() logger.info("Robô finalizado") def main(): """Função principal""" config_file = os.path.join( os.path.dirname(__file__), 'configuracoes.json' ) if not os.path.exists(config_file): logger.error(f"Arquivo de configuração não encontrado: {config_file}") sys.exit(1) # Verificar credenciais Binance with open(config_file) as f: config = json.load(f) if (config['binance_api_key'] == 'COLE_SUA_API_KEY_AQUI' or config['binance_secret_key'] == 'COLE_SUA_SECRET_KEY_AQUI'): logger.error("❌ Configure suas credenciais Binance no configuracoes.json") sys.exit(1) robo = RoboTrader(config_file) robo.executar() if __name__ == '__main__': main()