"""Aplicação de Análise de Mercado com IA para Scalping - Versão Refatorada.""" from typing import Dict, Any, Optional import gradio as gr from datetime import datetime # Importar módulos refatorados try: from config.config import AppConfig from src.analysis.market_analysis import TechnicalAnalysisEngine from src.analysis.sentiment_analysis import SentimentAnalysisEngine from src.ui.gradio_interface import GradioInterface from src.utils.utils import LogUtils, ValidationUtils from src.core.log_parser import VampireBotLogParser from src.analysis.fibonacci_analysis import AdvancedFibonacciEngine from src.core.advanced_market_processing import AdvancedMarketProcessor from src.integrations.real_time_integration import RealTimeIntegration, BotEvent from src.core.performance_monitor import PerformanceMonitor, measure_analysis_time from src.utils.request_logger import log_requests_responses, enable_logging, disable_logging # Sistema de logging avançado from src.core.database_logger import get_logger, initialize_logger, LogLevel, LogCategory from src.utils.logging_decorators import log_execution, log_api_call, log_ai_model_usage, LoggingContext, quick_log from src.ui.log_viewer import create_log_viewer_interface # Inicializar o sistema de logging db_logger = initialize_logger("logs/application.db") except ImportError: # Fallback para modo standalone se módulos não existirem print("⚠️ Módulos refatorados não encontrados. Executando em modo standalone.") AppConfig = None TechnicalAnalysisEngine = None SentimentAnalysisEngine = None GradioInterface = None LogUtils = None ValidationUtils = None VampireBotLogParser = None AdvancedFibonacciEngine = None AdvancedMarketProcessor = None RealTimeIntegration = None BotEvent = None PerformanceMonitor = None measure_analysis_time = None log_requests_responses = None enable_logging = None disable_logging = None # Fallback para sistema de logging get_logger = None initialize_logger = None LogLevel = None LogCategory = None log_execution = None log_api_call = None log_ai_model_usage = None LoggingContext = None quick_log = None create_log_viewer_interface = None db_logger = None # Engines de análise technical_engine = None sentiment_engine = None real_time_integration = None model_info = {'available': False, 'description': 'IA Indisponível'} # Estado global para eventos em tempo real recent_events = [] max_events_display = 50 # Monitor de performance global performance_monitor = PerformanceMonitor() if PerformanceMonitor else None # Inicializar engines def initialize_engines(): """Inicializa engines de análise técnica e de sentimento.""" global technical_engine, sentiment_engine, model_info try: # Log do início da inicialização if quick_log: quick_log("Iniciando inicialização dos engines", LogLevel.INFO, LogCategory.SYSTEM) # Habilitar logging de requisições/respostas if enable_logging: enable_logging() print("🔍 Sistema de logging de requisições habilitado") if TechnicalAnalysisEngine and SentimentAnalysisEngine: # Inicializar engine de análise técnica technical_engine = TechnicalAnalysisEngine() if LogUtils: LogUtils.log_analysis_result({'action': 'ENGINE_INIT', 'confidence': 100}) # Inicializar engine de análise de sentimento sentiment_engine = SentimentAnalysisEngine() # Obter informações do modelo de IA model_info = sentiment_engine.get_model_info() if LogUtils: LogUtils.log_model_status(model_info) # Log de sucesso if quick_log: quick_log("Engines inicializados com sucesso", LogLevel.INFO, LogCategory.SYSTEM, metadata={'engines': ['Technical Analysis', 'Sentiment Analysis']}) # Log evento do sistema if db_logger: db_logger.log_system_event( event_type="INITIALIZATION", event_name="ENGINES_STARTED", description="Todos os engines foram inicializados com sucesso", severity="INFO", metadata=model_info ) print("✅ Engines inicializadas com sucesso") else: # Fallback para modo standalone initialize_standalone_mode() except Exception as e: # Log de erro if quick_log: quick_log(f"Erro ao inicializar engines: {e}", LogLevel.ERROR, LogCategory.SYSTEM) if db_logger: db_logger.log_system_event( event_type="ERROR", event_name="ENGINES_INITIALIZATION_FAILED", description=f"Falha na inicialização dos engines: {str(e)}", severity="ERROR", metadata={'error': str(e)} ) print(f"Erro na inicialização: {str(e)}") # Fallback para modo básico initialize_standalone_mode() def initialize_standalone_mode(): """Inicializa modo standalone com funcionalidades básicas.""" global model_info # Importações opcionais para IA try: from transformers import AutoTokenizer, AutoModelForSequenceClassification, pipeline import torch TRANSFORMERS_AVAILABLE = True except ImportError: TRANSFORMERS_AVAILABLE = False print("⚠️ Transformers não disponível. Executando sem IA.") # Lista de modelos alternativos (do mais específico para o mais geral) FINANCIAL_MODELS = [ { "name": "ProsusAI/finbert", "description": "FinBERT - Modelo especializado em sentimento financeiro", "labels": {"LABEL_0": "NEGATIVO", "LABEL_1": "NEUTRO", "LABEL_2": "POSITIVO"} }, { "name": "mrm8488/distilroberta-finetuned-financial-news-sentiment-analysis", "description": "DistilRoBERTa - Modelo leve para notícias financeiras", "labels": {"LABEL_0": "NEGATIVO", "LABEL_1": "NEUTRO", "LABEL_2": "POSITIVO"} }, { "name": "soleimanian/financial-roberta-large-sentiment", "description": "Financial RoBERTa - Modelo robusto para textos financeiros", "labels": {"LABEL_0": "NEGATIVO", "LABEL_1": "NEUTRO", "LABEL_2": "POSITIVO"} }, { "name": "cardiffnlp/twitter-roberta-base-sentiment-latest", "description": "RoBERTa - Modelo geral de sentimento (fallback)", "labels": {"LABEL_0": "NEGATIVO", "LABEL_1": "NEUTRO", "LABEL_2": "POSITIVO"} } ] # Inicializar modelo de sentimento sentiment_pipeline = None current_model_info = None if TRANSFORMERS_AVAILABLE: for model_config in FINANCIAL_MODELS: try: print(f"🔄 Tentando carregar: {model_config['description']}") sentiment_pipeline = pipeline( "sentiment-analysis", model=model_config["name"], return_all_scores=True ) current_model_info = model_config print(f"✅ {model_config['description']} carregado com sucesso!") break except Exception as e: print(f"❌ Falha ao carregar {model_config['name']}: {e}") continue if sentiment_pipeline is None: print("❌ Nenhum modelo de sentimento pôde ser carregado.") TRANSFORMERS_AVAILABLE = False else: print("⚠️ Transformers não disponível. Executando sem análise de sentimento IA.") # Atualizar model_info global if current_model_info: model_info = {'available': True, 'description': current_model_info['description']} else: model_info = {'available': False, 'description': 'IA Indisponível'} def parse_market_data(text): """Extrai dados de mercado do texto de entrada usando engines ou fallback.""" if technical_engine and hasattr(technical_engine, 'parse_market_data'): return technical_engine.parse_market_data(text) # Fallback para modo standalone import re try: # Regex patterns para extrair dados price_match = re.search(r'P:([0-9,]+\.?\d*)', text) variation_match = re.search(r'\(([\+\-]\d+\.?\d*%?)\)', text) rsi_match = re.search(r'RSI:(\d+)', text) ema_match = re.search(r'EMA:(ALTA|BAIXA)', text) bb_match = re.search(r'BB:(DENTRO|SOBRE|ABAIXO|ACIMA)', text) vol_match = re.search(r'Vol:([0-9\.]+)', text) # Extrair valores 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 { 'price': price, 'variation': variation, 'rsi': rsi, 'ema_trend': ema_trend, 'bb_position': bb_position, 'volume': volume } except Exception as e: return None @log_requests_responses("analyze_sentiment") if log_requests_responses else lambda f: f @log_ai_model_usage("sentiment_analysis") if log_ai_model_usage else lambda f: f def analyze_sentiment(text): """Analisa o sentimento do texto usando engines ou fallback.""" if sentiment_engine and hasattr(sentiment_engine, 'analyze_sentiment'): return sentiment_engine.analyze_sentiment(text) # Fallback para modo standalone import re # Verificar se temos pipeline carregado no modo standalone if 'sentiment_pipeline' in globals() and sentiment_pipeline is not None: try: # Limpar e preparar o texto para análise clean_text = re.sub(r'[^\w\s\+\-\%\.]', ' ', text) clean_text = clean_text[:512] # Limitar tamanho para o modelo result = sentiment_pipeline(clean_text) # Processar resultado baseado no modelo if isinstance(result, list) and len(result) > 0: # Se return_all_scores=True, pegar o resultado com maior score if isinstance(result[0], list): predictions = result[0] best_prediction = max(predictions, key=lambda x: x['score']) else: best_prediction = result[0] # Mapear label usando o mapeamento do modelo atual label = best_prediction['label'] confidence = best_prediction['score'] # Usar mapeamento específico do modelo ou fallback genérico if 'current_model_info' in globals() and current_model_info and label in current_model_info['labels']: sentiment_label = current_model_info['labels'][label] else: # Fallback para mapeamento genérico label_lower = label.lower() if 'neg' in label_lower: sentiment_label = 'NEGATIVO' elif 'pos' in label_lower: sentiment_label = 'POSITIVO' else: sentiment_label = 'NEUTRO' return { 'sentiment': label.lower(), 'confidence': confidence, 'label': sentiment_label } except Exception as e: print(f"Erro na análise de sentimento: {e}") # Fallback padrão return { 'sentiment': 'neutral', 'confidence': 0.5, 'label': 'NEUTRO' } @measure_analysis_time if measure_analysis_time else lambda f: f @log_execution(LogCategory.MARKET_ANALYSIS, log_performance=True) if log_execution else lambda f: f def analyze_scalping_signals(market_data, original_text=""): """Analisa sinais para scalping usando engines ou fallback.""" if technical_engine and hasattr(technical_engine, 'analyze_scalping_signals'): return technical_engine.analyze_scalping_signals(market_data, original_text) # Fallback para modo standalone if not market_data: return { 'action': 'AGUARDAR', 'confidence': 0, 'signals': [], 'market_data': {}, 'sentiment': {'label': 'NEUTRO', 'confidence': 0.5} } # Análise de sentimento com FinBERT sentiment_analysis = analyze_sentiment(original_text) price = market_data['price'] variation = market_data['variation'] rsi = market_data['rsi'] ema_trend = market_data['ema_trend'] bb_position = market_data['bb_position'] volume = market_data['volume'] # Pontuação de confiança confidence_score = 0 signals = [] action = 'AGUARDAR' # === ANÁLISE RSI === if rsi <= 30: # Oversold signals.append("RSI em zona de sobrevenda ({}): COMPRA".format(rsi)) confidence_score += 25 if action == 'AGUARDAR': action = 'COMPRAR' elif rsi >= 70: # Overbought signals.append("RSI em zona de sobrecompra ({}): VENDA".format(rsi)) confidence_score += 25 if action == 'AGUARDAR': action = 'VENDER' elif 45 <= rsi <= 55: signals.append("RSI neutro ({}): aguardar confirmação".format(rsi)) # === ANÁLISE EMA === if ema_trend == 'ALTA': signals.append("Tendência EMA ALTA: viés de COMPRA") confidence_score += 15 if action == 'VENDER': confidence_score -= 10 # Conflito elif action == 'AGUARDAR': action = 'COMPRAR' elif ema_trend == 'BAIXA': signals.append("Tendência EMA BAIXA: viés de VENDA") confidence_score += 15 if action == 'COMPRAR': confidence_score -= 10 # Conflito elif action == 'AGUARDAR': action = 'VENDER' # === ANÁLISE BOLLINGER BANDS === if bb_position == 'ABAIXO': signals.append("Preço abaixo da banda inferior: COMPRA (reversão)") confidence_score += 20 if action == 'AGUARDAR': action = 'COMPRAR' elif bb_position == 'ACIMA' or bb_position == 'SOBRE': signals.append("Preço acima da banda superior: VENDA (reversão)") confidence_score += 20 if action == 'AGUARDAR': action = 'VENDER' elif bb_position == 'DENTRO': signals.append("Preço dentro das bandas: aguardar breakout") confidence_score += 5 # === ANÁLISE DE MOMENTUM (Variação) === if abs(variation) >= 0.05: # Movimento significativo if variation > 0: signals.append("Momentum positivo (+{:.2f}%): seguir tendência".format(variation)) confidence_score += 10 if action == 'VENDER': confidence_score -= 5 else: signals.append("Momentum negativo ({:.2f}%): seguir tendência".format(variation)) confidence_score += 10 if action == 'COMPRAR': confidence_score -= 5 # === ANÁLISE DE VOLUME === if volume > 1.0: # Volume alto signals.append("Volume alto ({:.1f}x): confirma movimento".format(volume)) confidence_score += 10 elif volume < 0.5: # Volume baixo signals.append("Volume baixo ({:.1f}x): cuidado com falsos sinais".format(volume)) confidence_score -= 5 # === ANÁLISE DE SENTIMENTO IA (FinBERT) === sentiment_label = sentiment_analysis['label'] sentiment_conf = sentiment_analysis['confidence'] if sentiment_label == 'POSITIVO': signals.append("🤖 IA Sentimento: POSITIVO ({:.1f}%): viés de COMPRA".format(sentiment_conf * 100)) confidence_score += int(sentiment_conf * 20) # Até 20 pontos if action == 'AGUARDAR': action = 'COMPRAR' elif action == 'VENDER': confidence_score -= 10 # Conflito com sentimento elif sentiment_label == 'NEGATIVO': signals.append("🤖 IA Sentimento: NEGATIVO ({:.1f}%): viés de VENDA".format(sentiment_conf * 100)) confidence_score += int(sentiment_conf * 20) # Até 20 pontos if action == 'AGUARDAR': action = 'VENDER' elif action == 'COMPRAR': confidence_score -= 10 # Conflito com sentimento else: signals.append("🤖 IA Sentimento: NEUTRO ({:.1f}%): sem viés claro".format(sentiment_conf * 100)) # === REGRAS ESPECÍFICAS DE SCALPING === # Scalping: RSI extremo + EMA contrária = reversão forte if (rsi <= 25 and ema_trend == 'BAIXA') or (rsi >= 75 and ema_trend == 'ALTA'): signals.append("🚨 SINAL FORTE: RSI extremo com EMA contrária - REVERSÃO") confidence_score += 30 # Scalping: RSI + BB alinhados if rsi <= 35 and bb_position == 'ABAIXO': signals.append("🎯 SETUP PERFEITO: RSI baixo + BB abaixo - COMPRA FORTE") confidence_score += 35 action = 'COMPRAR' elif rsi >= 65 and (bb_position == 'ACIMA' or bb_position == 'SOBRE'): signals.append("🎯 SETUP PERFEITO: RSI alto + BB acima - VENDA FORTE") confidence_score += 35 action = 'VENDER' # Limitar confiança confidence_score = min(confidence_score, 95) confidence_score = max(confidence_score, 10) return { 'action': action, 'confidence': confidence_score, 'signals': signals, 'market_data': market_data, 'sentiment': sentiment_analysis } def generate_trading_response(analysis): """Gera resposta formatada para trading com análise de sentimento IA""" action = analysis['action'] confidence = analysis['confidence'] signals = analysis.get('signals', []) market_data = analysis.get('market_data', {}) sentiment = analysis.get('sentiment', {'label': 'NEUTRO', 'confidence': 0.5}) # Emojis e cores baseados na ação if action == 'COMPRAR': emoji = "🟢" action_emoji = "📈" color = "verde" direction = "LONG" elif action == 'VENDER': emoji = "🔴" action_emoji = "📉" color = "vermelho" direction = "SHORT" else: emoji = "🟡" action_emoji = "⏸️" color = "amarelo" direction = "NEUTRO" # Emojis para sentimento sentiment_emojis = { 'POSITIVO': '😊💚', 'NEGATIVO': '😟💔', 'NEUTRO': '😐💛' } # Barra de confiança confidence_bars = "█" * int(confidence / 10) + "░" * (10 - int(confidence / 10)) # Classificação de confiança if confidence >= 80: conf_level = "MUITO ALTA" elif confidence >= 65: conf_level = "ALTA" elif confidence >= 50: conf_level = "MODERADA" else: conf_level = "BAIXA" # Timestamp timestamp = datetime.now().strftime("%H:%M:%S") response = f"""{emoji} **ANÁLISE SCALPING - WIN M1/M5** **⏰ Timestamp:** {timestamp} **📊 Instrumento:** WINV25 **🎯 DECISÃO DE TRADING:** • **Ação:** {action} {action_emoji} • **Direção:** {direction} • **Confiança:** {confidence:.0f}% ({conf_level}) • **Visual:** {confidence_bars} **🤖 ANÁLISE DE SENTIMENTO IA (FinBERT):** • **Sentimento:** {sentiment_emojis.get(sentiment['label'], '😐💛')} **{sentiment['label']}** ({sentiment['confidence']*100:.1f}%) **📈 DADOS DE MERCADO:** • **Preço:** {market_data.get('price', 'N/A'):,.0f} • **Variação:** {market_data.get('variation', 0):+.2f}% • **RSI:** {market_data.get('rsi', 'N/A')} • **EMA:** {market_data.get('ema_trend', 'N/A')} • **Bollinger:** {market_data.get('bb_position', 'N/A')} • **Volume:** {market_data.get('volume', 0):.1f}x **🔍 ANÁLISE TÉCNICA + IA:**""" for i, signal in enumerate(signals[:5], 1): # Máximo 5 sinais response += f"\n{i}. {signal}" # Recomendações específicas response += f"\n\n**⚡ RECOMENDAÇÕES SCALPING:**" if action == 'COMPRAR': response += f""" • **Stop Loss:** -{market_data.get('price', 0) * 0.0007:.0f} pts (0.07%) • **Take Profit:** +{market_data.get('price', 0) * 0.0015:.0f} pts (0.15%) • **Timeframe:** M1/M5 • **Risk/Reward:** 1:2""" elif action == 'VENDER': response += f""" • **Stop Loss:** +{market_data.get('price', 0) * 0.0007:.0f} pts (0.07%) • **Take Profit:** -{market_data.get('price', 0) * 0.0015:.0f} pts (0.15%) • **Timeframe:** M1/M5 • **Risk/Reward:** 1:2""" else: response += """ • **Aguardar:** Setup mais definido • **Monitorar:** Rompimentos de suporte/resistência • **Observar:** Confluência de sinais técnicos""" if confidence < 60: response += f"\n\n⚠️ **ATENÇÃO:** Confiança {conf_level} - Aguarde confirmação adicional!" response += f"\n\n---\n*🤖 Análise Scalping Automatizada - {timestamp}*" return response @log_requests_responses("process_trading_analysis") if log_requests_responses else lambda f: f @log_api_call("process_trading_analysis") if log_api_call else lambda f: f def process_trading_analysis(text): """Função principal que processa a análise de trading usando engines ou fallback.""" # Validação de entrada if ValidationUtils and hasattr(ValidationUtils, 'validate_text_input'): if not ValidationUtils.validate_text_input(text): return "⚠️ **Erro:** Entrada de texto inválida ou muito curta." else: # Fallback validation if not text or not text.strip(): return "⚠️ **Erro:** Nenhum dado de mercado fornecido para análise." try: # Verificar se é um log do bot if 'VAMPIRE TRADING BOT' in text or 'FIBONACCI AVANÇADO' in text: if VampireBotLogParser: log_parser = VampireBotLogParser() return log_parser.process_bot_log(text) else: # Fallback para processamento de log return process_bot_log_fallback(text) # Parse dos dados de mercado market_data = parse_market_data(text) if not market_data: return "⚠️ **Erro:** Não foi possível extrair dados válidos do texto fornecido." # Análise dos sinais de scalping analysis = analyze_scalping_signals(market_data, text) # Log da análise se disponível if LogUtils and hasattr(LogUtils, 'log_analysis_result'): LogUtils.log_analysis_result(analysis) # Gerar resposta formatada para a interface return generate_trading_response(analysis) except Exception as e: error_msg = f"Erro durante a análise: {str(e)}" if LogUtils and hasattr(LogUtils, 'log_error'): LogUtils.log_error(error_msg, "PROCESS_ANALYSIS") return f"❌ **Erro:** {error_msg}" def process_bot_log_fallback(text): """Processamento fallback para logs do bot quando VampireBotLogParser não está disponível.""" try: # Extrair informações básicas do log import re # Buscar por padrões específicos do bot fibonacci_match = re.search(r'FIBONACCI AVANÇADO.*?Nível: ([\d.]+)', text, re.DOTALL) action_match = re.search(r'(COMPRAR|VENDER|AGUARDAR)', text) confidence_match = re.search(r'Confiança: (\d+)%', text) if fibonacci_match or action_match: response = "🤖 **ANÁLISE DE LOG DO BOT DETECTADA**\n\n" if action_match: action = action_match.group(1) response += f"**Ação Recomendada:** {action}\n" if confidence_match: confidence = confidence_match.group(1) response += f"**Confiança:** {confidence}%\n" if fibonacci_match: fib_level = fibonacci_match.group(1) response += f"**Nível Fibonacci:** {fib_level}\n" response += "\n⚠️ **Nota:** Processamento básico de log. Para análise completa, instale os módulos avançados." return response # Se não conseguir extrair dados do log, processar como entrada normal return process_trading_analysis_basic(text) except Exception as e: return f"❌ **Erro no processamento do log:** {str(e)}" def process_trading_analysis_basic(text): """Processamento básico quando não há log do bot.""" market_data = parse_market_data(text) if not market_data: return "⚠️ **Erro:** Não foi possível extrair dados válidos do texto fornecido." analysis = analyze_scalping_signals(market_data, text) return generate_trading_response(analysis) # Função principal de análise para a interface @log_requests_responses("main_analysis_function") if log_requests_responses else lambda f: f @log_api_call("main_analysis_function") if log_api_call else lambda f: f def main_analysis_function(text: str) -> str: """Função principal de análise que será usada pela interface.""" result = process_trading_analysis(text) # Se o resultado é uma string (erro ou resposta formatada), retornar diretamente if isinstance(result, str): return result # Se é um dicionário (análise estruturada), formatar para string if isinstance(result, dict): return generate_trading_response(result) return "❌ **Erro:** Resultado de análise inválido." # Criar e configurar interface def create_interface(): """Cria interface usando a nova arquitetura modular.""" if GradioInterface: # Usar interface refatorada interface = GradioInterface( analysis_function=main_analysis_function, model_info=model_info ) return interface.create_interface() else: # Fallback para interface simples return create_fallback_interface() def create_fallback_interface(): """Cria interface simples para modo standalone.""" example_data = AppConfig.EXAMPLE_INPUT if AppConfig else """IBOV: 129.847 (+0,85%) RSI: 65 EMA 9: Tendência de alta Bollinger: Preço próximo à banda superior Volume: 2.3x da média Notícias: Mercado otimista com dados do PIB""" with gr.Blocks(title="📈 Análise de Mercado IA") as interface: gr.HTML("""

