"""Módulo de análise técnica de mercado para scalping.""" import re from typing import Dict, List, Optional, Any from dataclasses import dataclass from datetime import datetime import numpy as np import pandas as pd import logging from config.config import ( TechnicalAnalysisConfig, ScoringConfig, TradingConfig, RegexPatterns ) from ..utils.utils import calculate_rsi, calculate_bollinger_bands, calculate_ema, format_number from .fibonacci_analysis import AdvancedFibonacciEngine, AdvancedFibonacciAnalysis from ..core.log_parser import VampireBotLogParser, BotAnalysis from ..core.advanced_market_processing import AdvancedMarketProcessor # Configurar logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class MarketData: """Classe para representar dados de mercado.""" price: float variation: float rsi: int ema_trend: str bb_position: str volume: float def __post_init__(self): """Validação dos dados após inicialização.""" if self.price < 0: raise ValueError("Preço não pode ser negativo") if not 0 <= self.rsi <= 100: raise ValueError("RSI deve estar entre 0 e 100") if self.volume < 0: raise ValueError("Volume não pode ser negativo") @dataclass class TechnicalSignal: """Classe para representar um sinal técnico.""" indicator: str signal_type: str # 'BUY', 'SELL', 'NEUTRAL' strength: int # 0-100 description: str confidence_impact: int class MarketDataParser: """Classe responsável por extrair dados de mercado do texto.""" @staticmethod def parse_market_data(text: str) -> Optional[MarketData]: """Extrai dados de mercado do texto de entrada.""" try: # Extrair valores usando regex price_match = re.search(RegexPatterns.PRICE_PATTERN, text) variation_match = re.search(RegexPatterns.VARIATION_PATTERN, text) rsi_match = re.search(RegexPatterns.RSI_PATTERN, text) ema_match = re.search(RegexPatterns.EMA_PATTERN, text) bb_match = re.search(RegexPatterns.BB_PATTERN, text) vol_match = re.search(RegexPatterns.VOLUME_PATTERN, text) # Processar valores extraídos price = float(price_match.group(1).replace(',', '')) if price_match else 0 variation_str = variation_match.group(1) if variation_match else "0" variation = float(variation_str.replace('%', '').replace('+', '')) if variation_str != "0" else 0 rsi = int(rsi_match.group(1)) if rsi_match else 50 ema_trend = ema_match.group(1) if ema_match else "NEUTRO" bb_position = bb_match.group(1) if bb_match else "DENTRO" volume = float(vol_match.group(1)) if vol_match else 0 return MarketData( price=price, variation=variation, rsi=rsi, ema_trend=ema_trend, bb_position=bb_position, volume=volume ) except (ValueError, AttributeError) as e: print(f"Erro ao processar dados de mercado: {e}") return None class RSIAnalyzer: """Analisador de RSI.""" @staticmethod def analyze(rsi: int) -> TechnicalSignal: """Analisa o RSI e retorna sinal técnico.""" config = TechnicalAnalysisConfig() if rsi <= config.RSI_OVERSOLD: if rsi <= config.RSI_EXTREME_OVERSOLD: return TechnicalSignal( indicator="RSI", signal_type="BUY", strength=90, description=f"RSI em zona de sobrevenda extrema ({rsi}): COMPRA FORTE", confidence_impact=ScoringConfig.RSI_SCORE + 10 ) else: return TechnicalSignal( indicator="RSI", signal_type="BUY", strength=70, description=f"RSI em zona de sobrevenda ({rsi}): COMPRA", confidence_impact=ScoringConfig.RSI_SCORE ) elif rsi >= config.RSI_OVERBOUGHT: if rsi >= config.RSI_EXTREME_OVERBOUGHT: return TechnicalSignal( indicator="RSI", signal_type="SELL", strength=90, description=f"RSI em zona de sobrecompra extrema ({rsi}): VENDA FORTE", confidence_impact=ScoringConfig.RSI_SCORE + 10 ) else: return TechnicalSignal( indicator="RSI", signal_type="SELL", strength=70, description=f"RSI em zona de sobrecompra ({rsi}): VENDA", confidence_impact=ScoringConfig.RSI_SCORE ) elif config.RSI_NEUTRAL_MIN <= rsi <= config.RSI_NEUTRAL_MAX: return TechnicalSignal( indicator="RSI", signal_type="NEUTRAL", strength=30, description=f"RSI neutro ({rsi}): aguardar confirmação", confidence_impact=0 ) else: return TechnicalSignal( indicator="RSI", signal_type="NEUTRAL", strength=50, description=f"RSI em zona intermediária ({rsi})", confidence_impact=0 ) class EMAAnalyzer: """Analisador de EMA.""" @staticmethod def analyze(ema_trend: str) -> TechnicalSignal: """Analisa a tendência EMA e retorna sinal técnico.""" if ema_trend == 'ALTA': return TechnicalSignal( indicator="EMA", signal_type="BUY", strength=60, description="Tendência EMA ALTA: viés de COMPRA", confidence_impact=ScoringConfig.EMA_SCORE ) elif ema_trend == 'BAIXA': return TechnicalSignal( indicator="EMA", signal_type="SELL", strength=60, description="Tendência EMA BAIXA: viés de VENDA", confidence_impact=ScoringConfig.EMA_SCORE ) else: return TechnicalSignal( indicator="EMA", signal_type="NEUTRAL", strength=30, description="Tendência EMA neutra", confidence_impact=0 ) class BollingerBandsAnalyzer: """Analisador de Bollinger Bands.""" @staticmethod def analyze(bb_position: str) -> TechnicalSignal: """Analisa a posição nas Bollinger Bands e retorna sinal técnico.""" if bb_position == 'ABAIXO': return TechnicalSignal( indicator="BB", signal_type="BUY", strength=80, description="Preço abaixo da banda inferior: COMPRA (reversão)", confidence_impact=ScoringConfig.BB_SCORE ) elif bb_position in ['ACIMA', 'SOBRE']: return TechnicalSignal( indicator="BB", signal_type="SELL", strength=80, description="Preço acima da banda superior: VENDA (reversão)", confidence_impact=ScoringConfig.BB_SCORE ) else: # DENTRO return TechnicalSignal( indicator="BB", signal_type="NEUTRAL", strength=40, description="Preço dentro das bandas: aguardar breakout", confidence_impact=5 ) class MomentumAnalyzer: """Analisador de momentum (variação de preço).""" @staticmethod def analyze(variation: float) -> TechnicalSignal: """Analisa o momentum e retorna sinal técnico.""" config = TechnicalAnalysisConfig() if abs(variation) >= config.SIGNIFICANT_MOVEMENT_THRESHOLD: if variation > 0: return TechnicalSignal( indicator="MOMENTUM", signal_type="BUY", strength=60, description=f"Momentum positivo (+{variation:.2f}%): seguir tendência", confidence_impact=ScoringConfig.MOMENTUM_SCORE ) else: return TechnicalSignal( indicator="MOMENTUM", signal_type="SELL", strength=60, description=f"Momentum negativo ({variation:.2f}%): seguir tendência", confidence_impact=ScoringConfig.MOMENTUM_SCORE ) else: return TechnicalSignal( indicator="MOMENTUM", signal_type="NEUTRAL", strength=30, description=f"Momentum fraco ({variation:.2f}%)", confidence_impact=0 ) class VolumeAnalyzer: """Analisador de volume.""" @staticmethod def analyze(volume: float) -> TechnicalSignal: """Analisa o volume e retorna sinal técnico.""" config = TechnicalAnalysisConfig() if volume > config.VOLUME_HIGH_THRESHOLD: return TechnicalSignal( indicator="VOLUME", signal_type="NEUTRAL", strength=70, description=f"Volume alto ({volume:.1f}x): confirma movimento", confidence_impact=ScoringConfig.VOLUME_SCORE ) elif volume < config.VOLUME_LOW_THRESHOLD: return TechnicalSignal( indicator="VOLUME", signal_type="NEUTRAL", strength=20, description=f"Volume baixo ({volume:.1f}x): cuidado com falsos sinais", confidence_impact=-ScoringConfig.LOW_VOLUME_PENALTY ) else: return TechnicalSignal( indicator="VOLUME", signal_type="NEUTRAL", strength=50, description=f"Volume normal ({volume:.1f}x)", confidence_impact=0 ) class ScalpingSetupDetector: """Detector de setups específicos para scalping.""" @staticmethod def detect_perfect_setups(market_data: MarketData, signals: List[TechnicalSignal]) -> List[TechnicalSignal]: """Detecta setups perfeitos para scalping.""" special_signals = [] config = TechnicalAnalysisConfig() # Setup 1: RSI extremo + EMA contrária = reversão forte if ((market_data.rsi <= config.RSI_EXTREME_OVERSOLD and market_data.ema_trend == 'BAIXA') or (market_data.rsi >= config.RSI_EXTREME_OVERBOUGHT and market_data.ema_trend == 'ALTA')): special_signals.append(TechnicalSignal( indicator="SETUP_REVERSAL", signal_type="BUY" if market_data.rsi <= config.RSI_EXTREME_OVERSOLD else "SELL", strength=95, description="🚨 SINAL FORTE: RSI extremo com EMA contrária - REVERSÃO", confidence_impact=ScoringConfig.STRONG_REVERSAL_BONUS )) # Setup 2: RSI + BB alinhados if market_data.rsi <= 35 and market_data.bb_position == 'ABAIXO': special_signals.append(TechnicalSignal( indicator="SETUP_PERFECT_BUY", signal_type="BUY", strength=100, description="🎯 SETUP PERFEITO: RSI baixo + BB abaixo - COMPRA FORTE", confidence_impact=ScoringConfig.PERFECT_SETUP_BONUS )) elif market_data.rsi >= 65 and market_data.bb_position in ['ACIMA', 'SOBRE']: special_signals.append(TechnicalSignal( indicator="SETUP_PERFECT_SELL", signal_type="SELL", strength=100, description="🎯 SETUP PERFEITO: RSI alto + BB acima - VENDA FORTE", confidence_impact=ScoringConfig.PERFECT_SETUP_BONUS )) return special_signals class TechnicalAnalysisEngine: """Engine principal de análise técnica.""" def __init__(self): self.rsi_analyzer = RSIAnalyzer() self.ema_analyzer = EMAAnalyzer() self.bb_analyzer = BollingerBandsAnalyzer() self.momentum_analyzer = MomentumAnalyzer() self.volume_analyzer = VolumeAnalyzer() self.setup_detector = ScalpingSetupDetector() self.fibonacci_engine = AdvancedFibonacciEngine() self.log_parser = VampireBotLogParser() self.advanced_processor = AdvancedMarketProcessor() self.config = TechnicalAnalysisConfig logger.info("TechnicalAnalysisEngine inicializado com processamento avançado") def analyze(self, market_data: MarketData) -> Dict[str, Any]: """Executa análise técnica completa.""" # Análises individuais signals = [ self.rsi_analyzer.analyze(market_data.rsi), self.ema_analyzer.analyze(market_data.ema_trend), self.bb_analyzer.analyze(market_data.bb_position), self.momentum_analyzer.analyze(market_data.variation), self.volume_analyzer.analyze(market_data.volume) ] # Detectar setups especiais special_signals = self.setup_detector.detect_perfect_setups(market_data, signals) all_signals = signals + special_signals # Análise avançada de Fibonacci fibonacci_analysis = self._perform_fibonacci_analysis(market_data) # Processamento avançado de mercado advanced_analysis = self._perform_advanced_market_analysis(market_data) # Calcular ação e confiança action, confidence = self._calculate_action_and_confidence(all_signals) return { 'action': action, 'confidence': confidence, 'signals': all_signals, 'fibonacci': fibonacci_analysis, 'advanced_analysis': advanced_analysis, 'market_data': market_data } def _calculate_action_and_confidence(self, signals: List[TechnicalSignal]) -> tuple[str, int]: """Calcula a ação recomendada e nível de confiança.""" buy_score = 0 sell_score = 0 confidence_score = 0 # Somar pontuações por tipo de sinal for signal in signals: confidence_score += signal.confidence_impact if signal.signal_type == "BUY": buy_score += signal.strength elif signal.signal_type == "SELL": sell_score += signal.strength # Determinar ação baseada nas pontuações if buy_score > sell_score and buy_score > 100: action = "COMPRAR" elif sell_score > buy_score and sell_score > 100: action = "VENDER" else: action = "AGUARDAR" # Aplicar penalidade por conflito if abs(buy_score - sell_score) < 50 and max(buy_score, sell_score) > 100: confidence_score -= ScoringConfig.CONFLICT_PENALTY # Limitar confiança confidence_score = max(ScoringConfig.MIN_CONFIDENCE, min(ScoringConfig.MAX_CONFIDENCE, confidence_score)) return action, confidence_score def _perform_fibonacci_analysis(self, market_data: MarketData) -> Dict[str, Any]: """Executa análise avançada de Fibonacci.""" try: # Simular dados de preço para análise Fibonacci prices = np.array([market_data.price * (1 + np.random.normal(0, 0.01)) for _ in range(100)]) # Executar análise Fibonacci fib_analysis = self.fibonacci_engine.analyze_fibonacci_levels( prices=prices, current_price=market_data.price ) return { 'levels': fib_analysis.levels if fib_analysis else {}, 'signals': fib_analysis.signals if fib_analysis else [], 'confluence_zones': fib_analysis.confluence_zones if fib_analysis else [], 'strength': fib_analysis.overall_strength if fib_analysis else 0 } except Exception as e: logger.error(f"Erro na análise Fibonacci: {e}") return { 'levels': {}, 'signals': [], 'confluence_zones': [], 'strength': 0 } def process_bot_log_data(self, log_content: str) -> Dict[str, Any]: """Processa dados de log do bot externo.""" try: # Parse do log bot_analysis = self.log_parser.parse_log(log_content) if not bot_analysis: return {'error': 'Falha ao processar log do bot'} # Converter para MarketData market_data = MarketData( price=bot_analysis.market_info.price, variation=0, # Será calculado se necessário rsi=bot_analysis.technical_indicators.rsi if bot_analysis.technical_indicators else 50, ema_trend=bot_analysis.technical_indicators.ema if bot_analysis.technical_indicators else 'NEUTRO', bb_position=bot_analysis.technical_indicators.bollinger if bot_analysis.technical_indicators else 'DENTRO', volume=bot_analysis.market_info.volume ) # Executar análise completa analysis_result = self.analyze(market_data) # Adicionar dados específicos do bot analysis_result['bot_data'] = { 'fibonacci_alerts': bot_analysis.fibonacci_analysis.alerts if bot_analysis.fibonacci_analysis else 0, 'fibonacci_signal': bot_analysis.fibonacci_analysis.signal if bot_analysis.fibonacci_analysis else 'UNKNOWN', 'technical_indicators': { 'rsi': bot_analysis.technical_indicators.rsi if bot_analysis.technical_indicators else None, 'ema': bot_analysis.technical_indicators.ema if bot_analysis.technical_indicators else None, 'bollinger': bot_analysis.technical_indicators.bollinger if bot_analysis.technical_indicators else None, 'atr': bot_analysis.technical_indicators.atr if bot_analysis.technical_indicators else None }, 'original_analysis': bot_analysis } return analysis_result except Exception as e: logger.error(f"Erro ao processar dados do bot: {e}") return {'error': f'Erro no processamento: {str(e)}'} def _perform_advanced_market_analysis(self, market_data: MarketData) -> Dict[str, Any]: """Executa análise avançada de mercado com swing points e padrões harmônicos.""" try: # Simular dados históricos de preço para análise base_price = market_data.price prices = np.array([base_price * (1 + np.random.normal(0, 0.02)) for _ in range(100)]) volumes = np.array([1000 + np.random.randint(-200, 200) for _ in range(100)]) # Níveis de Fibonacci simulados fibonacci_levels = { '23.6%': base_price * 0.764, '38.2%': base_price * 0.618, '50.0%': base_price * 0.5, '61.8%': base_price * 0.382, '78.6%': base_price * 0.214 } # Níveis de suporte/resistência simulados support_resistance = [ base_price * 0.95, base_price * 0.98, base_price * 1.02, base_price * 1.05 ] # Executar processamento avançado advanced_result = self.advanced_processor.process_market_data( prices=prices, volumes=volumes, fibonacci_levels=fibonacci_levels, support_resistance_levels=support_resistance ) return advanced_result except Exception as e: logger.error(f"Erro na análise avançada de mercado: {e}") return { 'swing_points': {'count': 0, 'highs': [], 'lows': [], 'avg_strength': 0}, 'confluence_zones': {'count': 0, 'zones': [], 'strongest_zone': None}, 'harmonic_patterns': {'count': 0, 'patterns': [], 'most_reliable': None}, 'market_structure': 'UNKNOWN', 'key_levels': [] } class MarketAnalyzer: """Analisador principal de mercado.""" pass class RiskCalculator: """Calculadora de risco para trading.""" @staticmethod def calculate_stop_loss(price: float, action: str) -> float: """Calcula stop loss baseado no preço e ação.""" config = TradingConfig() stop_distance = price * config.STOP_LOSS_PERCENTAGE if action == "COMPRAR": return price - stop_distance elif action == "VENDER": return price + stop_distance else: return 0 @staticmethod def calculate_take_profit(price: float, action: str) -> float: """Calcula take profit baseado no preço e ação.""" config = TradingConfig() profit_distance = price * config.TAKE_PROFIT_PERCENTAGE if action == "COMPRAR": return price + profit_distance elif action == "VENDER": return price - profit_distance else: return 0 @staticmethod def get_risk_reward_ratio() -> float: """Retorna a relação risco/recompensa configurada.""" return TradingConfig.RISK_REWARD_RATIO