teste / app.py
torxyton's picture
feat: Adiciona compatibilidade completa com Hugging Face Spaces
0a38476
"""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("""
<div style="text-align: center; padding: 20px; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); border-radius: 10px; margin-bottom: 20px;">
<h1 style="color: white; margin: 0; font-size: 2.5em;">📈 Análise de Mercado IA</h1>
<p style="color: #f0f0f0; margin: 10px 0 0 0;">Sistema de análise técnica para scalping</p>
</div>
""")
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="<div style='color: #666;'>Monitor parado</div>"
)
gr.Markdown("### ⚡ Eventos Recentes")
recent_events_display = gr.HTML(
value="<div style='color: #666;'>Nenhum evento recente</div>"
)
with gr.Column(scale=2):
gr.Markdown("### 📊 Análise em Tempo Real")
real_time_analysis = gr.HTML(
value="<div style='text-align: center; color: #666; padding: 40px;'>Aguardando dados do bot...</div>"
)
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="<div style='color: #666;'>Monitoramento parado</div>"
)
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="<div style='color: #666;'>Nenhum alerta ativo</div>"
)
with gr.Column(scale=2):
gr.Markdown("### 📊 Métricas do Sistema")
system_metrics = gr.HTML(
value="<div style='text-align: center; color: #666; padding: 40px;'>Aguardando métricas...</div>"
)
gr.Markdown("### 🤖 Estatísticas do Bot")
bot_statistics = gr.HTML(
value="<div style='text-align: center; color: #666; padding: 40px;'>Aguardando estatísticas...</div>"
)
gr.Markdown("### 💡 Sugestões de Otimização")
optimization_suggestions = gr.HTML(
value="<div style='color: #666;'>Execute análises para obter sugestões</div>"
)
# 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 "<div style='color: green;'>✅ Monitor iniciado com sucesso</div>"
else:
return "<div style='color: orange;'>⚠️ Integração em tempo real não disponível</div>"
except Exception as e:
return f"<div style='color: red;'>❌ Erro ao iniciar monitor: {str(e)}</div>"
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 "<div style='color: orange;'>⏹️ Monitor parado</div>"
else:
return "<div style='color: #666;'>Monitor já estava parado</div>"
except Exception as e:
return f"<div style='color: red;'>❌ Erro ao parar monitor: {str(e)}</div>"
def get_recent_events_display():
"""Obtém display dos eventos recentes."""
global recent_events
if not recent_events:
return "<div style='color: #666;'>Nenhum evento recente</div>"
html = "<div style='max-height: 300px; overflow-y: auto;'>"
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"<div style='border-left: 3px solid {color}; padding: 5px; margin: 5px 0; background: #f9f9f9;'>"
html += f"<strong>{timestamp}</strong> - {event_type}<br>"
if 'action' in data:
html += f"Ação: {data['action']}<br>"
if 'confidence' in data:
html += f"Confiança: {data['confidence']}%"
html += "</div>"
html += "</div>"
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 "<div style='text-align: center; color: #666; padding: 40px;'>Aguardando dados do bot...</div>"
# Pegar o evento mais recente
latest_event = recent_events[-1] if recent_events else None
if not latest_event:
return "<div style='text-align: center; color: #666; padding: 40px;'>Nenhum evento disponível</div>"
# 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"<div style='border: 2px solid #4CAF50; padding: 15px; border-radius: 10px;'>{analysis_result}</div>"
else:
return "<div style='color: #666;'>Dados insuficientes para análise</div>"
except Exception as e:
return f"<div style='color: red;'>Erro na análise: {str(e)}</div>"
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"""
<div style='padding: 20px;'>
<h3>📊 Estatísticas do Sistema</h3>
<p><strong>Status do Monitor:</strong> {monitor_status}</p>
<p><strong>Total de Eventos:</strong> {total_events}</p>
<p><strong>Modelo IA:</strong> {model_info.get('description', 'N/A')}</p>
<h4>📈 Distribuição de Eventos:</h4>
<ul>
"""
for event_type, count in event_counts.items():
html += f"<li>{event_type}: {count}</li>"
html += "</ul></div>"
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 "<div style='color: red;'>❌ Monitor de performance não disponível no modo standalone</div>"
try:
if not performance_monitor.monitoring:
performance_monitor.start_monitoring(interval=5.0)
return "<div style='color: green;'>✅ Monitoramento de performance iniciado</div>"
else:
return "<div style='color: orange;'>⚠️ Monitoramento já está ativo</div>"
except Exception as e:
return f"<div style='color: red;'>❌ Erro ao iniciar monitoramento: {str(e)}</div>"
def stop_performance_monitoring():
"""Para o monitoramento de performance."""
global performance_monitor
if not performance_monitor:
return "<div style='color: red;'>❌ Monitor de performance não disponível no modo standalone</div>"
try:
if performance_monitor.monitoring:
performance_monitor.stop_monitoring()
return "<div style='color: orange;'>⏹️ Monitoramento de performance parado</div>"
else:
return "<div style='color: #666;'>Monitoramento já estava parado</div>"
except Exception as e:
return f"<div style='color: red;'>❌ Erro ao parar monitoramento: {str(e)}</div>"
def reset_performance_stats():
"""Reseta as estatísticas de performance."""
global performance_monitor
if not performance_monitor:
return "<div style='color: red;'>❌ Monitor de performance não disponível no modo standalone</div>"
try:
performance_monitor.reset_stats()
return "<div style='color: green;'>✅ Estatísticas resetadas com sucesso</div>"
except Exception as e:
return f"<div style='color: red;'>❌ Erro ao resetar estatísticas: {str(e)}</div>"
def export_performance_metrics():
"""Exporta métricas de performance."""
global performance_monitor
if not performance_monitor:
return "<div style='color: red;'>❌ Monitor de performance não disponível no modo standalone</div>"
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"<div style='color: green;'>✅ Métricas exportadas para: {filepath}</div>"
except Exception as e:
return f"<div style='color: red;'>❌ Erro ao exportar métricas: {str(e)}</div>"
def get_system_metrics_display():
"""Obtém display das métricas do sistema."""
global performance_monitor
if not performance_monitor:
return "<div style='color: #666;'>Monitor de performance não disponível no modo standalone</div>"
try:
current_metrics = performance_monitor.get_current_metrics()
if not current_metrics:
return "<div style='color: #666;'>Nenhuma métrica disponível</div>"
html = f"""
<div style='padding: 15px; border: 1px solid #ddd; border-radius: 8px;'>
<h4>🖥️ Métricas Atuais</h4>
<div style='display: grid; grid-template-columns: 1fr 1fr; gap: 10px;'>
<div>
<strong>CPU:</strong> {current_metrics.cpu_usage:.1f}%<br>
<div style='background: #f0f0f0; height: 10px; border-radius: 5px; margin: 5px 0;'>
<div style='background: {'red' if current_metrics.cpu_usage > 80 else 'orange' if current_metrics.cpu_usage > 60 else 'green'};
height: 100%; width: {current_metrics.cpu_usage}%; border-radius: 5px;'></div>
</div>
</div>
<div>
<strong>Memória:</strong> {current_metrics.memory_usage:.1f}%<br>
<div style='background: #f0f0f0; height: 10px; border-radius: 5px; margin: 5px 0;'>
<div style='background: {'red' if current_metrics.memory_usage > 85 else 'orange' if current_metrics.memory_usage > 70 else 'green'};
height: 100%; width: {current_metrics.memory_usage}%; border-radius: 5px;'></div>
</div>
</div>
</div>
<p><strong>Memória Disponível:</strong> {current_metrics.memory_available:.2f} GB</p>
<p><strong>Uso do Disco:</strong> {current_metrics.disk_usage:.1f}%</p>
<p><small>Última atualização: {current_metrics.timestamp.strftime('%H:%M:%S')}</small></p>
</div>
"""
return html
except Exception as e:
return f"<div style='color: red;'>Erro ao obter métricas: {str(e)}</div>"
def get_bot_statistics_display():
"""Obtém display das estatísticas do bot."""
global performance_monitor
if not performance_monitor:
return "<div style='color: #666;'>Monitor de performance não disponível no modo standalone</div>"
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"""
<div style='padding: 15px; border: 1px solid #ddd; border-radius: 8px;'>
<h4>🤖 Estatísticas do Bot</h4>
<div style='display: grid; grid-template-columns: 1fr 1fr; gap: 15px;'>
<div>
<p><strong>Total de Análises:</strong> {bot_stats.total_analyses}</p>
<p><strong>Sucessos:</strong> {bot_stats.successful_analyses}</p>
<p><strong>Falhas:</strong> {bot_stats.failed_analyses}</p>
<p><strong>Taxa de Sucesso:</strong> {success_rate:.1f}%</p>
</div>
<div>
<p><strong>Tempo Médio:</strong> {bot_stats.average_analysis_time:.2f}s</p>
<p><strong>Alertas Fibonacci:</strong> {bot_stats.fibonacci_alerts_count}</p>
<p><strong>Última Atualização:</strong><br>
<small>{bot_stats.last_update.strftime('%H:%M:%S') if bot_stats.last_update else 'N/A'}</small></p>
</div>
</div>
<h5>📊 Sinais Gerados:</h5>
<div style='display: flex; gap: 10px;'>
"""
for signal, count in bot_stats.signals_generated.items():
color = {'COMPRAR': 'green', 'VENDER': 'red', 'AGUARDAR': 'orange'}.get(signal, '#666')
html += f"<span style='background: {color}; color: white; padding: 5px 10px; border-radius: 15px; font-size: 12px;'>{signal}: {count}</span>"
html += "</div></div>"
return html
except Exception as e:
return f"<div style='color: red;'>Erro ao obter estatísticas: {str(e)}</div>"
def get_performance_alerts_display():
"""Obtém display dos alertas de performance."""
global performance_monitor
if not performance_monitor:
return "<div style='color: #666;'>Monitor de performance não disponível no modo standalone</div>"
try:
summary = performance_monitor.get_performance_summary()
active_alerts = summary.get('active_alerts', [])
if not active_alerts:
return "<div style='color: green;'>✅ Nenhum alerta ativo - Sistema operando normalmente</div>"
html = "<div style='padding: 10px;'>"
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"<div style='background: #ffebee; border-left: 4px solid #f44336; padding: 10px; margin: 5px 0;'>⚠️ {alert_text}</div>"
html += "</div>"
return html
except Exception as e:
return f"<div style='color: red;'>Erro ao obter alertas: {str(e)}</div>"
def get_optimization_suggestions_display():
"""Obtém display das sugestões de otimização."""
global performance_monitor
if not performance_monitor:
return "<div style='color: #666;'>Monitor de performance não disponível no modo standalone</div>"
try:
suggestions = performance_monitor.optimize_performance()
html = "<div style='padding: 10px;'>"
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"<div style='background: {color}; padding: 10px; margin: 5px 0; border-radius: 5px;'>{icon} {suggestion}</div>"
html += "</div>"
return html
except Exception as e:
return f"<div style='color: red;'>Erro ao obter sugestões: {str(e)}</div>"
# 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)