📈 Análise de Mercado IA

Sistema de análise técnica para scalping

""") with gr.Tab("📊 Análise de Mercado"): with gr.Row(): with gr.Column(scale=2): market_input = gr.Textbox( label="Dados do Mercado", placeholder=example_data, lines=8 ) analyze_btn = gr.Button("🔍 Analisar", variant="primary") with gr.Column(scale=3): result_output = gr.HTML() with gr.Tab("🤖 Monitor do Bot"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 🔄 Status do Sistema") with gr.Row(): start_monitor_btn = gr.Button("▶️ Iniciar Monitor", variant="primary") stop_monitor_btn = gr.Button("⏹️ Parar Monitor", variant="secondary") monitor_status = gr.HTML( value="
Monitor parado
" ) gr.Markdown("### ⚡ Eventos Recentes") recent_events_display = gr.HTML( value="
Nenhum evento recente
" ) with gr.Column(scale=2): gr.Markdown("### 📊 Análise em Tempo Real") real_time_analysis = gr.HTML( value="
Aguardando dados do bot...
" ) log_file_path = gr.Textbox( value="d:/hugging_face_spaces/text", label="📁 Caminho do Arquivo de Log" ) check_interval = gr.Slider( minimum=0.5, maximum=10.0, value=1.0, step=0.5, label="⏱️ Intervalo de Verificação (segundos)" ) with gr.Tab("📈 Performance"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 🔧 Controles de Performance") with gr.Row(): start_perf_btn = gr.Button("▶️ Iniciar Monitoramento", variant="primary") stop_perf_btn = gr.Button("⏹️ Parar Monitoramento", variant="secondary") perf_status = gr.HTML( value="
Monitoramento parado
" ) with gr.Row(): reset_stats_btn = gr.Button("🔄 Reset Estatísticas", variant="secondary") export_metrics_btn = gr.Button("💾 Exportar Métricas", variant="secondary") gr.Markdown("### 🚨 Alertas de Performance") performance_alerts = gr.HTML( value="
Nenhum alerta ativo
" ) with gr.Column(scale=2): gr.Markdown("### 📊 Métricas do Sistema") system_metrics = gr.HTML( value="
Aguardando métricas...
" ) gr.Markdown("### 🤖 Estatísticas do Bot") bot_statistics = gr.HTML( value="
Aguardando estatísticas...
" ) gr.Markdown("### 💡 Sugestões de Otimização") optimization_suggestions = gr.HTML( value="
Execute análises para obter sugestões
" ) # Adicionar aba de logs se disponível if create_log_viewer_interface: with gr.Tab("📋 Logs do Sistema"): log_viewer = create_log_viewer_interface(db_logger) log_viewer.render() analyze_btn.click( fn=main_analysis_function, inputs=[market_input], outputs=[result_output] ) start_monitor_btn.click( fn=start_real_time_monitor, inputs=[log_file_path, check_interval], outputs=[monitor_status] ) stop_monitor_btn.click( fn=stop_real_time_monitor, outputs=[monitor_status] ) # Event handlers para performance start_perf_btn.click( fn=start_performance_monitoring, outputs=[perf_status] ) stop_perf_btn.click( fn=stop_performance_monitoring, outputs=[perf_status] ) reset_stats_btn.click( fn=reset_performance_stats, outputs=[bot_statistics] ) export_metrics_btn.click( fn=export_performance_metrics, outputs=[perf_status] ) # Auto-refresh para eventos em tempo real interface.load( fn=get_recent_events_display, outputs=[recent_events_display] ) interface.load( fn=get_real_time_analysis_display, outputs=[real_time_analysis] ) # Auto-refresh para métricas de performance interface.load( fn=get_system_metrics_display, outputs=[system_metrics] ) interface.load( fn=get_bot_statistics_display, outputs=[bot_statistics] ) interface.load( fn=get_performance_alerts_display, outputs=[performance_alerts] ) interface.load( fn=get_optimization_suggestions_display, outputs=[optimization_suggestions] ) # Log evento de inicialização da interface if db_logger: db_logger.log_system_event( event_type="INITIALIZATION", event_name="INTERFACE_CREATED", description="Interface Gradio criada com sucesso", severity="INFO", metadata={'tabs': ['Análise de Mercado', 'Monitor do Bot', 'Performance', 'Logs do Sistema']} ) return interface # Funções para integração em tempo real def start_real_time_monitor(log_path: str, interval: float): """Inicia o monitor em tempo real.""" global real_time_integration, recent_events try: if RealTimeIntegration: real_time_integration = RealTimeIntegration(log_path) real_time_integration.config.check_interval = interval # Callback para capturar eventos def event_handler(event): global recent_events event_dict = { 'timestamp': event.timestamp.strftime('%H:%M:%S'), 'type': event.event_type, 'data': event.data, 'priority': event.priority } recent_events.append(event_dict) # Manter apenas os últimos eventos if len(recent_events) > max_events_display: recent_events = recent_events[-max_events_display:] real_time_integration.processor.subscribe(event_handler) real_time_integration.start() return "
✅ Monitor iniciado com sucesso
" else: return "
⚠️ Integração em tempo real não disponível
" except Exception as e: return f"
❌ Erro ao iniciar monitor: {str(e)}
" def stop_real_time_monitor(): """Para o monitor em tempo real.""" global real_time_integration try: if real_time_integration: real_time_integration.stop() real_time_integration = None return "
⏹️ Monitor parado
" else: return "
Monitor já estava parado
" except Exception as e: return f"
❌ Erro ao parar monitor: {str(e)}
" def get_recent_events_display(): """Obtém display dos eventos recentes.""" global recent_events if not recent_events: return "
Nenhum evento recente
" html = "
" for event in recent_events[-10:]: # Últimos 10 eventos timestamp = event.get('timestamp', 'N/A') event_type = event.get('type', 'UNKNOWN') data = event.get('data', {}) color = { 'COMPRAR': 'green', 'VENDER': 'red', 'AGUARDAR': 'orange' }.get(event_type, '#666') html += f"
" html += f"{timestamp} - {event_type}
" if 'action' in data: html += f"Ação: {data['action']}
" if 'confidence' in data: html += f"Confiança: {data['confidence']}%" html += "
" html += "
" return html def get_real_time_analysis_display(): """Obtém display da análise em tempo real.""" global real_time_integration, recent_events if not real_time_integration or not recent_events: return "
Aguardando dados do bot...
" # Pegar o evento mais recente latest_event = recent_events[-1] if recent_events else None if not latest_event: return "
Nenhum evento disponível
" # Processar o evento mais recente try: event_data = latest_event.get('data', {}) if 'raw_text' in event_data: analysis_result = main_analysis_function(event_data['raw_text']) return f"
{analysis_result}
" else: return "
Dados insuficientes para análise
" except Exception as e: return f"
Erro na análise: {str(e)}
" def get_system_stats(): """Obtém estatísticas do sistema.""" global recent_events, real_time_integration total_events = len(recent_events) monitor_status = "Ativo" if real_time_integration else "Inativo" # Contar tipos de eventos event_counts = {} for event in recent_events: event_type = event.get('type', 'UNKNOWN') event_counts[event_type] = event_counts.get(event_type, 0) + 1 html = f"""

