feat: Implementa sistema ensemble avançado de IA com múltiplos modelos

- Adiciona EnsembleAI com FinBERT, DistilBERT, RoBERTa e BERT-Base
- Implementa sistema de votação inteligente com 6 estratégias
- Adiciona otimizador de performance com cache inteligente
- Integra processamento paralelo e métricas em tempo real
- Melhora precisão e velocidade das análises de sentimento

src/ai/ensemble_ai.py

@@ -0,0 +1,493 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Sistema Ensemble de IA Avançado para Análise de Sentimento Financeiro
+Combina múltiplos modelos para melhor precisão e confiabilidade
+"""
+
+import asyncio
+import logging
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass, asdict
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Any, Tuple
+from collections import defaultdict
+import numpy as np
+import torch
+from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
+import warnings
+warnings.filterwarnings('ignore')
+
+# Importar sistema de otimização
+try:
+    from .performance_optimizer import performance_optimizer, optimize_ai_analysis
+except ImportError:
+    # Fallback se não conseguir importar
+    performance_optimizer = None
+    
+    async def optimize_ai_analysis(func, text, use_cache=True):
+        return await func(text) if asyncio.iscoroutinefunction(func) else func(text)
+
+import json
+import hashlib
+from functools import lru_cache
+
+try:
+    from transformers import (
+        AutoTokenizer, AutoModelForSequenceClassification,
+        pipeline, BertTokenizer, BertForSequenceClassification
+    )
+    TRANSFORMERS_AVAILABLE = True
+except ImportError:
+    TRANSFORMERS_AVAILABLE = False
+    logging.warning("Transformers não disponível. Sistema ensemble funcionará em modo limitado.")
+
+try:
+    import torch
+    TORCH_AVAILABLE = True
+except ImportError:
+    TORCH_AVAILABLE = False
+
+@dataclass
+class ModelPrediction:
+    """Resultado de predição de um modelo individual"""
+    model_name: str
+    confidence: float
+    prediction: str
+    sentiment_score: float
+    processing_time: float
+    metadata: Dict[str, Any]
+
+@dataclass
+class EnsembleResult:
+    """Resultado final do ensemble"""
+    final_prediction: str
+    confidence: float
+    sentiment_score: float
+    individual_predictions: List[ModelPrediction]
+    consensus_strength: float
+    processing_time: float
+    model_weights: Dict[str, float]
+
+class ModelCache:
+    """Sistema de cache inteligente para otimizar performance"""
+    
+    def __init__(self, max_size: int = 1000):
+        self.cache = {}
+        self.max_size = max_size
+        self.access_count = {}
+        
+    def _generate_key(self, text: str, model_name: str) -> str:
+        """Gera chave única para cache"""
+        combined = f"{model_name}:{text}"
+        return hashlib.md5(combined.encode()).hexdigest()
+    
+    def get(self, text: str, model_name: str) -> Optional[ModelPrediction]:
+        """Recupera resultado do cache"""
+        key = self._generate_key(text, model_name)
+        if key in self.cache:
+            self.access_count[key] = self.access_count.get(key, 0) + 1
+            return self.cache[key]
+        return None
+    
+    def set(self, text: str, model_name: str, result: ModelPrediction):
+        """Armazena resultado no cache"""
+        if len(self.cache) >= self.max_size:
+            self._evict_least_used()
+        
+        key = self._generate_key(text, model_name)
+        self.cache[key] = result
+        self.access_count[key] = 1
+    
+    def _evict_least_used(self):
+        """Remove item menos usado do cache"""
+        if not self.cache:
+            return
+        
+        least_used_key = min(self.access_count.keys(), key=lambda k: self.access_count[k])
+        del self.cache[least_used_key]
+        del self.access_count[least_used_key]
+
+class AIModel:
+    """Classe base para modelos de IA"""
+    
+    def __init__(self, name: str, model_path: str, weight: float = 1.0):
+        self.name = name
+        self.model_path = model_path
+        self.weight = weight
+        self.model = None
+        self.tokenizer = None
+        self.is_loaded = False
+        
+    async def load_model(self):
+        """Carrega modelo de forma assíncrona"""
+        if not TRANSFORMERS_AVAILABLE:
+            logging.warning(f"Modelo {self.name} não pode ser carregado - Transformers indisponível")
+            return False
+            
+        try:
+            loop = asyncio.get_event_loop()
+            with ThreadPoolExecutor() as executor:
+                future = loop.run_in_executor(executor, self._load_model_sync)
+                await future
+            self.is_loaded = True
+            logging.info(f"Modelo {self.name} carregado com sucesso")
+            return True
+        except Exception as e:
+            logging.error(f"Erro ao carregar modelo {self.name}: {e}")
+            return False
+    
+    def _load_model_sync(self):
+        """Carregamento síncrono do modelo"""
+        self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
+        self.model = AutoModelForSequenceClassification.from_pretrained(self.model_path)
+    
+    async def predict(self, text: str) -> ModelPrediction:
+        """Faz predição com o modelo"""
+        start_time = datetime.now()
+        
+        if not self.is_loaded:
+            await self.load_model()
+        
+        if not self.is_loaded:
+            # Fallback para modelo mock
+            return self._mock_prediction(text, start_time)
+        
+        try:
+            loop = asyncio.get_event_loop()
+            with ThreadPoolExecutor() as executor:
+                future = loop.run_in_executor(executor, self._predict_sync, text)
+                result = await future
+            
+            processing_time = (datetime.now() - start_time).total_seconds()
+            
+            return ModelPrediction(
+                model_name=self.name,
+                confidence=result['confidence'],
+                prediction=result['prediction'],
+                sentiment_score=result['sentiment_score'],
+                processing_time=processing_time,
+                metadata=result.get('metadata', {})
+            )
+        except Exception as e:
+            logging.error(f"Erro na predição do modelo {self.name}: {e}")
+            return self._mock_prediction(text, start_time)
+    
+    def _predict_sync(self, text: str) -> Dict[str, Any]:
+        """Predição síncrona"""
+        inputs = self.tokenizer(text, return_tensors="pt", truncation=True, padding=True)
+        
+        with torch.no_grad() if TORCH_AVAILABLE else contextlib.nullcontext():
+            outputs = self.model(**inputs)
+            predictions = torch.nn.functional.softmax(outputs.logits, dim=-1)
+            
+        confidence = float(torch.max(predictions))
+        predicted_class = int(torch.argmax(predictions))
+        
+        # Mapear classe para sentimento
+        sentiment_map = {0: "NEGATIVO", 1: "NEUTRO", 2: "POSITIVO"}
+        prediction = sentiment_map.get(predicted_class, "NEUTRO")
+        
+        # Calcular score de sentimento (-1 a 1)
+        sentiment_score = (predicted_class - 1) * confidence
+        
+        return {
+            'confidence': confidence,
+            'prediction': prediction,
+            'sentiment_score': sentiment_score,
+            'metadata': {
+                'predicted_class': predicted_class,
+                'raw_predictions': predictions.tolist()
+            }
+        }
+    
+    def _mock_prediction(self, text: str, start_time: datetime) -> ModelPrediction:
+        """Predição mock para fallback"""
+        processing_time = (datetime.now() - start_time).total_seconds()
+        
+        # Análise simples baseada em palavras-chave
+        positive_words = ['alta', 'subida', 'compra', 'bull', 'positivo', 'ganho']
+        negative_words = ['baixa', 'queda', 'venda', 'bear', 'negativo', 'perda']
+        
+        text_lower = text.lower()
+        pos_count = sum(1 for word in positive_words if word in text_lower)
+        neg_count = sum(1 for word in negative_words if word in text_lower)
+        
+        if pos_count > neg_count:
+            prediction = "POSITIVO"
+            sentiment_score = 0.6
+            confidence = 0.7
+        elif neg_count > pos_count:
+            prediction = "NEGATIVO"
+            sentiment_score = -0.6
+            confidence = 0.7
+        else:
+            prediction = "NEUTRO"
+            sentiment_score = 0.0
+            confidence = 0.5
+        
+        return ModelPrediction(
+            model_name=f"{self.name}_mock",
+            confidence=confidence,
+            prediction=prediction,
+            sentiment_score=sentiment_score,
+            processing_time=processing_time,
+            metadata={'method': 'keyword_analysis'}
+        )
+
+class EnsembleAI:
+    """Sistema Ensemble de IA para Trading"""
+    
+    def __init__(self):
+        self.models: List[AIModel] = []
+        self.cache = ModelCache()
+        self.performance_history = {}
+        self.logger = logging.getLogger(__name__)
+        
+        # Inicializar modelos
+        self._initialize_models()
+    
+    def _initialize_models(self):
+        """Inicializa os modelos do ensemble"""
+        model_configs = [
+            {
+                'name': 'FinBERT',
+                'path': 'ProsusAI/finbert',
+                'weight': 1.2  # Peso maior para modelo especializado
+            },
+            {
+                'name': 'DistilBERT-Financial',
+                'path': 'distilbert-base-uncased',
+                'weight': 1.0
+            },
+            {
+                'name': 'RoBERTa-Sentiment',
+                'path': 'cardiffnlp/twitter-roberta-base-sentiment-latest',
+                'weight': 0.9
+            },
+            {
+                'name': 'BERT-Base',
+                'path': 'bert-base-uncased',
+                'weight': 0.8
+            }
+        ]
+        
+        for config in model_configs:
+            model = AIModel(
+                name=config['name'],
+                model_path=config['path'],
+                weight=config['weight']
+            )
+            self.models.append(model)
+        
+        self.logger.info(f"Inicializados {len(self.models)} modelos no ensemble")
+    
+    async def analyze_sentiment(self, text: str, use_cache: bool = True) -> EnsembleResult:
+        """Análise de sentimento usando ensemble de modelos com otimização"""
+        # Usar sistema de otimização se disponível
+        if performance_optimizer:
+            return await optimize_ai_analysis(
+                lambda t: self._analyze_sentiment_internal(t, use_cache),
+                text,
+                use_cache=use_cache
+            )
+        else:
+            return await self._analyze_sentiment_internal(text, use_cache)
+    
+    async def _analyze_sentiment_internal(self, text: str, use_cache: bool = True) -> EnsembleResult:
+        """Implementação interna da análise de sentimento"""
+        start_time = datetime.now()
+        
+        # Verificar cache primeiro
+        if use_cache:
+            cached_results = []
+            for model in self.models:
+                cached = self.cache.get(text, model.name)
+                if cached:
+                    cached_results.append(cached)
+            
+            if len(cached_results) == len(self.models):
+                self.logger.info("Resultado completo encontrado no cache")
+                return self._combine_predictions(cached_results, start_time)
+        
+        # Executar predições em paralelo
+        tasks = []
+        for model in self.models:
+            if use_cache:
+                cached = self.cache.get(text, model.name)
+                if cached:
+                    # Criar task que retorna resultado do cache
+                    tasks.append(asyncio.create_task(self._return_cached(cached)))
+                    continue
+            
+            tasks.append(asyncio.create_task(model.predict(text)))
+        
+        # Aguardar todas as predições
+        predictions = await asyncio.gather(*tasks, return_exceptions=True)
+        
+        # Filtrar exceções e armazenar no cache
+        valid_predictions = []
+        for i, pred in enumerate(predictions):
+            if isinstance(pred, Exception):
+                self.logger.error(f"Erro na predição do modelo {self.models[i].name}: {pred}")
+                continue
+            
+            valid_predictions.append(pred)
+            
+            # Armazenar no cache
+            if use_cache:
+                self.cache.set(text, pred.model_name, pred)
+        
+        if not valid_predictions:
+            self.logger.error("Nenhuma predição válida obtida")
+            return self._fallback_result(text, start_time)
+        
+        return self._combine_predictions(valid_predictions, start_time)
+    
+    async def _return_cached(self, cached_result: ModelPrediction) -> ModelPrediction:
+        """Retorna resultado do cache de forma assíncrona"""
+        return cached_result
+    
+    def _combine_predictions(self, predictions: List[ModelPrediction], start_time: datetime) -> EnsembleResult:
+        """Combina predições usando votação ponderada"""
+        if not predictions:
+            return self._fallback_result("", start_time)
+        
+        # Calcular pesos baseados na performance histórica
+        model_weights = self._calculate_dynamic_weights(predictions)
+        
+        # Votação ponderada para sentimento
+        sentiment_scores = []
+        confidences = []
+        
+        for pred in predictions:
+            weight = model_weights.get(pred.model_name, 1.0)
+            sentiment_scores.append(pred.sentiment_score * weight * pred.confidence)
+            confidences.append(pred.confidence * weight)
+        
+        # Calcular resultado final
+        weighted_sentiment = sum(sentiment_scores) / sum(confidences) if confidences else 0.0
+        final_confidence = np.mean(confidences) if confidences else 0.5
+        
+        # Determinar predição final
+        if weighted_sentiment > 0.1:
+            final_prediction = "POSITIVO"
+        elif weighted_sentiment < -0.1:
+            final_prediction = "NEGATIVO"
+        else:
+            final_prediction = "NEUTRO"
+        
+        # Calcular força do consenso
+        consensus_strength = self._calculate_consensus(predictions)
+        
+        processing_time = (datetime.now() - start_time).total_seconds()
+        
+        return EnsembleResult(
+            final_prediction=final_prediction,
+            confidence=final_confidence,
+            sentiment_score=weighted_sentiment,
+            individual_predictions=predictions,
+            consensus_strength=consensus_strength,
+            processing_time=processing_time,
+            model_weights=model_weights
+        )
+    
+    def _calculate_dynamic_weights(self, predictions: List[ModelPrediction]) -> Dict[str, float]:
+        """Calcula pesos dinâmicos baseados na performance histórica"""
+        weights = {}
+        
+        for pred in predictions:
+            base_weight = next((m.weight for m in self.models if m.name == pred.model_name), 1.0)
+            
+            # Ajustar peso baseado na performance histórica
+            historical_performance = self.performance_history.get(pred.model_name, 0.8)
+            
+            # Ajustar peso baseado na confiança atual
+            confidence_factor = pred.confidence
+            
+            # Peso final
+            final_weight = base_weight * historical_performance * confidence_factor
+            weights[pred.model_name] = final_weight
+        
+        return weights
+    
+    def _calculate_consensus(self, predictions: List[ModelPrediction]) -> float:
+        """Calcula força do consenso entre modelos"""
+        if len(predictions) < 2:
+            return 1.0
+        
+        # Contar predições por categoria
+        prediction_counts = {}
+        for pred in predictions:
+            prediction_counts[pred.prediction] = prediction_counts.get(pred.prediction, 0) + 1
+        
+        # Calcular consenso
+        max_count = max(prediction_counts.values())
+        consensus_strength = max_count / len(predictions)
+        
+        return consensus_strength
+    
+    def _fallback_result(self, text: str, start_time: datetime) -> EnsembleResult:
+        """Resultado de fallback quando todos os modelos falham"""
+        processing_time = (datetime.now() - start_time).total_seconds()
+        
+        return EnsembleResult(
+            final_prediction="NEUTRO",
+            confidence=0.3,
+            sentiment_score=0.0,
+            individual_predictions=[],
+            consensus_strength=0.0,
+            processing_time=processing_time,
+            model_weights={}
+        )
+    
+    def update_performance(self, model_name: str, accuracy: float):
+        """Atualiza performance histórica de um modelo"""
+        if model_name not in self.performance_history:
+            self.performance_history[model_name] = accuracy
+        else:
+            # Média móvel exponencial
+            alpha = 0.1
+            self.performance_history[model_name] = (
+                alpha * accuracy + (1 - alpha) * self.performance_history[model_name]
+            )
+    
+    def get_model_stats(self) -> Dict[str, Any]:
+        """Retorna estatísticas dos modelos"""
+        stats = {
+            'total_models': len(self.models),
+            'loaded_models': sum(1 for m in self.models if m.is_loaded),
+            'cache_size': len(self.cache.cache),
+            'performance_history': self.performance_history.copy(),
+            'model_weights': {m.name: m.weight for m in self.models}
+        }
+        return stats
+
+# Instância global do ensemble
+ensemble_ai = EnsembleAI()
+
+# Função de conveniência para uso externo
+async def analyze_market_sentiment(text: str, use_cache: bool = True) -> EnsembleResult:
+    """Função principal para análise de sentimento de mercado"""
+    return await ensemble_ai.analyze_sentiment(text, use_cache)
+
+if __name__ == "__main__":
+    # Teste do sistema
+    async def test_ensemble():
+        test_texts = [
+            "O mercado está em alta, com forte tendência de compra",
+            "Queda acentuada nos preços, momento de cautela",
+            "Mercado lateral, sem direção definida"
+        ]
+        
+        for text in test_texts:
+            print(f"\nAnalisando: {text}")
+            result = await analyze_market_sentiment(text)
+            print(f"Resultado: {result.final_prediction}")
+            print(f"Confiança: {result.confidence:.2f}")
+            print(f"Score: {result.sentiment_score:.2f}")
+            print(f"Consenso: {result.consensus_strength:.2f}")
+            print(f"Tempo: {result.processing_time:.3f}s")
+    
+    asyncio.run(test_ensemble())

src/ai/performance_optimizer.py

@@ -0,0 +1,632 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Sistema de Otimização de Performance para Ensemble AI
+Implementa cache inteligente, processamento paralelo e otimizações avançadas
+"""
+
+import asyncio
+import hashlib
+import json
+import logging
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass, asdict
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Any, Callable, Tuple
+from collections import defaultdict, deque
+import threading
+import weakref
+import pickle
+import os
+from pathlib import Path
+
+@dataclass
+class CacheEntry:
+    """Entrada do cache com metadados"""
+    key: str
+    value: Any
+    timestamp: datetime
+    access_count: int
+    last_access: datetime
+    ttl: Optional[timedelta]
+    size_bytes: int
+    hit_count: int = 0
+
+@dataclass
+class PerformanceMetrics:
+    """Métricas de performance do sistema"""
+    cache_hits: int = 0
+    cache_misses: int = 0
+    total_requests: int = 0
+    avg_response_time: float = 0.0
+    parallel_executions: int = 0
+    memory_usage_mb: float = 0.0
+    cpu_usage_percent: float = 0.0
+    active_threads: int = 0
+    queue_size: int = 0
+
+class IntelligentCache:
+    """Cache inteligente com estratégias adaptativas"""
+    
+    def __init__(self, max_size: int = 1000, default_ttl: timedelta = timedelta(hours=1)):
+        self.max_size = max_size
+        self.default_ttl = default_ttl
+        self.cache: Dict[str, CacheEntry] = {}
+        self.access_order = deque()  # Para LRU
+        self.size_tracker = 0
+        self.lock = threading.RLock()
+        
+        # Estatísticas
+        self.hits = 0
+        self.misses = 0
+        self.evictions = 0
+        
+        # Cache persistente
+        self.persistent_cache_dir = Path("cache/ai_cache")
+        self.persistent_cache_dir.mkdir(parents=True, exist_ok=True)
+        self.enable_persistence = True
+        
+        # Estratégias de eviction
+        self.eviction_strategies = {
+            'lru': self._evict_lru,
+            'lfu': self._evict_lfu,
+            'ttl': self._evict_expired,
+            'size': self._evict_largest,
+            'adaptive': self._evict_adaptive
+        }
+        self.current_strategy = 'adaptive'
+    
+    def get(self, key: str) -> Optional[Any]:
+        """Recupera item do cache"""
+        with self.lock:
+            cache_key = self._generate_key(key)
+            
+            if cache_key in self.cache:
+                entry = self.cache[cache_key]
+                
+                # Verificar TTL
+                if self._is_expired(entry):
+                    self._remove_entry(cache_key)
+                    self.misses += 1
+                    return None
+                
+                # Atualizar estatísticas de acesso
+                entry.access_count += 1
+                entry.hit_count += 1
+                entry.last_access = datetime.now()
+                
+                # Atualizar ordem LRU
+                if cache_key in self.access_order:
+                    self.access_order.remove(cache_key)
+                self.access_order.append(cache_key)
+                
+                self.hits += 1
+                return entry.value
+            
+            self.misses += 1
+            
+            # Tentar cache persistente
+            if self.enable_persistence:
+                persistent_value = self._load_from_persistent_cache(cache_key)
+                if persistent_value is not None:
+                    # Recarregar no cache em memória
+                    self.put(key, persistent_value)
+                    return persistent_value
+            
+            return None
+    
+    def put(self, key: str, value: Any, ttl: Optional[timedelta] = None) -> None:
+        """Armazena item no cache"""
+        with self.lock:
+            cache_key = self._generate_key(key)
+            
+            # Calcular tamanho
+            try:
+                size_bytes = len(pickle.dumps(value))
+            except:
+                size_bytes = 1024  # Estimativa padrão
+            
+            # Verificar se precisa fazer eviction
+            while len(self.cache) >= self.max_size or self.size_tracker + size_bytes > self.max_size * 10000:
+                if not self._evict_one():
+                    break  # Não conseguiu fazer eviction
+            
+            # Criar entrada
+            entry = CacheEntry(
+                key=cache_key,
+                value=value,
+                timestamp=datetime.now(),
+                access_count=1,
+                last_access=datetime.now(),
+                ttl=ttl or self.default_ttl,
+                size_bytes=size_bytes
+            )
+            
+            # Remover entrada existente se houver
+            if cache_key in self.cache:
+                old_entry = self.cache[cache_key]
+                self.size_tracker -= old_entry.size_bytes
+            
+            # Adicionar nova entrada
+            self.cache[cache_key] = entry
+            self.size_tracker += size_bytes
+            
+            # Atualizar ordem LRU
+            if cache_key in self.access_order:
+                self.access_order.remove(cache_key)
+            self.access_order.append(cache_key)
+            
+            # Salvar no cache persistente
+            if self.enable_persistence:
+                self._save_to_persistent_cache(cache_key, value)
+    
+    def _generate_key(self, key: str) -> str:
+        """Gera chave hash para o cache"""
+        return hashlib.md5(key.encode()).hexdigest()
+    
+    def _is_expired(self, entry: CacheEntry) -> bool:
+        """Verifica se entrada expirou"""
+        if entry.ttl is None:
+            return False
+        return datetime.now() - entry.timestamp > entry.ttl
+    
+    def _evict_one(self) -> bool:
+        """Remove uma entrada usando estratégia atual"""
+        strategy_func = self.eviction_strategies.get(self.current_strategy, self._evict_lru)
+        return strategy_func()
+    
+    def _evict_lru(self) -> bool:
+        """Remove entrada menos recentemente usada"""
+        if not self.access_order:
+            return False
+        
+        key_to_remove = self.access_order.popleft()
+        self._remove_entry(key_to_remove)
+        return True
+    
+    def _evict_lfu(self) -> bool:
+        """Remove entrada menos frequentemente usada"""
+        if not self.cache:
+            return False
+        
+        # Encontrar entrada com menor access_count
+        min_access_key = min(self.cache.keys(), key=lambda k: self.cache[k].access_count)
+        self._remove_entry(min_access_key)
+        return True
+    
+    def _evict_expired(self) -> bool:
+        """Remove entradas expiradas"""
+        expired_keys = [k for k, v in self.cache.items() if self._is_expired(v)]
+        
+        if not expired_keys:
+            return False
+        
+        for key in expired_keys:
+            self._remove_entry(key)
+        
+        return True
+    
+    def _evict_largest(self) -> bool:
+        """Remove entrada com maior tamanho"""
+        if not self.cache:
+            return False
+        
+        largest_key = max(self.cache.keys(), key=lambda k: self.cache[k].size_bytes)
+        self._remove_entry(largest_key)
+        return True
+    
+    def _evict_adaptive(self) -> bool:
+        """Estratégia adaptativa de eviction"""
+        # Primeiro tentar remover expirados
+        if self._evict_expired():
+            return True
+        
+        # Se cache está muito cheio, remover os maiores
+        if len(self.cache) > self.max_size * 0.9:
+            return self._evict_largest()
+        
+        # Caso contrário, usar LRU
+        return self._evict_lru()
+    
+    def _remove_entry(self, key: str) -> None:
+        """Remove entrada do cache"""
+        if key in self.cache:
+            entry = self.cache[key]
+            self.size_tracker -= entry.size_bytes
+            del self.cache[key]
+            self.evictions += 1
+        
+        if key in self.access_order:
+            self.access_order.remove(key)
+    
+    def _save_to_persistent_cache(self, key: str, value: Any) -> None:
+        """Salva no cache persistente"""
+        try:
+            cache_file = self.persistent_cache_dir / f"{key}.pkl"
+            with open(cache_file, 'wb') as f:
+                pickle.dump({
+                    'value': value,
+                    'timestamp': datetime.now(),
+                    'key': key
+                }, f)
+        except Exception as e:
+            logging.warning(f"Erro ao salvar cache persistente: {e}")
+    
+    def _load_from_persistent_cache(self, key: str) -> Optional[Any]:
+        """Carrega do cache persistente"""
+        try:
+            cache_file = self.persistent_cache_dir / f"{key}.pkl"
+            if cache_file.exists():
+                with open(cache_file, 'rb') as f:
+                    data = pickle.load(f)
+                
+                # Verificar se não expirou (24 horas)
+                if datetime.now() - data['timestamp'] < timedelta(hours=24):
+                    return data['value']
+                else:
+                    # Remover arquivo expirado
+                    cache_file.unlink()
+        except Exception as e:
+            logging.warning(f"Erro ao carregar cache persistente: {e}")
+        
+        return None
+    
+    def get_stats(self) -> Dict[str, Any]:
+        """Retorna estatísticas do cache"""
+        with self.lock:
+            total_requests = self.hits + self.misses
+            hit_rate = (self.hits / total_requests * 100) if total_requests > 0 else 0
+            
+            return {
+                'hits': self.hits,
+                'misses': self.misses,
+                'hit_rate': hit_rate,
+                'evictions': self.evictions,
+                'current_size': len(self.cache),
+                'max_size': self.max_size,
+                'memory_usage_bytes': self.size_tracker,
+                'strategy': self.current_strategy
+            }
+    
+    def clear(self) -> None:
+        """Limpa o cache"""
+        with self.lock:
+            self.cache.clear()
+            self.access_order.clear()
+            self.size_tracker = 0
+
+class ParallelProcessor:
+    """Processador paralelo para análises de IA"""
+    
+    def __init__(self, max_workers: int = 4):
+        self.max_workers = max_workers
+        self.executor = ThreadPoolExecutor(max_workers=max_workers)
+        self.active_tasks = set()
+        self.task_queue = asyncio.Queue()
+        self.metrics = PerformanceMetrics()
+        self.lock = threading.Lock()
+    
+    async def process_parallel(self, tasks: List[Callable], timeout: float = 30.0) -> List[Any]:
+        """Processa tarefas em paralelo"""
+        if not tasks:
+            return []
+        
+        start_time = time.time()
+        
+        # Submeter tarefas
+        futures = []
+        for task in tasks:
+            future = self.executor.submit(task)
+            futures.append(future)
+            
+            with self.lock:
+                self.active_tasks.add(future)
+                self.metrics.parallel_executions += 1
+        
+        # Aguardar resultados
+        results = []
+        try:
+            for future in as_completed(futures, timeout=timeout):
+                try:
+                    result = future.result()
+                    results.append(result)
+                except Exception as e:
+                    logging.error(f"Erro em tarefa paralela: {e}")
+                    results.append(None)
+                finally:
+                    with self.lock:
+                        self.active_tasks.discard(future)
+        
+        except TimeoutError:
+            logging.warning(f"Timeout em processamento paralelo após {timeout}s")
+            # Cancelar tarefas pendentes
+            for future in futures:
+                future.cancel()
+                with self.lock:
+                    self.active_tasks.discard(future)
+        
+        # Atualizar métricas
+        processing_time = time.time() - start_time
+        with self.lock:
+            self.metrics.avg_response_time = (
+                self.metrics.avg_response_time * 0.9 + processing_time * 0.1
+            )
+            self.metrics.active_threads = len(self.active_tasks)
+        
+        return results
+    
+    def get_metrics(self) -> PerformanceMetrics:
+        """Retorna métricas de performance"""
+        with self.lock:
+            return PerformanceMetrics(
+                cache_hits=self.metrics.cache_hits,
+                cache_misses=self.metrics.cache_misses,
+                total_requests=self.metrics.total_requests,
+                avg_response_time=self.metrics.avg_response_time,
+                parallel_executions=self.metrics.parallel_executions,
+                active_threads=len(self.active_tasks),
+                queue_size=self.task_queue.qsize() if hasattr(self.task_queue, 'qsize') else 0
+            )
+    
+    def shutdown(self):
+        """Encerra o processador"""
+        self.executor.shutdown(wait=True)
+
+class PerformanceOptimizer:
+    """Sistema principal de otimização de performance"""
+    
+    def __init__(self, cache_size: int = 1000, max_workers: int = 4):
+        self.cache = IntelligentCache(max_size=cache_size)
+        self.parallel_processor = ParallelProcessor(max_workers=max_workers)
+        self.metrics_history = deque(maxlen=1000)
+        self.optimization_rules = []
+        self.logger = logging.getLogger(__name__)
+        
+        # Configurações adaptativas
+        self.adaptive_config = {
+            'cache_ttl_base': timedelta(hours=1),
+            'parallel_threshold': 3,  # Número mínimo de tarefas para paralelizar
+            'timeout_base': 30.0,
+            'memory_threshold_mb': 500
+        }
+        
+        # Inicializar regras de otimização
+        self._initialize_optimization_rules()
+    
+    def _initialize_optimization_rules(self):
+        """Inicializa regras de otimização adaptativa"""
+        self.optimization_rules = [
+            self._rule_adjust_cache_ttl,
+            self._rule_adjust_parallel_threshold,
+            self._rule_memory_management,
+            self._rule_timeout_adjustment
+        ]
+    
+    async def optimize_analysis(self, analysis_func: Callable, 
+                              text: str, 
+                              use_cache: bool = True,
+                              force_parallel: bool = False) -> Any:
+        """Otimiza execução de análise com cache e paralelização"""
+        start_time = time.time()
+        
+        # Gerar chave de cache
+        cache_key = f"analysis_{hashlib.md5(text.encode()).hexdigest()}"
+        
+        # Tentar cache primeiro
+        if use_cache:
+            cached_result = self.cache.get(cache_key)
+            if cached_result is not None:
+                self.logger.debug(f"Cache hit para análise: {cache_key[:8]}...")
+                return cached_result
+        
+        # Executar análise
+        try:
+            if force_parallel or self._should_use_parallel():
+                # Análise paralela (se aplicável)
+                result = await self._execute_parallel_analysis(analysis_func, text)
+            else:
+                # Análise sequencial
+                result = await self._execute_sequential_analysis(analysis_func, text)
+            
+            # Armazenar no cache
+            if use_cache and result is not None:
+                ttl = self._calculate_adaptive_ttl(text, result)
+                self.cache.put(cache_key, result, ttl)
+            
+            # Registrar métricas
+            processing_time = time.time() - start_time
+            self._record_metrics(processing_time, use_cache, cached_result is not None)
+            
+            return result
+            
+        except Exception as e:
+            self.logger.error(f"Erro na análise otimizada: {e}")
+            raise
+    
+    async def _execute_sequential_analysis(self, analysis_func: Callable, text: str) -> Any:
+        """Executa análise sequencial"""
+        if asyncio.iscoroutinefunction(analysis_func):
+            return await analysis_func(text)
+        else:
+            return analysis_func(text)
+    
+    async def _execute_parallel_analysis(self, analysis_func: Callable, text: str) -> Any:
+        """Executa análise paralela (quando aplicável)"""
+        # Para análises que podem ser paralelizadas (ex: múltiplos modelos)
+        # Por enquanto, executa sequencialmente
+        return await self._execute_sequential_analysis(analysis_func, text)
+    
+    def _should_use_parallel(self) -> bool:
+        """Determina se deve usar processamento paralelo"""
+        # Lógica para decidir paralelização
+        current_load = len(self.parallel_processor.active_tasks)
+        return current_load < self.adaptive_config['parallel_threshold']
+    
+    def _calculate_adaptive_ttl(self, text: str, result: Any) -> timedelta:
+        """Calcula TTL adaptativo baseado no conteúdo"""
+        base_ttl = self.adaptive_config['cache_ttl_base']
+        
+        # Ajustar baseado no tamanho do texto
+        text_factor = min(2.0, len(text) / 1000)  # Textos maiores = TTL maior
+        
+        # Ajustar baseado na confiança do resultado
+        confidence_factor = 1.0
+        if hasattr(result, 'confidence'):
+            confidence_factor = result.confidence  # Alta confiança = TTL maior
+        
+        adjusted_ttl = base_ttl * text_factor * confidence_factor
+        return max(timedelta(minutes=5), min(timedelta(hours=6), adjusted_ttl))
+    
+    def _record_metrics(self, processing_time: float, used_cache: bool, cache_hit: bool):
+        """Registra métricas de performance"""
+        metrics = {
+            'timestamp': datetime.now(),
+            'processing_time': processing_time,
+            'used_cache': used_cache,
+            'cache_hit': cache_hit,
+            'memory_usage': self._get_memory_usage()
+        }
+        
+        self.metrics_history.append(metrics)
+        
+        # Aplicar regras de otimização
+        self._apply_optimization_rules()
+    
+    def _get_memory_usage(self) -> float:
+        """Estima uso de memória em MB"""
+        try:
+            import psutil
+            process = psutil.Process()
+            return process.memory_info().rss / 1024 / 1024
+        except ImportError:
+            return 0.0
+    
+    def _apply_optimization_rules(self):
+        """Aplica regras de otimização adaptativa"""
+        for rule in self.optimization_rules:
+            try:
+                rule()
+            except Exception as e:
+                self.logger.warning(f"Erro ao aplicar regra de otimização: {e}")
+    
+    def _rule_adjust_cache_ttl(self):
+        """Regra: Ajustar TTL do cache baseado na taxa de hit"""
+        if len(self.metrics_history) < 10:
+            return
+        
+        recent_metrics = list(self.metrics_history)[-10:]
+        hit_rate = sum(1 for m in recent_metrics if m['cache_hit']) / len(recent_metrics)
+        
+        if hit_rate > 0.8:  # Alta taxa de hit - aumentar TTL
+            self.adaptive_config['cache_ttl_base'] *= 1.1
+        elif hit_rate < 0.3:  # Baixa taxa de hit - diminuir TTL
+            self.adaptive_config['cache_ttl_base'] *= 0.9
+        
+        # Limitar TTL
+        self.adaptive_config['cache_ttl_base'] = max(
+            timedelta(minutes=10),
+            min(timedelta(hours=4), self.adaptive_config['cache_ttl_base'])
+        )
+    
+    def _rule_adjust_parallel_threshold(self):
+        """Regra: Ajustar threshold de paralelização"""
+        if len(self.metrics_history) < 20:
+            return
+        
+        recent_metrics = list(self.metrics_history)[-20:]
+        avg_processing_time = sum(m['processing_time'] for m in recent_metrics) / len(recent_metrics)
+        
+        if avg_processing_time > 5.0:  # Processamento lento - mais paralelização
+            self.adaptive_config['parallel_threshold'] = max(1, self.adaptive_config['parallel_threshold'] - 1)
+        elif avg_processing_time < 1.0:  # Processamento rápido - menos paralelização
+            self.adaptive_config['parallel_threshold'] = min(8, self.adaptive_config['parallel_threshold'] + 1)
+    
+    def _rule_memory_management(self):
+        """Regra: Gerenciar memória"""
+        current_memory = self._get_memory_usage()
+        
+        if current_memory > self.adaptive_config['memory_threshold_mb']:
+            # Limpar cache parcialmente
+            self.cache.clear()
+            self.logger.info(f"Cache limpo devido ao uso de memória: {current_memory:.1f}MB")
+    
+    def _rule_timeout_adjustment(self):
+        """Regra: Ajustar timeouts"""
+        if len(self.metrics_history) < 15:
+            return
+        
+        recent_metrics = list(self.metrics_history)[-15:]
+        avg_time = sum(m['processing_time'] for m in recent_metrics) / len(recent_metrics)
+        
+        # Ajustar timeout baseado no tempo médio
+        self.adaptive_config['timeout_base'] = max(10.0, min(60.0, avg_time * 3))
+    
+    def get_performance_report(self) -> Dict[str, Any]:
+        """Gera relatório completo de performance"""
+        cache_stats = self.cache.get_stats()
+        processor_metrics = self.parallel_processor.get_metrics()
+        
+        # Estatísticas históricas
+        if self.metrics_history:
+            recent_metrics = list(self.metrics_history)[-50:]
+            avg_processing_time = sum(m['processing_time'] for m in recent_metrics) / len(recent_metrics)
+            cache_hit_rate = sum(1 for m in recent_metrics if m['cache_hit']) / len(recent_metrics) * 100
+        else:
+            avg_processing_time = 0.0
+            cache_hit_rate = 0.0
+        
+        return {
+            'cache': cache_stats,
+            'parallel_processing': asdict(processor_metrics),
+            'adaptive_config': {
+                k: str(v) if isinstance(v, timedelta) else v 
+                for k, v in self.adaptive_config.items()
+            },
+            'performance_summary': {
+                'avg_processing_time': avg_processing_time,
+                'cache_hit_rate': cache_hit_rate,
+                'total_analyses': len(self.metrics_history),
+                'memory_usage_mb': self._get_memory_usage()
+            }
+        }
+    
+    def cleanup(self):
+        """Limpeza de recursos"""
+        self.parallel_processor.shutdown()
+        self.cache.clear()
+
+# Instância global do otimizador
+performance_optimizer = PerformanceOptimizer()
+
+# Função de conveniência
+async def optimize_ai_analysis(analysis_func: Callable, text: str, use_cache: bool = True) -> Any:
+    """Função principal para análise otimizada"""
+    return await performance_optimizer.optimize_analysis(analysis_func, text, use_cache)
+
+if __name__ == "__main__":
+    # Teste do sistema de otimização
+    async def test_analysis(text: str):
+        await asyncio.sleep(0.1)  # Simular processamento
+        return {'result': f'Análise de: {text[:20]}...', 'confidence': 0.8}
+    
+    async def test_optimizer():
+        print("Testando sistema de otimização...")
+        
+        # Teste de cache
+        result1 = await optimize_ai_analysis(test_analysis, "Texto de teste para análise")
+        result2 = await optimize_ai_analysis(test_analysis, "Texto de teste para análise")  # Deve usar cache
+        
+        print(f"Resultado 1: {result1}")
+        print(f"Resultado 2: {result2}")
+        
+        # Relatório de performance
+        report = performance_optimizer.get_performance_report()
+        print(f"\nRelatório de Performance:")
+        print(f"Cache Hit Rate: {report['performance_summary']['cache_hit_rate']:.1f}%")
+        print(f"Tempo Médio: {report['performance_summary']['avg_processing_time']:.3f}s")
+        
+        performance_optimizer.cleanup()
+    
+    # Executar teste
+    asyncio.run(test_optimizer())

