"""Módulo de utilitários e funções auxiliares.""" import json from datetime import datetime from typing import Dict, Any, Optional from config.config import ( TradingConfig, UIConfig, ScoringConfig ) class DateTimeUtils: """Utilitários para manipulação de data e hora.""" @staticmethod def get_current_timestamp() -> str: """Retorna timestamp atual formatado.""" return datetime.now().strftime("%H:%M:%S") @staticmethod def get_current_datetime() -> str: """Retorna data e hora atual formatada.""" return datetime.now().strftime("%d/%m/%Y %H:%M:%S") @staticmethod def format_timestamp(dt: datetime) -> str: """Formata datetime para timestamp.""" return dt.strftime("%H:%M:%S") class NumberUtils: """Utilitários para manipulação de números.""" @staticmethod def format_price(price: float) -> str: """Formata preço com separadores de milhares.""" return f"{price:,.0f}" @staticmethod def format_percentage(value: float) -> str: """Formata porcentagem com sinal.""" return f"{value:+.2f}%" @staticmethod def format_volume(volume: float) -> str: """Formata volume com uma casa decimal.""" return f"{volume:.1f}x" @staticmethod def calculate_points_from_percentage(price: float, percentage: float) -> float: """Calcula pontos baseado em porcentagem do preço.""" return price * (percentage / 100) class ConfidenceUtils: """Utilitários para manipulação de níveis de confiança.""" @staticmethod def get_confidence_level(confidence: int) -> str: """Retorna nível de confiança textual.""" config = TradingConfig.CONFIDENCE_LEVELS if confidence >= config['MUITO_ALTA']: return "MUITO ALTA" elif confidence >= config['ALTA']: return "ALTA" elif confidence >= config['MODERADA']: return "MODERADA" else: return "BAIXA" @staticmethod def generate_confidence_bar(confidence: int) -> str: """Gera barra visual de confiança.""" filled_bars = int(confidence / 10) empty_bars = 10 - filled_bars return "█" * filled_bars + "░" * empty_bars @staticmethod def is_high_confidence(confidence: int) -> bool: """Verifica se confiança é alta.""" return confidence >= TradingConfig.CONFIDENCE_LEVELS['ALTA'] class ActionUtils: """Utilitários para manipulação de ações de trading.""" @staticmethod def get_action_emojis(action: str) -> Dict[str, str]: """Retorna emojis para ação específica.""" return UIConfig.ACTION_EMOJIS.get(action, { 'main': '⚪', 'action': '❓' }) def calculate_rsi(prices: list, period: int = 14) -> float: """Calcula o RSI (Relative Strength Index).""" if len(prices) < period + 1: return 50.0 # Valor neutro se não há dados suficientes gains = [] losses = [] for i in range(1, len(prices)): change = prices[i] - prices[i-1] if change > 0: gains.append(change) losses.append(0) else: gains.append(0) losses.append(abs(change)) if len(gains) < period: return 50.0 avg_gain = sum(gains[-period:]) / period avg_loss = sum(losses[-period:]) / period if avg_loss == 0: return 100.0 rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return rsi def calculate_bollinger_bands(prices: list, period: int = 20, std_dev: float = 2.0) -> dict: """Calcula as Bandas de Bollinger.""" if len(prices) < period: return {'upper': 0, 'middle': 0, 'lower': 0} recent_prices = prices[-period:] sma = sum(recent_prices) / period variance = sum((price - sma) ** 2 for price in recent_prices) / period std = variance ** 0.5 upper_band = sma + (std_dev * std) lower_band = sma - (std_dev * std) return { 'upper': upper_band, 'middle': sma, 'lower': lower_band } def calculate_ema(prices: list, period: int = 9) -> float: """Calcula a EMA (Exponential Moving Average).""" if len(prices) < period: return sum(prices) / len(prices) if prices else 0 multiplier = 2 / (period + 1) ema = sum(prices[:period]) / period # SMA inicial for price in prices[period:]: ema = (price * multiplier) + (ema * (1 - multiplier)) return ema def format_number(value: float, decimals: int = 2) -> str: """Formata número com casas decimais especificadas.""" return f"{value:.{decimals}f}" @staticmethod def get_action_color(action: str) -> str: """Retorna cor para ação específica.""" return UIConfig.ACTION_COLORS.get(action, 'cinza') @staticmethod def get_trading_direction(action: str) -> str: """Retorna direção de trading para ação.""" return UIConfig.TRADING_DIRECTIONS.get(action, 'INDEFINIDO') class SentimentUtils: """Utilitários para manipulação de sentimento.""" @staticmethod def get_sentiment_emoji(sentiment_label: str) -> str: """Retorna emoji para sentimento.""" return UIConfig.SENTIMENT_EMOJIS.get(sentiment_label, '😐💛') @staticmethod def normalize_sentiment_label(label: str) -> str: """Normaliza label de sentimento.""" label_upper = label.upper() valid_labels = ['POSITIVO', 'NEGATIVO', 'NEUTRO'] if label_upper in valid_labels: return label_upper # Mapeamento de labels alternativos label_mapping = { 'POSITIVE': 'POSITIVO', 'NEGATIVE': 'NEGATIVO', 'NEUTRAL': 'NEUTRO', 'POS': 'POSITIVO', 'NEG': 'NEGATIVO', 'NEU': 'NEUTRO' } return label_mapping.get(label_upper, 'NEUTRO') class ValidationUtils: """Utilitários para validação de dados.""" @staticmethod def validate_market_data(data: Dict[str, Any]) -> bool: """Valida dados de mercado.""" required_fields = ['price', 'variation', 'rsi', 'ema_trend', 'bb_position', 'volume'] # Verificar se todos os campos obrigatórios estão presentes for field in required_fields: if field not in data: return False # Validar tipos e valores try: price = float(data['price']) variation = float(data['variation']) rsi = int(data['rsi']) volume = float(data['volume']) # Validar ranges if price < 0 or not (0 <= rsi <= 100) or volume < 0: return False # Validar strings valid_ema_trends = ['ALTA', 'BAIXA', 'NEUTRO'] valid_bb_positions = ['DENTRO', 'SOBRE', 'ABAIXO', 'ACIMA'] if (data['ema_trend'] not in valid_ema_trends or data['bb_position'] not in valid_bb_positions): return False return True except (ValueError, TypeError): return False @staticmethod def validate_confidence_score(score: int) -> int: """Valida e normaliza pontuação de confiança.""" return max(ScoringConfig.MIN_CONFIDENCE, min(ScoringConfig.MAX_CONFIDENCE, score)) @staticmethod def validate_text_input(text: str) -> bool: """Valida entrada de texto.""" if not text or not isinstance(text, str): return False # Verificar se não é apenas espaços em branco if not text.strip(): return False # Verificar tamanho mínimo if len(text.strip()) < 3: return False return True class FormatUtils: """Utilitários para formatação de texto e dados.""" @staticmethod def format_signal_list(signals: list) -> str: """Formata lista de sinais para exibição.""" if not signals: return "Nenhum sinal detectado" formatted_signals = [] for i, signal in enumerate(signals[:5], 1): # Máximo 5 sinais if hasattr(signal, 'description'): formatted_signals.append(f"{i}. {signal.description}") else: formatted_signals.append(f"{i}. {str(signal)}") return "\n".join(formatted_signals) @staticmethod def format_market_summary(market_data: Dict[str, Any]) -> str: """Formata resumo dos dados de mercado.""" price = NumberUtils.format_price(market_data.get('price', 0)) variation = NumberUtils.format_percentage(market_data.get('variation', 0)) volume = NumberUtils.format_volume(market_data.get('volume', 0)) return f"""• **Preço:** {price} • **Variação:** {variation} • **RSI:** {market_data.get('rsi', 'N/A')} • **EMA:** {market_data.get('ema_trend', 'N/A')} • **Bollinger:** {market_data.get('bb_position', 'N/A')} • **Volume:** {volume}""" @staticmethod def format_trading_recommendations(action: str, price: float) -> str: """Formata recomendações de trading.""" if action == 'COMPRAR': stop_loss = price * (1 - TradingConfig.STOP_LOSS_PERCENTAGE) take_profit = price * (1 + TradingConfig.TAKE_PROFIT_PERCENTAGE) return f"""• **Stop Loss:** -{NumberUtils.calculate_points_from_percentage(price, TradingConfig.STOP_LOSS_PERCENTAGE * 100):.0f} pts ({TradingConfig.STOP_LOSS_PERCENTAGE * 100:.2f}%) • **Take Profit:** +{NumberUtils.calculate_points_from_percentage(price, TradingConfig.TAKE_PROFIT_PERCENTAGE * 100):.0f} pts ({TradingConfig.TAKE_PROFIT_PERCENTAGE * 100:.2f}%) • **Timeframe:** {'/'.join(TradingConfig.SCALPING_TIMEFRAMES)} • **Risk/Reward:** 1:{TradingConfig.RISK_REWARD_RATIO}""" elif action == 'VENDER': stop_loss = price * (1 + TradingConfig.STOP_LOSS_PERCENTAGE) take_profit = price * (1 - TradingConfig.TAKE_PROFIT_PERCENTAGE) return f"""• **Stop Loss:** +{NumberUtils.calculate_points_from_percentage(price, TradingConfig.STOP_LOSS_PERCENTAGE * 100):.0f} pts ({TradingConfig.STOP_LOSS_PERCENTAGE * 100:.2f}%) • **Take Profit:** -{NumberUtils.calculate_points_from_percentage(price, TradingConfig.TAKE_PROFIT_PERCENTAGE * 100):.0f} pts ({TradingConfig.TAKE_PROFIT_PERCENTAGE * 100:.2f}%) • **Timeframe:** {'/'.join(TradingConfig.SCALPING_TIMEFRAMES)} • **Risk/Reward:** 1:{TradingConfig.RISK_REWARD_RATIO}""" else: return """• **Aguardar:** Setup mais definido • **Monitorar:** Rompimentos de suporte/resistência • **Observar:** Confluência de sinais técnicos""" class LogUtils: """Utilitários para logging e debug.""" @staticmethod def log_analysis_result(analysis_result: Dict[str, Any]) -> None: """Registra resultado de análise para debug.""" timestamp = DateTimeUtils.get_current_datetime() action = analysis_result.get('action', 'UNKNOWN') confidence = analysis_result.get('confidence', 0) print(f"[{timestamp}] Análise: {action} (Confiança: {confidence}%)") @staticmethod def log_error(error_message: str, context: str = "") -> None: """Registra erro com contexto.""" timestamp = DateTimeUtils.get_current_datetime() context_str = f" [{context}]" if context else "" print(f"[{timestamp}] ERRO{context_str}: {error_message}") @staticmethod def log_model_status(model_info: Dict[str, Any]) -> None: """Registra status do modelo de IA.""" timestamp = DateTimeUtils.get_current_datetime() status = "ATIVO" if model_info.get('available', False) else "INATIVO" model_name = model_info.get('description', 'Desconhecido') print(f"[{timestamp}] Modelo IA: {status} - {model_name}") class DataExportUtils: """Utilitários para exportação de dados.""" @staticmethod def export_analysis_to_json(analysis_result: Dict[str, Any]) -> str: """Exporta resultado de análise para JSON.""" # Preparar dados para serialização export_data = { 'timestamp': DateTimeUtils.get_current_datetime(), 'action': analysis_result.get('action'), 'confidence': analysis_result.get('confidence'), 'market_data': analysis_result.get('market_data'), 'sentiment': analysis_result.get('sentiment') } # Converter objetos complexos para dicionários if 'signals' in analysis_result: export_data['signals'] = [ { 'indicator': getattr(signal, 'indicator', 'unknown'), 'signal_type': getattr(signal, 'signal_type', 'unknown'), 'strength': getattr(signal, 'strength', 0), 'description': getattr(signal, 'description', '') } for signal in analysis_result['signals'] ] return json.dumps(export_data, indent=2, ensure_ascii=False) @staticmethod def create_analysis_summary(analysis_result: Dict[str, Any]) -> Dict[str, Any]: """Cria resumo da análise para relatórios.""" return { 'timestamp': DateTimeUtils.get_current_datetime(), 'action': analysis_result.get('action', 'UNKNOWN'), 'confidence': analysis_result.get('confidence', 0), 'confidence_level': ConfidenceUtils.get_confidence_level( analysis_result.get('confidence', 0) ), 'signals_count': len(analysis_result.get('signals', [])), 'sentiment_label': analysis_result.get('sentiment', {}).get('label', 'NEUTRO'), 'market_price': analysis_result.get('market_data', {}).get('price', 0), 'market_rsi': analysis_result.get('market_data', {}).get('rsi', 50) }