📊 Estatísticas do Sistema

Status do Monitor: {monitor_status}

Total de Eventos: {total_events}

Modelo IA: {model_info.get('description', 'N/A')}

📈 Distribuição de Eventos:

" return html # Funções de controle de performance def start_performance_monitoring(): """Inicia o monitoramento de performance.""" global performance_monitor if not performance_monitor: return "
❌ Monitor de performance não disponível no modo standalone
" try: if not performance_monitor.monitoring: performance_monitor.start_monitoring(interval=5.0) return "
✅ Monitoramento de performance iniciado
" else: return "
⚠️ Monitoramento já está ativo
" except Exception as e: return f"
❌ Erro ao iniciar monitoramento: {str(e)}
" def stop_performance_monitoring(): """Para o monitoramento de performance.""" global performance_monitor if not performance_monitor: return "
❌ Monitor de performance não disponível no modo standalone
" try: if performance_monitor.monitoring: performance_monitor.stop_monitoring() return "
⏹️ Monitoramento de performance parado
" else: return "
Monitoramento já estava parado
" except Exception as e: return f"
❌ Erro ao parar monitoramento: {str(e)}
" def reset_performance_stats(): """Reseta as estatísticas de performance.""" global performance_monitor if not performance_monitor: return "
❌ Monitor de performance não disponível no modo standalone
" try: performance_monitor.reset_stats() return "
✅ Estatísticas resetadas com sucesso
" except Exception as e: return f"
❌ Erro ao resetar estatísticas: {str(e)}
" def export_performance_metrics(): """Exporta métricas de performance.""" global performance_monitor if not performance_monitor: return "
❌ Monitor de performance não disponível no modo standalone
" try: from datetime import datetime timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filepath = f"d:/hugging_face_spaces/performance_metrics_{timestamp}.json" performance_monitor.export_metrics(filepath, hours=24) return f"
✅ Métricas exportadas para: {filepath}
" except Exception as e: return f"
❌ Erro ao exportar métricas: {str(e)}
" def get_system_metrics_display(): """Obtém display das métricas do sistema.""" global performance_monitor if not performance_monitor: return "
Monitor de performance não disponível no modo standalone
" try: current_metrics = performance_monitor.get_current_metrics() if not current_metrics: return "
Nenhuma métrica disponível
" html = f"""