src/ai/voting_system.py

@@ -0,0 +1,576 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Sistema de Votação Inteligente para Ensemble AI
+Otimiza decisões através de algoritmos avançados de consenso
+"""
+
+import numpy as np
+import logging
+from typing import Dict, List, Optional, Tuple, Any
+from dataclasses import dataclass
+from enum import Enum
+from datetime import datetime, timedelta
+import json
+from collections import defaultdict, deque
+import statistics
+
+class VotingStrategy(Enum):
+    """Estratégias de votação disponíveis"""
+    SIMPLE_MAJORITY = "simple_majority"
+    WEIGHTED_AVERAGE = "weighted_average"
+    CONFIDENCE_WEIGHTED = "confidence_weighted"
+    DYNAMIC_CONSENSUS = "dynamic_consensus"
+    BAYESIAN_FUSION = "bayesian_fusion"
+    ADAPTIVE_ENSEMBLE = "adaptive_ensemble"
+
+@dataclass
+class VoteResult:
+    """Resultado de uma votação"""
+    decision: str
+    confidence: float
+    consensus_strength: float
+    strategy_used: VotingStrategy
+    individual_votes: List[Dict[str, Any]]
+    metadata: Dict[str, Any]
+    processing_time: float
+
+@dataclass
+class ModelPerformance:
+    """Métricas de performance de um modelo"""
+    accuracy_history: deque
+    recent_accuracy: float
+    long_term_accuracy: float
+    consistency_score: float
+    response_time_avg: float
+    last_updated: datetime
+
+class AdaptiveWeightCalculator:
+    """Calculadora de pesos adaptativos para modelos"""
+    
+    def __init__(self, window_size: int = 100):
+        self.window_size = window_size
+        self.performance_tracker = defaultdict(lambda: ModelPerformance(
+            accuracy_history=deque(maxlen=window_size),
+            recent_accuracy=0.5,
+            long_term_accuracy=0.5,
+            consistency_score=0.5,
+            response_time_avg=1.0,
+            last_updated=datetime.now()
+        ))
+        self.market_conditions = {
+            'volatility': 0.5,
+            'trend_strength': 0.5,
+            'volume_profile': 0.5
+        }
+    
+    def update_performance(self, model_name: str, accuracy: float, response_time: float):
+        """Atualiza métricas de performance de um modelo"""
+        perf = self.performance_tracker[model_name]
+        
+        # Adicionar nova accuracy
+        perf.accuracy_history.append(accuracy)
+        
+        # Calcular métricas
+        if len(perf.accuracy_history) >= 10:
+            perf.recent_accuracy = np.mean(list(perf.accuracy_history)[-10:])
+        else:
+            perf.recent_accuracy = np.mean(list(perf.accuracy_history))
+        
+        perf.long_term_accuracy = np.mean(list(perf.accuracy_history))
+        
+        # Calcular consistência (inverso do desvio padrão)
+        if len(perf.accuracy_history) >= 5:
+            std_dev = np.std(list(perf.accuracy_history))
+            perf.consistency_score = max(0.1, 1.0 - std_dev)
+        
+        # Atualizar tempo de resposta médio
+        alpha = 0.1
+        perf.response_time_avg = alpha * response_time + (1 - alpha) * perf.response_time_avg
+        
+        perf.last_updated = datetime.now()
+    
+    def calculate_adaptive_weights(self, model_names: List[str], 
+                                 market_context: Optional[Dict[str, float]] = None) -> Dict[str, float]:
+        """Calcula pesos adaptativos baseados em performance e contexto"""
+        weights = {}
+        
+        # Atualizar condições de mercado se fornecidas
+        if market_context:
+            self.market_conditions.update(market_context)
+        
+        for model_name in model_names:
+            perf = self.performance_tracker[model_name]
+            
+            # Peso base da accuracy recente
+            accuracy_weight = perf.recent_accuracy
+            
+            # Ajuste por consistência
+            consistency_factor = perf.consistency_score
+            
+            # Ajuste por tempo de resposta (modelos mais rápidos têm vantagem)
+            speed_factor = min(2.0, 2.0 / max(0.1, perf.response_time_avg))
+            
+            # Ajuste por condições de mercado
+            market_factor = self._calculate_market_adjustment(model_name)
+            
+            # Peso final
+            final_weight = accuracy_weight * consistency_factor * speed_factor * market_factor
+            weights[model_name] = max(0.1, min(2.0, final_weight))  # Limitar entre 0.1 e 2.0
+        
+        # Normalizar pesos
+        total_weight = sum(weights.values())
+        if total_weight > 0:
+            weights = {k: v / total_weight for k, v in weights.items()}
+        
+        return weights
+    
+    def _calculate_market_adjustment(self, model_name: str) -> float:
+        """Calcula ajuste baseado nas condições de mercado"""
+        # Diferentes modelos podem ter performance melhor em diferentes condições
+        model_preferences = {
+            'FinBERT': {
+                'high_volatility': 1.2,
+                'strong_trend': 1.1,
+                'high_volume': 1.0
+            },
+            'DistilBERT-Financial': {
+                'high_volatility': 1.0,
+                'strong_trend': 1.2,
+                'high_volume': 1.1
+            },
+            'RoBERTa-Sentiment': {
+                'high_volatility': 0.9,
+                'strong_trend': 1.0,
+                'high_volume': 1.2
+            },
+            'BERT-Base': {
+                'high_volatility': 1.0,
+                'strong_trend': 1.0,
+                'high_volume': 1.0
+            }
+        }
+        
+        preferences = model_preferences.get(model_name, {
+            'high_volatility': 1.0,
+            'strong_trend': 1.0,
+            'high_volume': 1.0
+        })
+        
+        # Calcular fator de ajuste
+        volatility_factor = preferences['high_volatility'] if self.market_conditions['volatility'] > 0.7 else 1.0
+        trend_factor = preferences['strong_trend'] if self.market_conditions['trend_strength'] > 0.7 else 1.0
+        volume_factor = preferences['high_volume'] if self.market_conditions['volume_profile'] > 0.7 else 1.0
+        
+        return (volatility_factor + trend_factor + volume_factor) / 3.0
+
+class IntelligentVotingSystem:
+    """Sistema de votação inteligente com múltiplas estratégias"""
+    
+    def __init__(self):
+        self.weight_calculator = AdaptiveWeightCalculator()
+        self.voting_history = deque(maxlen=1000)
+        self.strategy_performance = defaultdict(lambda: deque(maxlen=100))
+        self.logger = logging.getLogger(__name__)
+        
+        # Configurações de estratégias
+        self.strategy_configs = {
+            VotingStrategy.SIMPLE_MAJORITY: {'threshold': 0.5},
+            VotingStrategy.WEIGHTED_AVERAGE: {'min_confidence': 0.3},
+            VotingStrategy.CONFIDENCE_WEIGHTED: {'confidence_power': 2.0},
+            VotingStrategy.DYNAMIC_CONSENSUS: {'consensus_threshold': 0.7},
+            VotingStrategy.BAYESIAN_FUSION: {'prior_strength': 0.1},
+            VotingStrategy.ADAPTIVE_ENSEMBLE: {'adaptation_rate': 0.1}
+        }
+    
+    def vote(self, predictions: List[Dict[str, Any]], 
+             strategy: VotingStrategy = VotingStrategy.ADAPTIVE_ENSEMBLE,
+             market_context: Optional[Dict[str, float]] = None) -> VoteResult:
+        """Executa votação usando estratégia especificada"""
+        start_time = datetime.now()
+        
+        if not predictions:
+            return self._empty_vote_result(strategy, start_time)
+        
+        # Selecionar estratégia automaticamente se for ADAPTIVE_ENSEMBLE
+        if strategy == VotingStrategy.ADAPTIVE_ENSEMBLE:
+            strategy = self._select_best_strategy(predictions, market_context)
+        
+        # Executar votação
+        result = self._execute_voting_strategy(predictions, strategy, market_context)
+        
+        # Calcular tempo de processamento
+        processing_time = (datetime.now() - start_time).total_seconds()
+        result.processing_time = processing_time
+        
+        # Armazenar no histórico
+        self.voting_history.append({
+            'timestamp': datetime.now(),
+            'strategy': strategy,
+            'result': result,
+            'num_predictions': len(predictions)
+        })
+        
+        return result
+    
+    def _select_best_strategy(self, predictions: List[Dict[str, Any]], 
+                            market_context: Optional[Dict[str, float]]) -> VotingStrategy:
+        """Seleciona a melhor estratégia baseada no contexto"""
+        # Analisar características das predições
+        confidences = [p.get('confidence', 0.5) for p in predictions]
+        avg_confidence = np.mean(confidences)
+        confidence_variance = np.var(confidences)
+        
+        # Analisar consenso
+        predictions_count = defaultdict(int)
+        for p in predictions:
+            predictions_count[p.get('prediction', 'NEUTRO')] += 1
+        
+        max_agreement = max(predictions_count.values()) / len(predictions)
+        
+        # Selecionar estratégia baseada nas características
+        if max_agreement > 0.8:  # Alto consenso
+            return VotingStrategy.SIMPLE_MAJORITY
+        elif avg_confidence > 0.8:  # Alta confiança
+            return VotingStrategy.CONFIDENCE_WEIGHTED
+        elif confidence_variance > 0.1:  # Alta variância na confiança
+            return VotingStrategy.WEIGHTED_AVERAGE
+        elif len(predictions) >= 4:  # Muitos modelos
+            return VotingStrategy.BAYESIAN_FUSION
+        else:
+            return VotingStrategy.DYNAMIC_CONSENSUS
+    
+    def _execute_voting_strategy(self, predictions: List[Dict[str, Any]], 
+                               strategy: VotingStrategy,
+                               market_context: Optional[Dict[str, float]]) -> VoteResult:
+        """Executa a estratégia de votação especificada"""
+        
+        if strategy == VotingStrategy.SIMPLE_MAJORITY:
+            return self._simple_majority_vote(predictions)
+        elif strategy == VotingStrategy.WEIGHTED_AVERAGE:
+            return self._weighted_average_vote(predictions, market_context)
+        elif strategy == VotingStrategy.CONFIDENCE_WEIGHTED:
+            return self._confidence_weighted_vote(predictions)
+        elif strategy == VotingStrategy.DYNAMIC_CONSENSUS:
+            return self._dynamic_consensus_vote(predictions)
+        elif strategy == VotingStrategy.BAYESIAN_FUSION:
+            return self._bayesian_fusion_vote(predictions)
+        else:
+            # Fallback para weighted average
+            return self._weighted_average_vote(predictions, market_context)
+    
+    def _simple_majority_vote(self, predictions: List[Dict[str, Any]]) -> VoteResult:
+        """Votação por maioria simples"""
+        vote_counts = defaultdict(int)
+        
+        for pred in predictions:
+            vote_counts[pred.get('prediction', 'NEUTRO')] += 1
+        
+        # Encontrar vencedor
+        winner = max(vote_counts.keys(), key=lambda k: vote_counts[k])
+        max_votes = vote_counts[winner]
+        
+        # Calcular confiança e consenso
+        confidence = max_votes / len(predictions)
+        consensus_strength = confidence
+        
+        return VoteResult(
+            decision=winner,
+            confidence=confidence,
+            consensus_strength=consensus_strength,
+            strategy_used=VotingStrategy.SIMPLE_MAJORITY,
+            individual_votes=[{'prediction': p.get('prediction'), 'confidence': p.get('confidence')} for p in predictions],
+            metadata={'vote_counts': dict(vote_counts)},
+            processing_time=0.0
+        )
+    
+    def _weighted_average_vote(self, predictions: List[Dict[str, Any]], 
+                             market_context: Optional[Dict[str, float]]) -> VoteResult:
+        """Votação por média ponderada"""
+        model_names = [p.get('model_name', f'model_{i}') for i, p in enumerate(predictions)]
+        weights = self.weight_calculator.calculate_adaptive_weights(model_names, market_context)
+        
+        # Calcular scores ponderados
+        sentiment_scores = []
+        total_weight = 0
+        
+        for i, pred in enumerate(predictions):
+            model_name = model_names[i]
+            weight = weights.get(model_name, 1.0)
+            confidence = pred.get('confidence', 0.5)
+            sentiment_score = pred.get('sentiment_score', 0.0)
+            
+            weighted_score = sentiment_score * weight * confidence
+            sentiment_scores.append(weighted_score)
+            total_weight += weight * confidence
+        
+        # Calcular resultado final
+        if total_weight > 0:
+            final_sentiment = sum(sentiment_scores) / total_weight
+        else:
+            final_sentiment = 0.0
+        
+        # Determinar decisão
+        if final_sentiment > 0.1:
+            decision = "POSITIVO"
+        elif final_sentiment < -0.1:
+            decision = "NEGATIVO"
+        else:
+            decision = "NEUTRO"
+        
+        # Calcular confiança média ponderada
+        weighted_confidences = [p.get('confidence', 0.5) * weights.get(model_names[i], 1.0) 
+                              for i, p in enumerate(predictions)]
+        confidence = sum(weighted_confidences) / sum(weights.values()) if weights else 0.5
+        
+        # Calcular consenso
+        consensus_strength = self._calculate_consensus_strength(predictions)
+        
+        return VoteResult(
+            decision=decision,
+            confidence=confidence,
+            consensus_strength=consensus_strength,
+            strategy_used=VotingStrategy.WEIGHTED_AVERAGE,
+            individual_votes=[{'prediction': p.get('prediction'), 'confidence': p.get('confidence'), 
+                             'weight': weights.get(model_names[i], 1.0)} for i, p in enumerate(predictions)],
+            metadata={'final_sentiment': final_sentiment, 'weights': weights},
+            processing_time=0.0
+        )
+    
+    def _confidence_weighted_vote(self, predictions: List[Dict[str, Any]]) -> VoteResult:
+        """Votação ponderada pela confiança"""
+        power = self.strategy_configs[VotingStrategy.CONFIDENCE_WEIGHTED]['confidence_power']
+        
+        # Calcular pesos baseados na confiança
+        weighted_votes = defaultdict(float)
+        total_weight = 0
+        
+        for pred in predictions:
+            confidence = pred.get('confidence', 0.5)
+            prediction = pred.get('prediction', 'NEUTRO')
+            weight = confidence ** power
+            
+            weighted_votes[prediction] += weight
+            total_weight += weight
+        
+        # Normalizar
+        if total_weight > 0:
+            weighted_votes = {k: v / total_weight for k, v in weighted_votes.items()}
+        
+        # Encontrar vencedor
+        winner = max(weighted_votes.keys(), key=lambda k: weighted_votes[k])
+        confidence = weighted_votes[winner]
+        
+        # Calcular consenso
+        consensus_strength = confidence
+        
+        return VoteResult(
+            decision=winner,
+            confidence=confidence,
+            consensus_strength=consensus_strength,
+            strategy_used=VotingStrategy.CONFIDENCE_WEIGHTED,
+            individual_votes=[{'prediction': p.get('prediction'), 'confidence': p.get('confidence')} for p in predictions],
+            metadata={'weighted_votes': dict(weighted_votes)},
+            processing_time=0.0
+        )
+    
+    def _dynamic_consensus_vote(self, predictions: List[Dict[str, Any]]) -> VoteResult:
+        """Votação por consenso dinâmico"""
+        threshold = self.strategy_configs[VotingStrategy.DYNAMIC_CONSENSUS]['consensus_threshold']
+        
+        # Agrupar por predição
+        groups = defaultdict(list)
+        for pred in predictions:
+            groups[pred.get('prediction', 'NEUTRO')].append(pred)
+        
+        # Encontrar grupo com maior consenso
+        best_group = None
+        best_consensus = 0
+        
+        for prediction, group in groups.items():
+            # Calcular consenso do grupo
+            confidences = [p.get('confidence', 0.5) for p in group]
+            group_size_factor = len(group) / len(predictions)
+            avg_confidence = np.mean(confidences)
+            consensus = group_size_factor * avg_confidence
+            
+            if consensus > best_consensus:
+                best_consensus = consensus
+                best_group = (prediction, group)
+        
+        if best_group and best_consensus >= threshold:
+            decision = best_group[0]
+            confidence = best_consensus
+        else:
+            # Fallback para neutro se não há consenso suficiente
+            decision = "NEUTRO"
+            confidence = 0.5
+        
+        return VoteResult(
+            decision=decision,
+            confidence=confidence,
+            consensus_strength=best_consensus,
+            strategy_used=VotingStrategy.DYNAMIC_CONSENSUS,
+            individual_votes=[{'prediction': p.get('prediction'), 'confidence': p.get('confidence')} for p in predictions],
+            metadata={'threshold': threshold, 'groups': {k: len(v) for k, v in groups.items()}},
+            processing_time=0.0
+        )
+    
+    def _bayesian_fusion_vote(self, predictions: List[Dict[str, Any]]) -> VoteResult:
+        """Votação usando fusão Bayesiana"""
+        prior_strength = self.strategy_configs[VotingStrategy.BAYESIAN_FUSION]['prior_strength']
+        
+        # Prior uniforme
+        classes = ['POSITIVO', 'NEUTRO', 'NEGATIVO']
+        prior = {cls: 1.0/len(classes) for cls in classes}
+        
+        # Calcular likelihood para cada classe
+        posteriors = prior.copy()
+        
+        for pred in predictions:
+            prediction = pred.get('prediction', 'NEUTRO')
+            confidence = pred.get('confidence', 0.5)
+            
+            # Atualizar posterior
+            for cls in classes:
+                if cls == prediction:
+                    likelihood = confidence
+                else:
+                    likelihood = (1 - confidence) / (len(classes) - 1)
+                
+                posteriors[cls] *= (prior_strength * prior[cls] + likelihood)
+        
+        # Normalizar
+        total = sum(posteriors.values())
+        if total > 0:
+            posteriors = {k: v / total for k, v in posteriors.items()}
+        
+        # Encontrar classe com maior probabilidade
+        winner = max(posteriors.keys(), key=lambda k: posteriors[k])
+        confidence = posteriors[winner]
+        
+        # Calcular consenso baseado na distribuição
+        entropy = -sum(p * np.log(p + 1e-10) for p in posteriors.values())
+        max_entropy = np.log(len(classes))
+        consensus_strength = 1 - (entropy / max_entropy)
+        
+        return VoteResult(
+            decision=winner,
+            confidence=confidence,
+            consensus_strength=consensus_strength,
+            strategy_used=VotingStrategy.BAYESIAN_FUSION,
+            individual_votes=[{'prediction': p.get('prediction'), 'confidence': p.get('confidence')} for p in predictions],
+            metadata={'posteriors': posteriors, 'entropy': entropy},
+            processing_time=0.0
+        )
+    
+    def _calculate_consensus_strength(self, predictions: List[Dict[str, Any]]) -> float:
+        """Calcula força do consenso entre predições"""
+        if not predictions:
+            return 0.0
+        
+        # Contar predições por classe
+        counts = defaultdict(int)
+        for pred in predictions:
+            counts[pred.get('prediction', 'NEUTRO')] += 1
+        
+        # Calcular consenso
+        max_count = max(counts.values())
+        consensus = max_count / len(predictions)
+        
+        return consensus
+    
+    def _empty_vote_result(self, strategy: VotingStrategy, start_time: datetime) -> VoteResult:
+        """Resultado para quando não há predições"""
+        return VoteResult(
+            decision="NEUTRO",
+            confidence=0.0,
+            consensus_strength=0.0,
+            strategy_used=strategy,
+            individual_votes=[],
+            metadata={'error': 'no_predictions'},
+            processing_time=(datetime.now() - start_time).total_seconds()
+        )
+    
+    def update_strategy_performance(self, strategy: VotingStrategy, accuracy: float):
+        """Atualiza performance de uma estratégia"""
+        self.strategy_performance[strategy].append(accuracy)
+    
+    def get_best_strategy(self) -> VotingStrategy:
+        """Retorna a estratégia com melhor performance recente"""
+        if not self.strategy_performance:
+            return VotingStrategy.ADAPTIVE_ENSEMBLE
+        
+        best_strategy = VotingStrategy.ADAPTIVE_ENSEMBLE
+        best_performance = 0.0
+        
+        for strategy, performances in self.strategy_performance.items():
+            if len(performances) >= 5:  # Mínimo de amostras
+                avg_performance = np.mean(list(performances)[-10:])  # Últimas 10
+                if avg_performance > best_performance:
+                    best_performance = avg_performance
+                    best_strategy = strategy
+        
+        return best_strategy
+    
+    def get_voting_stats(self) -> Dict[str, Any]:
+        """Retorna estatísticas do sistema de votação"""
+        stats = {
+            'total_votes': len(self.voting_history),
+            'strategy_usage': defaultdict(int),
+            'avg_processing_time': 0.0,
+            'avg_consensus_strength': 0.0,
+            'strategy_performance': {}
+        }
+        
+        if self.voting_history:
+            # Contar uso de estratégias
+            for vote in self.voting_history:
+                stats['strategy_usage'][vote['strategy'].value] += 1
+            
+            # Calcular médias
+            processing_times = [vote['result'].processing_time for vote in self.voting_history]
+            consensus_strengths = [vote['result'].consensus_strength for vote in self.voting_history]
+            
+            stats['avg_processing_time'] = np.mean(processing_times)
+            stats['avg_consensus_strength'] = np.mean(consensus_strengths)
+        
+        # Performance das estratégias
+        for strategy, performances in self.strategy_performance.items():
+            if performances:
+                stats['strategy_performance'][strategy.value] = {
+                    'avg_accuracy': np.mean(list(performances)),
+                    'recent_accuracy': np.mean(list(performances)[-10:]) if len(performances) >= 10 else np.mean(list(performances)),
+                    'sample_count': len(performances)
+                }
+        
+        return dict(stats)
+
+# Instância global do sistema de votação
+voting_system = IntelligentVotingSystem()
+
+# Função de conveniência
+def intelligent_vote(predictions: List[Dict[str, Any]], 
+                   strategy: VotingStrategy = VotingStrategy.ADAPTIVE_ENSEMBLE,
+                   market_context: Optional[Dict[str, float]] = None) -> VoteResult:
+    """Função principal para votação inteligente"""
+    return voting_system.vote(predictions, strategy, market_context)
+
+if __name__ == "__main__":
+    # Teste do sistema
+    test_predictions = [
+        {'model_name': 'FinBERT', 'prediction': 'POSITIVO', 'confidence': 0.8, 'sentiment_score': 0.6},
+        {'model_name': 'DistilBERT', 'prediction': 'POSITIVO', 'confidence': 0.7, 'sentiment_score': 0.4},
+        {'model_name': 'RoBERTa', 'prediction': 'NEUTRO', 'confidence': 0.6, 'sentiment_score': 0.1},
+        {'model_name': 'BERT', 'prediction': 'POSITIVO', 'confidence': 0.9, 'sentiment_score': 0.7}
+    ]
+    
+    print("Testando sistema de votação inteligente...")
+    
+    for strategy in VotingStrategy:
+        result = intelligent_vote(test_predictions, strategy)
+        print(f"\nEstratégia: {strategy.value}")
+        print(f"Decisão: {result.decision}")
+        print(f"Confiança: {result.confidence:.3f}")
+        print(f"Consenso: {result.consensus_strength:.3f}")
+        print(f"Tempo: {result.processing_time:.3f}s")

src/analysis/sentiment_analysis.py

@@ -1,12 +1,22 @@
-"""Módulo de análise de sentimento usando IA financeira."""
+"""Módulo de análise de sentimento usando IA financeira com sistema Ensemble."""
 
 import re
+import asyncio
 from typing import Dict, Optional, Any
 from dataclasses import dataclass
 
 from config.config import FINANCIAL_MODELS, AIConfig, AppConfig
 
-# Importações opcionais para IA
+# Importações do sistema Ensemble
+try:
+    from src.ai.ensemble_ai import ensemble_ai, EnsembleResult
+    from src.ai.voting_system import intelligent_vote, VotingStrategy
+    ENSEMBLE_AVAILABLE = True
+except ImportError:
+    ENSEMBLE_AVAILABLE = False
+    print("Sistema Ensemble não disponível, usando fallback...")
+
+# Importações opcionais para IA (fallback)
 try:
     from transformers import pipeline
     import torch
@@ -284,32 +294,66 @@ class SentimentScorer:
 
 
 class SentimentAnalysisEngine:
-    """Engine principal de análise de sentimento."""
+    """Engine principal de análise de sentimento com sistema Ensemble."""
     
     def __init__(self):
+        # Sistema Ensemble (preferido)
+        self.ensemble_available = ENSEMBLE_AVAILABLE
+        
+        # Sistema tradicional (fallback)
         self.model_manager = ModelManager()
         self.analyzer = SentimentAnalyzer(self.model_manager)
         self.scorer = SentimentScorer()
+        
+        # Configurações do ensemble
+        self.voting_strategy = VotingStrategy.ADAPTIVE_ENSEMBLE
+        self.use_ensemble = self.ensemble_available
     
     def analyze_text(self, text: str) -> Dict[str, Any]:
-        """Executa análise completa de sentimento."""
-        # Análise de sentimento
-        sentiment_result = self.analyzer.analyze(text)
+        """Executa análise completa de sentimento usando sistema Ensemble ou fallback."""
+        if not text:
+            return self._get_empty_result()
         
-        # Calcular pontuação
-        score = self.scorer.calculate_sentiment_score(sentiment_result)
+        # Usar sistema Ensemble se disponível
+        if self.use_ensemble and self.ensemble_available:
+            try:
+                return self._analyze_with_ensemble(text)
+            except Exception as e:
+                print(f"Erro no sistema Ensemble, usando fallback: {e}")
+                # Continuar com sistema tradicional
         
-        # Gerar descrição
+        # Sistema tradicional (fallback)
+        sentiment_result = self.analyzer.analyze(text)
+        score = self.scorer.calculate_sentiment_score(sentiment_result)
         description = self.scorer.get_sentiment_signal_description(sentiment_result)
         
         return {
             'result': sentiment_result,
             'score': score,
-            'description': description
+            'description': description,
+            'ensemble_used': False
         }
     
     def get_model_status(self) -> Dict[str, Any]:
-        """Retorna status do modelo atual."""
+        """Retorna status dos modelos de IA (Ensemble + Fallback)."""
+        if self.use_ensemble and self.ensemble_available:
+            # Status do sistema Ensemble
+            try:
+                ensemble_stats = ensemble_ai.get_performance_stats()
+                active_models = len([m for m in ensemble_ai.models if m.is_available])
+                
+                return {
+                    'available': True,
+                    'model_name': f'Ensemble AI ({active_models} modelos)',
+                    'description': f'Sistema Ensemble com {active_models} modelos ativos',
+                    'status': 'active',
+                    'ensemble_stats': ensemble_stats,
+                    'voting_strategy': self.voting_strategy.value
+                }
+            except Exception as e:
+                print(f"Erro ao obter status do Ensemble: {e}")
+        
+        # Status do sistema tradicional
         if self.model_manager.is_model_available():
             model_info = self.model_manager.get_model_info()
             return {
@@ -328,4 +372,78 @@ class SentimentAnalysisEngine:
     
     def is_available(self) -> bool:
         """Verifica se análise de IA está disponível."""
-        return self.model_manager.is_model_available()
+        return (self.use_ensemble and self.ensemble_available) or self.model_manager.is_model_available()
+    
+    def _get_empty_result(self) -> Dict[str, Any]:
+        """Retorna resultado vazio para texto inválido."""
+        from dataclasses import asdict
+        empty_result = SentimentResult(
+            sentiment='neutral',
+            confidence=0.5,
+            label='NEUTRO',
+            model_used='empty_input'
+        )
+        return {
+            'result': empty_result,
+            'score': 0,
+            'description': 'Texto vazio ou inválido',
+            'ensemble_used': False
+        }
+    
+    def _analyze_with_ensemble(self, text: str) -> Dict[str, Any]:
+        """Analisa texto usando sistema Ensemble."""
+        # Executar análise ensemble de forma síncrona
+        loop = None
+        try:
+            loop = asyncio.get_event_loop()
+        except RuntimeError:
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+        
+        if loop.is_running():
+            # Se já há um loop rodando, criar uma task
+            import concurrent.futures
+            with concurrent.futures.ThreadPoolExecutor() as executor:
+                future = executor.submit(asyncio.run, ensemble_ai.analyze_sentiment(text))
+                ensemble_result = future.result()
+        else:
+            # Executar diretamente
+            ensemble_result = loop.run_until_complete(ensemble_ai.analyze_sentiment(text))
+        
+        # Converter resultado do ensemble para formato compatível
+        sentiment_result = SentimentResult(
+            sentiment=ensemble_result.final_prediction.lower(),
+            confidence=ensemble_result.confidence,
+            label=ensemble_result.final_prediction,
+            model_used=f'Ensemble ({len(ensemble_result.individual_predictions)} modelos)'
+        )
+        
+        sentiment_score = self._convert_sentiment_to_score(ensemble_result.sentiment_score)
+        description = self.scorer.get_sentiment_signal_description(sentiment_result)
+        
+        return {
+            'result': sentiment_result,
+            'score': sentiment_score,
+            'description': description,
+            'ensemble_used': True,
+            'ensemble_details': {
+                'consensus_strength': ensemble_result.consensus_strength,
+                'processing_time': ensemble_result.processing_time,
+                'individual_predictions': ensemble_result.individual_predictions,
+                'model_weights': ensemble_result.model_weights
+            }
+        }
+    
+    def _convert_sentiment_to_score(self, sentiment_score: float) -> int:
+        """Converte score de sentimento (-1 a 1) para escala de pontos."""
+        # Converter de [-1, 1] para [0, 100]
+        normalized_score = (sentiment_score + 1) / 2
+        return int(normalized_score * 100)
+    
+    def set_voting_strategy(self, strategy):
+        """Define estratégia de votação do ensemble."""
+        self.voting_strategy = strategy
+    
+    def toggle_ensemble(self, use_ensemble: bool):
+        """Ativa/desativa uso do sistema Ensemble."""
+        self.use_ensemble = use_ensemble and self.ensemble_available