#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ SINCRONIZADOR BALANCEAMENTO -> BANCO DE DADOS Lê dados do robô (JSON e log) e sincroniza com MySQL Executar periodicamente (ex: a cada 1 hora) via cron job Uso: python sincronizar_db_balanceamento.py Cron job para rodar a cada hora: 0 * * * * cd /path/to/balanceamento && python sincronizar_db_balanceamento.py """ import json import re import os import sys import logging from datetime import datetime, timedelta from pathlib import Path import pytz try: import mysql.connector except ImportError: print("❌ Erro: mysql-connector-python não instalado") print(" Instale com: pip install mysql-connector-python") sys.exit(1) # ===================== CONFIGURAÇÃO ===================== SUBDIR = "robobalanceamento" CONFIG_FILE = f"{SUBDIR}/configuracoes_balanceamento.json" ESTADO_FILE = f"{SUBDIR}/portfolio_estado.json" LOG_FILE = f"{SUBDIR}/log_balanceamento.log" DB_HOST = "localhost" DB_USER = "root" DB_PASS = "" DB_NAME = "balanceamento_db" TIMEZONE = pytz.timezone("America/Sao_Paulo") # ===================== LOGGING ===================== logging.basicConfig( level=logging.INFO, format='[%(asctime)s] %(levelname)s: %(message)s', handlers=[ logging.FileHandler(f"{SUBDIR}/sync_database.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # ===================== FUNÇÕES DE BANCO DE DADOS ===================== def conectar_db(): """Conecta ao banco de dados MySQL""" try: conn = mysql.connector.connect( host=DB_HOST, user=DB_USER, password=DB_PASS, database=DB_NAME, charset='utf8mb4' ) return conn except mysql.connector.Error as e: logger.error(f"Erro ao conectar ao banco: {e}") return None def executar_query(conn, sql, params=None): """Executa query e retorna resultado""" try: cursor = conn.cursor(dictionary=True) if params: cursor.execute(sql, params) else: cursor.execute(sql) return cursor except mysql.connector.Error as e: logger.error(f"Erro ao executar query: {e}") return None # ===================== PARSING DE DADOS ===================== def carregar_json(filepath): """Carrega arquivo JSON""" if os.path.exists(filepath): try: with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.error(f"Erro ao carregar {filepath}: {e}") return None def extrair_ultima_secao_log(arquivo_log, separador="=========="): """Extrai última seção do log entre separadores""" if not os.path.exists(arquivo_log): return [] try: with open(arquivo_log, 'r', encoding='utf-8') as f: linhas = f.readlines() except Exception as e: logger.error(f"Erro ao ler {arquivo_log}: {e}") return [] em_secao = False secao = [] for i in range(len(linhas) - 1, -1, -1): linha = linhas[i].strip() if separador in linha: if em_secao: break em_secao = True continue if em_secao: secao.insert(0, linha) return secao def parsear_metricas(linhas_log, estado_json): """Parse das métricas da última execução""" metricas = { 'fase': 'desconhecido', 'capital_inicial': None, 'valor_total': None, 'ganho_total': None, 'ganho_percentual': None, 'moedas_ativas': 0, 'moedas_pendentes': 0, 'total_trades': 0 } # Parse do log for linha in linhas_log: linha = re.sub(r'^\[\d{2}/\d{2}/\d{4} \d{2}:\d{2}:\d{2}\]\s+', '', linha) if re.search(r'Fase:\s*construcao', linha, re.IGNORECASE): metricas['fase'] = 'construcao' elif re.search(r'Fase:\s*balanceamento', linha, re.IGNORECASE): metricas['fase'] = 'balanceamento' match = re.search(r'Capital\s*(?:Inicial)?:\s*[\$]?\s*([\d\.]+)', linha, re.IGNORECASE) if match: metricas['capital_inicial'] = float(match.group(1)) match = re.search(r'Valor.*?(?:Total.*?)?Portf[ó]lio:\s*[\$]?\s*([\d\.]+)', linha, re.IGNORECASE) if match: metricas['valor_total'] = float(match.group(1)) match = re.search(r'Ganho.*?Total:\s*[\$]?\s*([+-]?[\d\.]+)', linha, re.IGNORECASE) if match: metricas['ganho_total'] = float(match.group(1)) match = re.search(r'([+-]?\d+\.?\d*)%', linha) if match: metricas['ganho_percentual'] = float(match.group(1)) # Complementar com JSON if estado_json: metricas['moedas_ativas'] = len(estado_json.get('moedas_compradas', [])) metricas['moedas_pendentes'] = len(estado_json.get('moedas_pendentes', [])) metricas['total_trades'] = len(estado_json.get('historico_trades', [])) # Calcular ganho se não fornecido if metricas['ganho_total'] is None and metricas['capital_inicial'] and metricas['valor_total']: metricas['ganho_total'] = metricas['valor_total'] - metricas['capital_inicial'] if metricas['capital_inicial'] > 0: metricas['ganho_percentual'] = (metricas['ganho_total'] / metricas['capital_inicial']) * 100 return metricas # ===================== SINCRONIZAÇÃO ===================== def sincronizar_portfolio_snapshot(conn, metricas): """Insere snapshot do portfólio na tabela portfolio_snapshots""" try: data_hoje = datetime.now(TIMEZONE).date() sql = """ INSERT INTO portfolio_snapshots (data, fase, capital_inicial, valor_total, ganho_total, ganho_percentual, moedas_ativas, moedas_pendentes, total_trades) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE fase = VALUES(fase), valor_total = VALUES(valor_total), ganho_total = VALUES(ganho_total), ganho_percentual = VALUES(ganho_percentual), moedas_ativas = VALUES(moedas_ativas), moedas_pendentes = VALUES(moedas_pendentes), total_trades = VALUES(total_trades) """ cursor = conn.cursor() cursor.execute(sql, ( data_hoje, metricas['fase'], metricas['capital_inicial'], metricas['valor_total'], metricas['ganho_total'], metricas['ganho_percentual'], metricas['moedas_ativas'], metricas['moedas_pendentes'], metricas['total_trades'] )) conn.commit() logger.info(f"✓ Portfolio snapshot sincronizado para {data_hoje}") return True except Exception as e: logger.error(f"Erro ao sincronizar portfolio snapshot: {e}") return False def sincronizar_trades(conn, estado_json): """Insere histórico de trades na tabela rebalancing_transactions""" if not estado_json or not estado_json.get('historico_trades'): return True try: trades = estado_json['historico_trades'] cursor = conn.cursor() sql = """ INSERT IGNORE INTO rebalancing_transactions (data_hora, moeda, tipo, preco, quantidade, valor_usdt, status) VALUES (%s, %s, %s, %s, %s, %s, %s) """ for trade in trades: # Parse do timestamp try: ts = datetime.fromisoformat(trade.get('timestamp', '')[:19]) ts_br = pytz.utc.localize(ts).astimezone(TIMEZONE) except: ts_br = datetime.now(TIMEZONE) cursor.execute(sql, ( ts_br, trade.get('moeda'), trade.get('tipo', 'desconhecido'), float(trade.get('preco', 0)), float(trade.get('quantidade', 0)), float(trade.get('valor', 0)), 'executada' )) conn.commit() logger.info(f"✓ {len(trades)} trades sincronizados") return True except Exception as e: logger.error(f"Erro ao sincronizar trades: {e}") return False def sincronizar_performance_moedas(conn, estado_json, metricas): """Insere performance por moeda na tabela coin_performance""" if not estado_json: return True try: data_hoje = datetime.now(TIMEZONE).date() percentuais_alvo = estado_json.get('percentuais_alvo', {}) precos_compra = estado_json.get('precos_compra', {}) moedas_compradas = estado_json.get('moedas_compradas', []) cursor = conn.cursor() sql = """ INSERT INTO coin_performance (data, moeda, preco_medio_compra, percentual_alvo, status, timestamp) VALUES (%s, %s, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE preco_medio_compra = VALUES(preco_medio_compra), percentual_alvo = VALUES(percentual_alvo) """ for moeda, percentual_alvo in percentuais_alvo.items(): preco = precos_compra.get(moeda, 0) cursor.execute(sql, ( data_hoje, moeda, float(preco), float(percentual_alvo), 'ativa' if moeda in moedas_compradas else 'pendente' )) conn.commit() logger.info(f"✓ {len(percentuais_alvo)} moedas sincronizadas") return True except Exception as e: logger.error(f"Erro ao sincronizar performance de moedas: {e}") return False def registrar_evento(conn, tipo_evento, moeda=None, descricao=None, severidade='info', dados=None): """Registra evento na tabela portfolio_events""" try: cursor = conn.cursor() sql = """ INSERT INTO portfolio_events (data_hora, tipo_evento, moeda, descricao, severidade, dados_adicionais) VALUES (NOW(), %s, %s, %s, %s, %s) """ dados_json = json.dumps(dados) if dados else None cursor.execute(sql, ( tipo_evento, moeda, descricao, severidade, dados_json )) conn.commit() logger.info(f"✓ Evento registrado: {tipo_evento}") return True except Exception as e: logger.error(f"Erro ao registrar evento: {e}") return False # ===================== PROGRAMA PRINCIPAL ===================== def main(): logger.info("=" * 60) logger.info("Iniciando sincronização com banco de dados") logger.info("=" * 60) # Validar arquivos if not os.path.exists(CONFIG_FILE): logger.error(f"❌ Arquivo não encontrado: {CONFIG_FILE}") return False # Conectar ao banco conn = conectar_db() if not conn: logger.error("❌ Não foi possível conectar ao banco de dados") return False try: # Carregar dados config = carregar_json(CONFIG_FILE) estado = carregar_json(ESTADO_FILE) linhas_log = extrair_ultima_secao_log(LOG_FILE) if not linhas_log: logger.warning("⚠ Nenhuma linha de log encontrada") # Parse de métricas metricas = parsear_metricas(linhas_log, estado) logger.info(f"Fase: {metricas['fase']}, Valor: ${metricas['valor_total']}") # Sincronizar dados sincronizar_portfolio_snapshot(conn, metricas) sincronizar_trades(conn, estado) sincronizar_performance_moedas(conn, estado, metricas) # Registrar evento de sincronização registrar_evento( conn, 'sincronizacao_banco', descricao=f"Sincronização bem-sucedida: {metricas['moedas_ativas']} moedas ativas", severidade='info' ) logger.info("✅ Sincronização concluída com sucesso!") return True except Exception as e: logger.error(f"❌ Erro durante sincronização: {e}") registrar_evento( conn, 'erro_sincronizacao', descricao=str(e), severidade='critico' ) return False finally: conn.close() if __name__ == "__main__": sucesso = main() sys.exit(0 if sucesso else 1)