🖥️ Métricas Atuais

CPU: {current_metrics.cpu_usage:.1f}%
80 else 'orange' if current_metrics.cpu_usage > 60 else 'green'}; height: 100%; width: {current_metrics.cpu_usage}%; border-radius: 5px;'>
Memória: {current_metrics.memory_usage:.1f}%
85 else 'orange' if current_metrics.memory_usage > 70 else 'green'}; height: 100%; width: {current_metrics.memory_usage}%; border-radius: 5px;'>

Memória Disponível: {current_metrics.memory_available:.2f} GB

Uso do Disco: {current_metrics.disk_usage:.1f}%

Última atualização: {current_metrics.timestamp.strftime('%H:%M:%S')}

""" return html except Exception as e: return f"
Erro ao obter métricas: {str(e)}
" def get_bot_statistics_display(): """Obtém display das estatísticas do bot.""" global performance_monitor if not performance_monitor: return "
Monitor de performance não disponível no modo standalone
" try: bot_stats = performance_monitor.get_bot_stats() success_rate = 0 if bot_stats.total_analyses > 0: success_rate = (bot_stats.successful_analyses / bot_stats.total_analyses) * 100 html = f"""

🤖 Estatísticas do Bot

Total de Análises: {bot_stats.total_analyses}

Sucessos: {bot_stats.successful_analyses}

