import re import json from datetime import datetime from typing import Dict, List, Optional, Any from dataclasses import dataclass, asdict import logging # Configurar logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class MarketData: """Estrutura para dados de mercado""" symbol: str current_price: float variation: float variation_percent: float high: float low: float volume: int timestamp: str @dataclass class TechnicalIndicators: """Estrutura para indicadores técnicos""" rsi: float rsi_status: str ema_fast: float ema_slow: float ema_trend: str bollinger_status: str bollinger_upper: float bollinger_lower: float atr: float volatility: str volatility_multiplier: float @dataclass class FibonacciAnalysis: """Estrutura para análise de Fibonacci""" swing_high: float swing_low: float current_price: float swing_difference: float retracement_levels: int extension_levels: int projection_levels: int total_levels: int confluence_zones: int harmonic_patterns: int temporal_projections: int analysis_strength: float zone: str support: float resistance: float alerts: int signal: str @dataclass class BotAnalysis: """Estrutura completa da análise do bot""" analysis_number: int timestamp: str market_data: MarketData technical_indicators: TechnicalIndicators fibonacci_analysis: FibonacciAnalysis performance_time: Optional[float] = None class VampireBotLogParser: """Parser para logs do Vampire Trading Bot""" def __init__(self): self.patterns = self._compile_patterns() def _compile_patterns(self) -> Dict[str, re.Pattern]: """Compila padrões regex para extração de dados""" return { 'analysis_header': re.compile(r'⏰ Análise #(\d+) - ([\d:]+)'), 'market_symbol': re.compile(r'📊 DADOS DE MERCADO - (\w+)'), 'current_price': re.compile(r'Preço Atual: ([\d.]+) ↗'), 'variation': re.compile(r'Variação: ([+-][\d.]+) \(([+-][\d.]+)%\)'), 'high_low': re.compile(r'Máxima: ([\d.]+)\nMínima: ([\d.]+)'), 'volume': re.compile(r'Volume: (\d+)'), 'rsi': re.compile(r'RSI \(14\): ([\d.]+) \((\w+)\)'), 'ema': re.compile(r'EMA Rápida: ([\d.]+)\nEMA Lenta: ([\d.]+)'), 'ema_trend': re.compile(r'Tendência EMA: (\w+)'), 'bollinger': re.compile(r'Bollinger: ([\w\s]+)\n\s+Superior: ([\d.]+)\n\s+Inferior: ([\d.]+)'), 'atr': re.compile(r'ATR: ([\d.]+)'), 'volatility': re.compile(r'Volatilidade: (\w+) \(([\d.]+)x\)'), 'swing_points': re.compile(r'📊 Swing Points - Alta: ([\d,]+\.\d+), Baixa: ([\d,]+\.\d+), Atual: ([\d,]+\.\d+)'), 'swing_difference': re.compile(r'📏 Diferença Swing: ([\d,]+\.\d+) pontos'), 'retracement_levels': re.compile(r'📈 Níveis de Retracement calculados: (\d+) níveis'), 'extension_levels': re.compile(r'📊 Níveis de Extensão calculados: (\d+) níveis'), 'projection_levels': re.compile(r'🎯 Níveis de Projeção calculados: (\d+) níveis'), 'total_levels': re.compile(r'🔢 Total de níveis Fibonacci: (\d+)'), 'confluence_zones': re.compile(r'🎯 Zonas de Confluência detectadas: (\d+)'), 'harmonic_patterns': re.compile(r'🎼 Padrões Harmônicos detectados: (\d+)'), 'temporal_projections': re.compile(r'⏰ Projeções Temporais calculadas: (\d+)'), 'analysis_strength': re.compile(r'💪 Força Geral da Análise: ([\d.]+)'), 'fibonacci_conclusion': re.compile(r'🔮 ANÁLISE CONCLUÍDA - Zona: (\w+), Suporte: ([\d.]+), Resistência: ([\d.]+)'), 'performance_time': re.compile(r'Análise de mercado lenta: ([\d.]+)s'), 'fibonacci_signal': re.compile(r'🔮 Fibonacci Avançado:\s+Alertas:(\d+) FibSinal:(\w+)') } def parse_log_content(self, log_content: str) -> Optional[BotAnalysis]: """Parseia o conteúdo completo do log""" try: # Extrair cabeçalho da análise analysis_match = self.patterns['analysis_header'].search(log_content) if not analysis_match: logger.warning("Cabeçalho da análise não encontrado") return None analysis_number = int(analysis_match.group(1)) timestamp = analysis_match.group(2) # Extrair dados de mercado market_data = self._extract_market_data(log_content) if not market_data: logger.warning("Dados de mercado não encontrados") return None # Extrair indicadores técnicos technical_indicators = self._extract_technical_indicators(log_content) if not technical_indicators: logger.warning("Indicadores técnicos não encontrados") return None # Extrair análise de Fibonacci fibonacci_analysis = self._extract_fibonacci_analysis(log_content) if not fibonacci_analysis: logger.warning("Análise de Fibonacci não encontrada") return None # Extrair tempo de performance (opcional) performance_match = self.patterns['performance_time'].search(log_content) performance_time = float(performance_match.group(1)) if performance_match else None return BotAnalysis( analysis_number=analysis_number, timestamp=timestamp, market_data=market_data, technical_indicators=technical_indicators, fibonacci_analysis=fibonacci_analysis, performance_time=performance_time ) except Exception as e: logger.error(f"Erro ao parsear log: {e}") return None def _extract_market_data(self, content: str) -> Optional[MarketData]: """Extrai dados de mercado do log""" try: symbol_match = self.patterns['market_symbol'].search(content) price_match = self.patterns['current_price'].search(content) variation_match = self.patterns['variation'].search(content) high_low_match = self.patterns['high_low'].search(content) volume_match = self.patterns['volume'].search(content) if not all([symbol_match, price_match, variation_match, high_low_match, volume_match]): return None return MarketData( symbol=symbol_match.group(1), current_price=float(price_match.group(1)), variation=float(variation_match.group(1)), variation_percent=float(variation_match.group(2)), high=float(high_low_match.group(1)), low=float(high_low_match.group(2)), volume=int(volume_match.group(1)), timestamp=datetime.now().isoformat() ) except Exception as e: logger.error(f"Erro ao extrair dados de mercado: {e}") return None def _extract_technical_indicators(self, content: str) -> Optional[TechnicalIndicators]: """Extrai indicadores técnicos do log""" try: rsi_match = self.patterns['rsi'].search(content) ema_match = self.patterns['ema'].search(content) ema_trend_match = self.patterns['ema_trend'].search(content) bollinger_match = self.patterns['bollinger'].search(content) atr_match = self.patterns['atr'].search(content) volatility_match = self.patterns['volatility'].search(content) if not all([rsi_match, ema_match, ema_trend_match, bollinger_match, atr_match, volatility_match]): return None return TechnicalIndicators( rsi=float(rsi_match.group(1)), rsi_status=rsi_match.group(2), ema_fast=float(ema_match.group(1)), ema_slow=float(ema_match.group(2)), ema_trend=ema_trend_match.group(1), bollinger_status=bollinger_match.group(1).strip(), bollinger_upper=float(bollinger_match.group(2)), bollinger_lower=float(bollinger_match.group(3)), atr=float(atr_match.group(1)), volatility=volatility_match.group(1), volatility_multiplier=float(volatility_match.group(2)) ) except Exception as e: logger.error(f"Erro ao extrair indicadores técnicos: {e}") return None def _extract_fibonacci_analysis(self, content: str) -> Optional[FibonacciAnalysis]: """Extrai análise de Fibonacci do log""" try: # Buscar pelos últimos valores (análise final) swing_matches = list(self.patterns['swing_points'].finditer(content)) swing_diff_matches = list(self.patterns['swing_difference'].finditer(content)) retracement_matches = list(self.patterns['retracement_levels'].finditer(content)) extension_matches = list(self.patterns['extension_levels'].finditer(content)) projection_matches = list(self.patterns['projection_levels'].finditer(content)) total_matches = list(self.patterns['total_levels'].finditer(content)) confluence_matches = list(self.patterns['confluence_zones'].finditer(content)) harmonic_matches = list(self.patterns['harmonic_patterns'].finditer(content)) temporal_matches = list(self.patterns['temporal_projections'].finditer(content)) strength_matches = list(self.patterns['analysis_strength'].finditer(content)) conclusion_matches = list(self.patterns['fibonacci_conclusion'].finditer(content)) signal_match = self.patterns['fibonacci_signal'].search(content) if not (swing_matches and conclusion_matches and signal_match): return None # Usar os últimos valores encontrados swing_match = swing_matches[-1] conclusion_match = conclusion_matches[-1] swing_high = float(swing_match.group(1).replace(',', '')) swing_low = float(swing_match.group(2).replace(',', '')) current_price = float(swing_match.group(3).replace(',', '')) return FibonacciAnalysis( swing_high=swing_high, swing_low=swing_low, current_price=current_price, swing_difference=float(swing_diff_matches[-1].group(1).replace(',', '')) if swing_diff_matches else 0, retracement_levels=int(retracement_matches[-1].group(1)) if retracement_matches else 0, extension_levels=int(extension_matches[-1].group(1)) if extension_matches else 0, projection_levels=int(projection_matches[-1].group(1)) if projection_matches else 0, total_levels=int(total_matches[-1].group(1)) if total_matches else 0, confluence_zones=int(confluence_matches[-1].group(1)) if confluence_matches else 0, harmonic_patterns=int(harmonic_matches[-1].group(1)) if harmonic_matches else 0, temporal_projections=int(temporal_matches[-1].group(1)) if temporal_matches else 0, analysis_strength=float(strength_matches[-1].group(1)) if strength_matches else 0.0, zone=conclusion_match.group(1), support=float(conclusion_match.group(2)), resistance=float(conclusion_match.group(3)), alerts=int(signal_match.group(1)), signal=signal_match.group(2) ) except Exception as e: logger.error(f"Erro ao extrair análise de Fibonacci: {e}") return None def parse_log_file(self, file_path: str) -> Optional[BotAnalysis]: """Parseia arquivo de log""" try: with open(file_path, 'r', encoding='utf-8') as file: content = file.read() return self.parse_log_content(content) except Exception as e: logger.error(f"Erro ao ler arquivo de log: {e}") return None def to_dict(self, analysis: BotAnalysis) -> Dict[str, Any]: """Converte análise para dicionário""" return asdict(analysis) def to_json(self, analysis: BotAnalysis) -> str: """Converte análise para JSON""" return json.dumps(self.to_dict(analysis), indent=2, ensure_ascii=False) # Exemplo de uso if __name__ == "__main__": parser = VampireBotLogParser() # Exemplo com o log fornecido sample_log = """ ⏰ Análise #8 - 09:46:58 ================================================================================ 🧛 VAMPIRE TRADING BOT - ANÁLISE DETALHADA ================================================================================ 📊 DADOS DE MERCADO - WINV25 ────────────────────────────────────────────────── Preço Atual: 140135.00000 ↗ Variação: +5.00000 (+0.00%) Máxima: 140155.00000 Mínima: 140075.00000 Volume: 5023 📈 INDICADORES TÉCNICOS ────────────────────────────────────────────────── RSI (14): 46.39 (NEUTRO) EMA Rápida: 140192.30752 EMA Lenta: 140221.86717 Tendência EMA: BAIXA Bollinger: DENTRO DAS BANDAS Superior: 140672.37317 Inferior: 139913.62683 ATR: 170.73782 Volatilidade: MÉDIA (1.23x) 🔮 Fibonacci Avançado: Alertas:15 FibSinal:HOLD """ # Simular dados de Fibonacci (já que não estão completos no exemplo) full_sample = sample_log + """ 2025-08-27 09:46:58,333 - src.core.analysis.advanced_fibonacci - INFO - 🔮 ANÁLISE CONCLUÍDA - Zona: ZONA_MEDIA_ALTA, Suporte: 140133.28, Resistência: 140176.54 2025-08-27 09:46:58,218 - src.core.analysis.advanced_fibonacci - INFO - 📊 Swing Points - Alta: 140,570.00, Baixa: 139,540.00, Atual: 140,135.00 2025-08-27 09:46:58,219 - src.core.analysis.advanced_fibonacci - INFO - 📏 Diferença Swing: 1,030.00 pontos 2025-08-27 09:46:58,244 - src.core.analysis.advanced_fibonacci - INFO - 📈 Níveis de Retracement calculados: 13 níveis 2025-08-27 09:46:58,297 - src.core.analysis.advanced_fibonacci - INFO - 📊 Níveis de Extensão calculados: 11 níveis 2025-08-27 09:46:58,323 - src.core.analysis.advanced_fibonacci - INFO - 🎯 Níveis de Projeção calculados: 8 níveis 2025-08-27 09:46:58,325 - src.core.analysis.advanced_fibonacci - INFO - 🔢 Total de níveis Fibonacci: 32 2025-08-27 09:46:58,327 - src.core.analysis.advanced_fibonacci - INFO - 🎯 Zonas de Confluência detectadas: 0 2025-08-27 09:46:58,329 - src.core.analysis.advanced_fibonacci - INFO - 🎼 Padrões Harmônicos detectados: 0 2025-08-27 09:46:58,332 - src.core.analysis.advanced_fibonacci - INFO - ⏰ Projeções Temporais calculadas: 0 2025-08-27 09:46:58,332 - src.core.analysis.advanced_fibonacci - INFO - 💪 Força Geral da Análise: 0.00 """ result = parser.parse_log_content(full_sample) if result: print("✅ Log parseado com sucesso!") print(parser.to_json(result)) else: print("❌ Erro ao parsear log")