Falhas: {bot_stats.failed_analyses}

Taxa de Sucesso: {success_rate:.1f}%

Tempo Médio: {bot_stats.average_analysis_time:.2f}s

Alertas Fibonacci: {bot_stats.fibonacci_alerts_count}

Última Atualização:
{bot_stats.last_update.strftime('%H:%M:%S') if bot_stats.last_update else 'N/A'}

📊 Sinais Gerados:
""" for signal, count in bot_stats.signals_generated.items(): color = {'COMPRAR': 'green', 'VENDER': 'red', 'AGUARDAR': 'orange'}.get(signal, '#666') html += f"{signal}: {count}" html += "
" return html except Exception as e: return f"
Erro ao obter estatísticas: {str(e)}
" def get_performance_alerts_display(): """Obtém display dos alertas de performance.""" global performance_monitor if not performance_monitor: return "
Monitor de performance não disponível no modo standalone
" try: summary = performance_monitor.get_performance_summary() active_alerts = summary.get('active_alerts', []) if not active_alerts: return "
✅ Nenhum alerta ativo - Sistema operando normalmente
" html = "
" for alert in active_alerts: alert_text = { 'high_cpu': '🔥 CPU Alto', 'high_memory': '💾 Memória Alta', 'slow_analysis': '🐌 Análise Lenta', 'high_error_rate': '❌ Taxa de Erro Alta' }.get(alert, alert) html += f"
⚠️ {alert_text}
" html += "
" return html except Exception as e: return f"
Erro ao obter alertas: {str(e)}
" def get_optimization_suggestions_display(): """Obtém display das sugestões de otimização.""" global performance_monitor if not performance_monitor: return "
Monitor de performance não disponível no modo standalone
" try: suggestions = performance_monitor.optimize_performance() html = "
" for i, suggestion in enumerate(suggestions, 1): icon = "💡" if "normal" in suggestion.lower() else "⚠️" color = "#e8f5e8" if "normal" in suggestion.lower() else "#fff3cd" html += f"
{icon} {suggestion}
" html += "
" return html except Exception as e: return f"
Erro ao obter sugestões: {str(e)}
" # Inicializar engines e criar interface initialize_engines() demo = create_interface() if __name__ == "__main__": # Importar configurações específicas para HF Spaces try: from hf_spaces_config import HuggingFaceSpacesConfig # Aplicar correções para HF Spaces HuggingFaceSpacesConfig.apply_hf_spaces_fixes() # Usar configurações otimizadas para HF Spaces launch_config = HuggingFaceSpacesConfig.get_launch_config() demo.launch(**launch_config) except ImportError: # Fallback para configuração básica print("⚠️ Configurações HF Spaces não encontradas, usando configuração básica") demo.launch(ssr_mode=False)