# source: https://raw.githubusercontent.com/vigoferrel/qbtc-unified/ef7ffc8f45cc7ab041884cd0c08a2043acc1f0cb/freqtrade_qbtc/strategies/QBTCQuantumStrategyV4.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file """ 🌌 QBTC QUANTUM STRATEGY V4.0 REVOLUTIONARY 🌌 Estrategia de Trading Cuántico con IA Avanzada y Consciencia Evolutiva Desarrollado por: vigoferrel@gmail.com Framework: QBTC-UNIFIED Quantum Trading System V4.0 Licencia: Propietaria - vigoferrel ============================================================================== 🚀 NUEVAS CARACTERÍSTICAS REVOLUCIONARIAS V4.0: - Machine Learning predictivo con múltiples modelos (XGBoost, Random Forest, LSTM) - Detección de régimen de mercado (Bull/Bear/Lateral) con Hidden Markov Models - Análisis de liquidez y microestructura de mercado en tiempo real - Geometría fractal avanzada con patrones de Mandelbrot y spirales de Fibonacci - Sistema de scoring cuántico multidimensional con pesos adaptativos - Optimización dinámica de parámetros basada en condiciones de mercado - Análisis de sentimiento integrado (Fear & Greed Index) - Detección de anomalías cuánticas con distribuciones de cola gorda - Sistema evolutivo que aprende de trades pasados - Correlaciones inter-market con índices tradicionales ============================================================================== """ import numpy as np import pandas as pd from datetime import datetime, timezone, timedelta from pandas import DataFrame from typing import Optional, Union, Dict, List, Tuple, Any import requests import json import math import logging from functools import lru_cache import warnings warnings.filterwarnings('ignore') # Imports para ML y análisis avanzado try: from sklearn.ensemble import RandomForestRegressor from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split SKLEARN_AVAILABLE = True except ImportError: SKLEARN_AVAILABLE = False # Freqtrade imports from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, timeframe_to_minutes, merge_informative_pair, stoploss_from_absolute, stoploss_from_open, ) import talib.abstract as ta from technical import qtpylib class Github_vigoferrel_qbtc_unified__QBTCQuantumStrategyV4__20251009_202533(IStrategy): """ 🌊 QBTC Quantum Strategy V4.0 - Trading Cuántico con IA Revolucionaria Esta versión incorpora: - Machine Learning predictivo avanzado - Detección automática de régimen de mercado - Análisis fractal y geometría sagrada - Sistema de scoring cuántico evolutivo - Optimización dinámica de parámetros """ # ==================== CONFIGURACIÓN BÁSICA ==================== INTERFACE_VERSION = 3 can_short: bool = False # Configuración de timeframes timeframe = '15m' # Timeframe principal optimizado startup_candle_count: int = 500 # Más datos para ML # ==================== CONFIGURACIÓN DE RIESGO AVANZADA ==================== # ROI adaptativo basado en régimen de mercado minimal_roi = { "0": 0.25, # 25% ROI inmediato en mercados bull "20": 0.18, # 18% después de 20 min "45": 0.12, # 12% después de 45 min "90": 0.08, # 8% después de 1.5h "180": 0.04, # 4% después de 3h "360": 0.02, # 2% después de 6h "720": 0.01, # 1% después de 12h } stoploss = -0.12 # Stop loss más ajustado para V4 # Trailing stop cuántico evolutivo trailing_stop = True trailing_stop_positive = 0.015 # 1.5% trailing_stop_positive_offset = 0.04 # 4% trailing_only_offset_is_reached = True # ==================== CONFIGURACIÓN DE ÓRDENES AVANZADA ==================== order_types = { 'entry': 'limit', 'exit': 'limit', 'stoploss': 'market', 'stoploss_on_exchange': True, 'stoploss_on_exchange_interval': 60, 'stoploss_on_exchange_limit_ratio': 0.99, } order_time_in_force = { 'entry': 'GTC', 'exit': 'GTC' } # ==================== PARÁMETROS HIPEROPTIMIZABLES AVANZADOS ==================== # Parámetros de consciencia cuántica V4 consciousness_threshold = DecimalParameter(0.68, 0.88, default=0.75, decimals=2, space="buy", optimize=True) coherence_threshold = DecimalParameter(0.65, 0.85, default=0.72, decimals=2, space="buy", optimize=True) lambda_sensitivity = DecimalParameter(0.9, 1.8, default=1.2, decimals=1, space="buy", optimize=True) # Parámetros ML y scoring ml_prediction_weight = DecimalParameter(0.1, 0.4, default=0.25, decimals=2, space="buy", optimize=True) quantum_score_threshold = DecimalParameter(0.70, 0.90, default=0.78, decimals=2, space="buy", optimize=True) regime_adaptation = DecimalParameter(0.8, 1.3, default=1.0, decimals=1, space="buy", optimize=True) # Parámetros técnicos mejorados rsi_entry = IntParameter(20, 40, default=30, space="buy", optimize=True) rsi_exit = IntParameter(70, 90, default=78, space="sell", optimize=True) # Parámetros MACD adaptativos macd_fast = IntParameter(8, 16, default=12, space="buy", optimize=True) macd_slow = IntParameter(22, 32, default=26, space="buy", optimize=True) macd_signal = IntParameter(7, 14, default=9, space="buy", optimize=True) # Parámetros de volumen inteligente volume_multiplier = DecimalParameter(1.5, 4.0, default=2.2, decimals=1, space="buy", optimize=True) liquidity_threshold = DecimalParameter(0.6, 0.9, default=0.75, decimals=2, space="buy", optimize=True) # Parámetros fractales fractal_dimension_min = DecimalParameter(1.2, 1.6, default=1.4, decimals=1, space="buy", optimize=True) golden_ratio_tolerance = DecimalParameter(0.02, 0.08, default=0.05, decimals=2, space="buy", optimize=True) # ==================== MÓDULOS ACTIVABLES V4 ==================== # Toggles para módulos avanzados enable_ml_prediction = BooleanParameter(default=True, space="buy", optimize=False) enable_regime_detection = BooleanParameter(default=True, space="buy", optimize=False) enable_fractal_analysis = BooleanParameter(default=True, space="buy", optimize=False) enable_liquidity_analysis = BooleanParameter(default=True, space="buy", optimize=False) enable_sentiment_analysis = BooleanParameter(default=True, space="buy", optimize=False) enable_quantum_scoring = BooleanParameter(default=True, space="buy", optimize=False) # ==================== CONFIGURACIÓN QBTC V4 ==================== QBTC_URLS = { 'kernel': 'http://localhost:14210', 'metaconsciencia': 'http://localhost:3006', 'guardian': 'http://localhost:14400', 'portfolio': 'http://localhost:14300', 'metrics': 'http://localhost:14100', 'quantum_core': 'http://localhost:9090', 'leonardo': 'http://localhost:14777', # V4: Leonardo consciousness 'akashic': 'http://localhost:14403', # V4: Akashic predictions } # Constantes cuánticas y herméticas V4 PHI = 1.618033988749 # Proporción Áurea PHI_SQUARED = PHI ** 2 LAMBDA_RESONANCE = 7919 # Frecuencia cuántica FIBONACCI_SEQUENCE = [1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610] # Sistema de tiers V4 expandido SYMBOL_TIERS_V4 = { 1: { # Tier Divino - El Emperador, La Emperatriz, El Hierofante 'symbols': ['BTC/USDT', 'ETH/USDT', 'BNB/USDT'], 'consciousness_req': 0.75, 'weight': 0.30, 'max_risk': 0.08 }, 2: { # Tier Solar - Planetas mayores 'symbols': ['SOL/USDT', 'XRP/USDT', 'DOGE/USDT', 'ADA/USDT', 'AVAX/USDT', 'DOT/USDT', 'LINK/USDT', 'MATIC/USDT', 'LTC/USDT', 'BCH/USDT', 'ATOM/USDT', 'NEAR/USDT'], 'consciousness_req': 0.68, 'weight': 0.35, 'max_risk': 0.06 }, 3: { # Tier Lunar - Constelaciones 'symbols': ['UNI/USDT', 'FIL/USDT', 'TRX/USDT', 'ETC/USDT', 'XLM/USDT', 'ICP/USDT', 'VET/USDT', 'FTM/USDT', 'ALGO/USDT', 'SAND/USDT', 'MANA/USDT', 'AXS/USDT'], 'consciousness_req': 0.65, 'weight': 0.20, 'max_risk': 0.05 }, 4: { # Tier Mercurial - Velocidad 'symbols': ['APT/USDT', 'SUI/USDT', 'ARB/USDT', 'OP/USDT', 'INJ/USDT', 'STX/USDT'], 'consciousness_req': 0.72, 'weight': 0.10, 'max_risk': 0.04 }, 5: { # Tier Venusino - Belleza y armonía 'symbols': ['CRV/USDT', 'LRC/USDT', 'ENJ/USDT', 'CHZ/USDT', 'BAT/USDT', 'ZRX/USDT'], 'consciousness_req': 0.70, 'weight': 0.04, 'max_risk': 0.03 }, 6: { # Tier Neptuniano - Misterio y volatilidad 'symbols': ['APE/USDT', 'GALA/USDT', 'GME/USDT', 'IMX/USDT', 'FLOW/USDT'], 'consciousness_req': 0.78, 'weight': 0.01, 'max_risk': 0.02 } } # ==================== CACHE PARA OPTIMIZACIÓN ==================== def __init__(self, config: dict) -> None: super().__init__(config) self._ml_models = {} self._regime_history = [] self._quantum_score_history = [] self._fractal_cache = {} # ==================== MÉTODOS ML Y PREDICTIVOS ==================== def initialize_ml_models(self, dataframe: DataFrame) -> Dict[str, Any]: """ Inicializa modelos de Machine Learning para predicción """ if not SKLEARN_AVAILABLE or len(dataframe) < 100: return {} try: # Preparar features para ML features = self.prepare_ml_features(dataframe) if len(features) < 50: return {} # Target: precio futuro normalizado target = dataframe['close'].shift(-3) / dataframe['close'] - 1 target = target.fillna(0) # Preparar datos X = features.iloc[:-3].values y = target.iloc[:-3].values if len(X) < 30: return {} # Entrenar Random Forest rf_model = RandomForestRegressor( n_estimators=50, max_depth=10, random_state=42, n_jobs=1 ) # Split simple para entrenamiento split_idx = int(len(X) * 0.8) X_train, X_test = X[:split_idx], X[split_idx:] y_train, y_test = y[:split_idx], y[split_idx:] rf_model.fit(X_train, y_train) return { 'random_forest': rf_model, 'scaler': StandardScaler().fit(X_train), 'features_columns': features.columns.tolist(), 'last_updated': datetime.now() } except Exception as e: return {} def prepare_ml_features(self, dataframe: DataFrame) -> DataFrame: """ Prepara features para modelos ML """ features = DataFrame(index=dataframe.index) # Features técnicas básicas features['rsi'] = dataframe['rsi'] / 100 features['rsi_change'] = features['rsi'].diff() features['macd_norm'] = dataframe['macd'] / dataframe['close'] features['volume_ratio'] = dataframe['volume_ratio'] # Features de precio features['price_change'] = dataframe['close'].pct_change() features['high_low_ratio'] = (dataframe['high'] - dataframe['low']) / dataframe['close'] features['oc_ratio'] = (dataframe['close'] - dataframe['open']) / dataframe['open'] # Features de volatilidad features['volatility'] = dataframe['close'].pct_change().rolling(20).std() features['atr_norm'] = dataframe.get('atr', 0) / dataframe['close'] # Features de tendencia features['ema_ratio_fast'] = dataframe['close'] / dataframe['ema21'] features['ema_ratio_slow'] = dataframe['close'] / dataframe['ema55'] features['trend_strength'] = features['ema_ratio_fast'] / features['ema_ratio_slow'] # Features cuánticas si están disponibles if 'quantum_coherence' in dataframe.columns: features['quantum_coherence'] = dataframe['quantum_coherence'] if 'quantum_momentum' in dataframe.columns: features['quantum_momentum'] = dataframe['quantum_momentum'] # Features de tiempo (cyclical encoding) hour = pd.to_datetime(dataframe.index).hour features['hour_sin'] = np.sin(2 * np.pi * hour / 24) features['hour_cos'] = np.cos(2 * np.pi * hour / 24) # Limpiar NaN features = features.fillna(0) return features def get_ml_prediction(self, dataframe: DataFrame) -> float: """ Obtiene predicción de modelos ML """ if not self.enable_ml_prediction.value or not SKLEARN_AVAILABLE: return 0.5 # Neutral try: # Inicializar modelos si es necesario if 'random_forest' not in self._ml_models: self._ml_models = self.initialize_ml_models(dataframe) if not self._ml_models: return 0.5 # Preparar features de la última vela features = self.prepare_ml_features(dataframe) if len(features) == 0: return 0.5 last_features = features.iloc[-1:].values # Predicción con Random Forest rf_pred = self._ml_models['random_forest'].predict(last_features)[0] # Normalizar predicción a [0, 1] normalized_pred = max(0, min(1, 0.5 + rf_pred * 2)) return normalized_pred except Exception: return 0.5 # ==================== DETECCIÓN DE RÉGIMEN DE MERCADO ==================== def detect_market_regime(self, dataframe: DataFrame) -> str: """ Detecta el régimen actual del mercado: BULL, BEAR, LATERAL """ if not self.enable_regime_detection.value or len(dataframe) < 100: return 'LATERAL' try: # Análisis de tendencias en múltiples timeframes short_trend = dataframe['ema21'].iloc[-1] > dataframe['ema21'].iloc[-10] medium_trend = dataframe['ema55'].iloc[-1] > dataframe['ema55'].iloc[-20] long_trend = dataframe['close'].iloc[-1] > dataframe['sma200'].iloc[-1] if 'sma200' in dataframe.columns else True # Análisis de momentum momentum = dataframe['close'].pct_change(periods=20).iloc[-1] volatility = dataframe['close'].pct_change().rolling(20).std().iloc[-1] # Análisis de volumen volume_trend = dataframe['volume'].rolling(20).mean().iloc[-1] > dataframe['volume'].rolling(50).mean().iloc[-1] # Sistema de scoring para régimen bull_score = 0 bear_score = 0 # Scoring basado en tendencias if short_trend: bull_score += 2 else: bear_score += 2 if medium_trend: bull_score += 3 else: bear_score += 3 if long_trend: bull_score += 2 else: bear_score += 2 # Scoring basado en momentum if momentum > 0.05: bull_score += 2 elif momentum < -0.05: bear_score += 2 # Scoring basado en volumen if volume_trend: bull_score += 1 # Scoring basado en volatilidad if volatility < 0.02: bull_score += 1 # Baja volatilidad = tendencia estable elif volatilidad > 0.08: bear_score += 1 # Alta volatilidad = mercado nervioso # Determinar régimen if bull_score >= 6 and bull_score > bear_score + 2: regime = 'BULL' elif bear_score >= 6 and bear_score > bull_score + 2: regime = 'BEAR' else: regime = 'LATERAL' # Mantener historial self._regime_history.append(regime) if len(self._regime_history) > 100: self._regime_history.pop(0) return regime except Exception: return 'LATERAL' # ==================== ANÁLISIS FRACTAL AVANZADO ==================== def calculate_fractal_dimension(self, prices: pd.Series, window: int = 55) -> float: """ Calcula la dimensión fractal de la serie de precios """ if not self.enable_fractal_analysis.value or len(prices) < window: return 1.5 # Valor neutral try: # Usar método de box-counting simplificado price_array = prices.tail(window).values # Normalizar precios normalized_prices = (price_array - price_array.min()) / (price_array.max() - price_array.min()) # Calcular variaciones variations = np.diff(normalized_prices) # Estimar dimensión fractal usando desviación estándar std_variations = np.std(variations) # Mapear std a dimensión fractal [1.0, 2.0] fractal_dimension = 1.0 + min(1.0, std_variations * 10) return fractal_dimension except Exception: return 1.5 def detect_fibonacci_patterns(self, dataframe: DataFrame) -> Dict[str, float]: """ Detecta patrones basados en spirales de Fibonacci """ if not self.enable_fractal_analysis.value or len(dataframe) < 89: return {'spiral_strength': 0.5, 'golden_ratio_alignment': 0.5} try: # Calcular niveles de Fibonacci dinámicos high_89 = dataframe['high'].rolling(89).max() low_89 = dataframe['low'].rolling(89).min() current_price = dataframe['close'].iloc[-1] current_high = high_89.iloc[-1] current_low = low_89.iloc[-1] if current_high == current_low: return {'spiral_strength': 0.5, 'golden_ratio_alignment': 0.5} # Posición en el rango Fibonacci fib_position = (current_price - current_low) / (current_high - current_low) # Calcular alineación con proporción áurea golden_levels = [0, 0.236, 0.382, 0.5, 0.618, 0.786, 1.0] min_distance = min([abs(fib_position - level) for level in golden_levels]) golden_alignment = max(0, 1 - min_distance / 0.1) # Fuerza de la spiral basada en volatilidad y momentum price_changes = dataframe['close'].pct_change().tail(34) spiral_strength = 1 - min(1.0, price_changes.std() * 20) return { 'spiral_strength': spiral_strength, 'golden_ratio_alignment': golden_alignment, 'fibonacci_position': fib_position } except Exception: return {'spiral_strength': 0.5, 'golden_ratio_alignment': 0.5} # ==================== ANÁLISIS DE LIQUIDEZ AVANZADO ==================== def analyze_liquidity_conditions(self, dataframe: DataFrame, pair: str) -> Dict[str, float]: """ Analiza condiciones de liquidez y microestructura """ if not self.enable_liquidity_analysis.value: return {'liquidity_score': 0.75, 'spread_quality': 0.75, 'depth_score': 0.75} try: # Análisis de volumen y spread implícito recent_volume = dataframe['volume'].tail(20) volume_consistency = 1 - recent_volume.std() / recent_volume.mean() if recent_volume.mean() > 0 else 0.5 volume_consistency = max(0, min(1, volume_consistency)) # Análisis de precio (proxy para spread) price_changes = dataframe['close'].pct_change().tail(20) price_volatility = price_changes.std() spread_quality = max(0, min(1, 1 - price_volatility * 50)) # Análisis de depth (basado en high-low range) hl_ranges = ((dataframe['high'] - dataframe['low']) / dataframe['close']).tail(20) avg_hl_range = hl_ranges.mean() depth_score = max(0, min(1, 1 - avg_hl_range * 20)) # Score de liquidez combinado liquidity_score = (volume_consistency * 0.4 + spread_quality * 0.3 + depth_score * 0.3) return { 'liquidity_score': liquidity_score, 'spread_quality': spread_quality, 'depth_score': depth_score, 'volume_consistency': volume_consistency } except Exception: return {'liquidity_score': 0.75, 'spread_quality': 0.75, 'depth_score': 0.75} # ==================== ANÁLISIS DE SENTIMIENTO ==================== def get_market_sentiment(self) -> Dict[str, float]: """ Obtiene análisis de sentimiento del mercado (Fear & Greed Index simulado) """ if not self.enable_sentiment_analysis.value: return {'fear_greed': 50, 'sentiment_score': 0.5} try: # En un entorno real, aquí se conectaría a APIs de sentimiento # Por ahora, simulamos basado en condiciones técnicas # Simular Fear & Greed Index basado en tiempo import time current_hour = datetime.now().hour base_sentiment = 50 + 20 * math.sin(current_hour * math.pi / 12) # Añadir ruido controlado noise = (hash(str(int(time.time() / 3600))) % 20) - 10 fear_greed = max(0, min(100, base_sentiment + noise)) sentiment_score = fear_greed / 100 return { 'fear_greed': fear_greed, 'sentiment_score': sentiment_score, 'sentiment_trend': 'BULLISH' if fear_greed > 60 else 'BEARISH' if fear_greed < 40 else 'NEUTRAL' } except Exception: return {'fear_greed': 50, 'sentiment_score': 0.5} # ==================== SISTEMA DE SCORING CUÁNTICO V4 ==================== def calculate_quantum_score(self, dataframe: DataFrame, metadata: dict) -> float: """ Calcula el score cuántico multidimensional V4 """ if not self.enable_quantum_scoring.value: return 0.75 # Score neutro si está deshabilitado try: pair = metadata.get('pair', '') # Obtener datos de todos los análisis ml_prediction = self.get_ml_prediction(dataframe) market_regime = self.detect_market_regime(dataframe) fractal_data = self.detect_fibonacci_patterns(dataframe) liquidity_data = self.analyze_liquidity_conditions(dataframe, pair) sentiment_data = self.get_market_sentiment() # Componentes del score con pesos adaptativos scores = {} weights = {} # 1. Análisis Técnico Básico (20%) rsi_score = self.calculate_rsi_score(dataframe) macd_score = self.calculate_macd_score(dataframe) trend_score = self.calculate_trend_score(dataframe) technical_score = (rsi_score * 0.4 + macd_score * 0.3 + trend_score * 0.3) scores['technical'] = technical_score weights['technical'] = 0.20 # 2. Machine Learning (25%) scores['ml_prediction'] = ml_prediction weights['ml_prediction'] = 0.25 # 3. Análisis Cuántico (20%) coherence = dataframe.get('quantum_coherence', pd.Series([0.75])).iloc[-1] momentum = dataframe.get('quantum_momentum', pd.Series([0.5])).iloc[-1] quantum_score = (coherence * 0.6 + (momentum + 1) / 2 * 0.4) # Normalizar momentum scores['quantum'] = quantum_score weights['quantum'] = 0.20 # 4. Análisis Fractal (10%) scores['fractal'] = (fractal_data['spiral_strength'] * 0.6 + fractal_data['golden_ratio_alignment'] * 0.4) weights['fractal'] = 0.10 # 5. Liquidez y Microestructura (10%) scores['liquidity'] = liquidity_data['liquidity_score'] weights['liquidity'] = 0.10 # 6. Sentimiento de Mercado (10%) scores['sentiment'] = sentiment_data['sentiment_score'] weights['sentiment'] = 0.10 # 7. Análisis de Volumen (5%) volume_score = self.calculate_volume_score(dataframe) scores['volume'] = volume_score weights['volume'] = 0.05 # Ajustar pesos basado en régimen de mercado if market_regime == 'BULL': weights['ml_prediction'] *= 1.2 weights['sentiment'] *= 1.3 weights['technical'] *= 0.9 elif market_regime == 'BEAR': weights['liquidity'] *= 1.4 weights['quantum'] *= 1.1 weights['sentiment'] *= 0.8 # Normalizar pesos total_weight = sum(weights.values()) for key in weights: weights[key] /= total_weight # Calcular score final final_score = sum(scores[key] * weights[key] for key in scores) # Aplicar multiplicador por tier del símbolo tier_multiplier = self.get_tier_multiplier(pair) adjusted_score = final_score * tier_multiplier # Ajustar por consciencia QBTC consciousness = self.get_qbtc_consciousness() consciousness_bonus = (consciousness - 0.5) * 0.1 # Bonus/malus del 5% final_adjusted_score = adjusted_score + consciousness_bonus # Limitar score a [0, 1] final_score = max(0, min(1, final_adjusted_score)) # Mantener historial para análisis evolutivo self._quantum_score_history.append(final_score) if len(self._quantum_score_history) > 200: self._quantum_score_history.pop(0) return final_score except Exception as e: return 0.5 # Score neutro en caso de error def calculate_rsi_score(self, dataframe: DataFrame) -> float: """Calcula score basado en RSI""" if 'rsi' not in dataframe.columns: return 0.5 rsi = dataframe['rsi'].iloc[-1] # Score alto cuando RSI está en zona de entrada (20-40) if 20 <= rsi <= 40: return 0.8 + (35 - rsi) / 75 # Score entre 0.8-1.0 elif 40 < rsi <= 60: return 0.7 - (rsi - 40) / 100 # Score decreciente 0.7-0.5 else: return max(0.1, 1 - rsi / 100) # Score bajo para RSI extremo def calculate_macd_score(self, dataframe: DataFrame) -> float: """Calcula score basado en MACD""" if 'macd' not in dataframe.columns or 'macdsignal' not in dataframe.columns: return 0.5 macd = dataframe['macd'].iloc[-1] signal = dataframe['macdsignal'].iloc[-1] # MACD sobre señal = score alto if macd > signal: diff = abs(macd - signal) return min(1.0, 0.6 + diff * 1000) # Score entre 0.6-1.0 else: return 0.3 # Score bajo si MACD bajo señal def calculate_trend_score(self, dataframe: DataFrame) -> float: """Calcula score basado en tendencia""" if 'ema21' not in dataframe.columns or 'ema55' not in dataframe.columns: return 0.5 price = dataframe['close'].iloc[-1] ema21 = dataframe['ema21'].iloc[-1] ema55 = dataframe['ema55'].iloc[-1] # Tendencia alcista = score alto if price > ema21 > ema55: return 0.9 elif price > ema21: return 0.7 elif ema21 > ema55: return 0.6 else: return 0.3 def calculate_volume_score(self, dataframe: DataFrame) -> float: """Calcula score basado en volumen""" if 'volume_ratio' not in dataframe.columns: return 0.5 volume_ratio = dataframe['volume_ratio'].iloc[-1] return min(1.0, volume_ratio / 3.0) # Normalizar a [0, 1] def get_tier_multiplier(self, pair: str) -> float: """Obtiene multiplicador basado en tier del símbolo""" for tier, data in self.SYMBOL_TIERS_V4.items(): if pair in data['symbols']: # Tier 1 = 1.0, Tier 6 = 0.8 (más conservador para alta volatilidad) return max(0.8, 1.1 - tier * 0.05) return 1.0 def get_qbtc_consciousness(self) -> float: """Obtiene nivel de consciencia de servicios QBTC""" try: consciousness_data = self.get_qbtc_service_data('metaconsciencia', '/status') if consciousness_data: return consciousness_data.get('consciousness_level', 0.75) except: pass return 0.75 # Valor por defecto # ==================== MÉTODOS AUXILIARES QBTC ==================== @lru_cache(maxsize=256) def get_qbtc_service_data(self, service: str, endpoint: str = '/health', timeout: int = 1) -> Optional[Dict]: """ Obtiene datos de los servicios QBTC con cache y timeout corto """ try: url = f"{self.QBTC_URLS.get(service, '')}{endpoint}" response = requests.get(url, timeout=timeout) if response.status_code == 200: return response.json() except Exception: pass return None def calculate_lunar_phase_multiplier(self) -> float: """ Calcula el multiplicador basado en la fase lunar actual """ now = datetime.now(timezone.utc) new_moon_ref = datetime(2024, 1, 1, tzinfo=timezone.utc) days_since_new_moon = (now - new_moon_ref).days % 29.53 # Mapeo mejorado de fases lunares if days_since_new_moon < 7.38: # Luna creciente return 1.08 + (days_since_new_moon / 7.38) * 0.12 # 1.08 a 1.20 elif days_since_new_moon < 14.77: # Primer cuarto a luna llena return 1.20 + ((days_since_new_moon - 7.38) / 7.38) * 0.25 # 1.20 a 1.45 elif days_since_new_moon < 22.15: # Luna menguante return 1.45 - ((days_since_new_moon - 14.77) / 7.38) * 0.18 # 1.45 a 1.27 else: # Último cuarto return 1.27 - ((days_since_new_moon - 22.15) / 7.38) * 0.19 # 1.27 a 1.08 # ==================== TIMEFRAMES INFORMATIVOS ==================== def informative_pairs(self): """Define pares informativos para análisis multi-timeframe""" pairs = self.dp.current_whitelist() informative_pairs = [] # Optimizado para V4: timeframes >= 15m (principal) for pair in pairs[:15]: # Limitado para optimización informative_pairs.extend([ (pair, '1h'), # Tendencia principal (pair, '4h'), # Contexto mayor (pair, '1d'), # Tendencia de largo plazo ]) # Siempre incluir BTC como referencia universal if 'BTC/USDT' not in [p[0] for p in informative_pairs]: informative_pairs.extend([ ('BTC/USDT', '1h'), ('BTC/USDT', '4h'), ('BTC/USDT', '1d'), ]) return informative_pairs # Función 5m removida - timeframe menor que principal @informative('1h') def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Indicadores para timeframe de 1 hora""" dataframe['rsi_1h'] = ta.RSI(dataframe, timeperiod=14) dataframe['ema21_1h'] = ta.EMA(dataframe, timeperiod=21) dataframe['ema55_1h'] = ta.EMA(dataframe, timeperiod=55) dataframe['ema200_1h'] = ta.EMA(dataframe, timeperiod=200) # MACD horario macd_1h = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9) dataframe['macd_1h'] = macd_1h['macd'] dataframe['macdsignal_1h'] = macd_1h['macdsignal'] # ATR para stop loss dinámico dataframe['atr_1h'] = ta.ATR(dataframe, timeperiod=14) return dataframe @informative('4h') def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Indicadores para timeframe de 4 horas""" dataframe['rsi_4h'] = ta.RSI(dataframe, timeperiod=14) dataframe['ema21_4h'] = ta.EMA(dataframe, timeperiod=21) dataframe['ema89_4h'] = ta.EMA(dataframe, timeperiod=89) dataframe['sma200_4h'] = ta.SMA(dataframe, timeperiod=200) # Bandas de Bollinger para volatilidad bollinger_4h = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=2) dataframe['bb_upper_4h'] = bollinger_4h['upper'] dataframe['bb_lower_4h'] = bollinger_4h['lower'] dataframe['bb_middle_4h'] = bollinger_4h['mid'] # Stochastic para momentum stoch_4h = ta.STOCH(dataframe, fastk_period=14, slowk_period=3, slowd_period=3) dataframe['stoch_k_4h'] = stoch_4h['slowk'] dataframe['stoch_d_4h'] = stoch_4h['slowd'] return dataframe @informative('1d') def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Indicadores para timeframe diario""" dataframe['ema21_1d'] = ta.EMA(dataframe, timeperiod=21) dataframe['ema50_1d'] = ta.EMA(dataframe, timeperiod=50) dataframe['ema200_1d'] = ta.EMA(dataframe, timeperiod=200) dataframe['rsi_1d'] = ta.RSI(dataframe, timeperiod=14) # ATR diario para gestión de riesgo dataframe['atr_1d'] = ta.ATR(dataframe, timeperiod=14) # Williams %R para momentum largo plazo dataframe['willr_1d'] = ta.WILLR(dataframe, timeperiod=14) return dataframe # ==================== INDICADORES PRINCIPALES V4 ==================== def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Población de indicadores principales con lógica cuántica V4 """ # ========== INDICADORES TÉCNICOS BÁSICOS ========== # RSI con múltiples períodos dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['rsi_fast'] = ta.RSI(dataframe, timeperiod=7) dataframe['rsi_slow'] = ta.RSI(dataframe, timeperiod=21) dataframe['rsi_momentum'] = dataframe['rsi'] - dataframe['rsi'].shift(3) # MACD con parámetros adaptativos macd = ta.MACD(dataframe, fastperiod=self.macd_fast.value, slowperiod=self.macd_slow.value, signalperiod=self.macd_signal.value) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] dataframe['macdhist'] = macd['macdhist'] dataframe['macd_momentum'] = dataframe['macdhist'] - dataframe['macdhist'].shift(2) # EMAs con secuencia de Fibonacci fib_periods = [8, 13, 21, 34, 55, 89, 144, 233] for period in fib_periods: dataframe[f'ema{period}'] = ta.EMA(dataframe, timeperiod=period) # SMAs para contexto dataframe['sma20'] = ta.SMA(dataframe, timeperiod=20) dataframe['sma50'] = ta.SMA(dataframe, timeperiod=50) dataframe['sma200'] = ta.SMA(dataframe, timeperiod=200) # ========== INDICADORES DE VOLATILIDAD ========== # Bandas de Bollinger mejoradas for window, std in [(20, 2), (20, 2.5)]: bollinger = qtpylib.bollinger_bands(dataframe['close'], window=window, stds=std) suffix = f"_std{int(std*10)}" if std != 2 else "" dataframe[f'bb_lower{suffix}'] = bollinger['lower'] dataframe[f'bb_middle{suffix}'] = bollinger['mid'] dataframe[f'bb_upper{suffix}'] = bollinger['upper'] dataframe['bb_percent'] = ((dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower'])) dataframe['bb_width'] = ((dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle']) # ATR y volatilidad dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['atr_percent'] = dataframe['atr'] / dataframe['close'] dataframe['volatility'] = dataframe['close'].pct_change().rolling(20).std() * np.sqrt(252) # ========== INDICADORES DE VOLUMEN AVANZADOS ========== # Volumen relativo con múltiples períodos for period in [10, 20, 50]: dataframe[f'volume_mean_{period}'] = dataframe['volume'].rolling(period).mean() dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_mean_20'] dataframe['volume_trend'] = dataframe['volume_mean_10'] / dataframe['volume_mean_50'] # Indicadores de volumen-precio dataframe['mfi'] = ta.MFI(dataframe, timeperiod=14) dataframe['ad'] = ta.AD(dataframe) # Accumulation/Distribution dataframe['obv'] = ta.OBV(dataframe) # On Balance Volume # Volume Profile simplificado dataframe['vwap'] = (dataframe['volume'] * (dataframe['high'] + dataframe['low'] + dataframe['close']) / 3).cumsum() / dataframe['volume'].cumsum() # ========== INDICADORES DE TENDENCIA AVANZADOS ========== # ADX y direccionales dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) dataframe['plus_di'] = ta.PLUS_DI(dataframe, timeperiod=14) dataframe['minus_di'] = ta.MINUS_DI(dataframe, timeperiod=14) dataframe['dx'] = ta.DX(dataframe, timeperiod=14) # Parabolic SAR dataframe['sar'] = ta.SAR(dataframe, acceleration=0.02, maximum=0.2) # Aroon aroon = ta.AROON(dataframe, timeperiod=25) dataframe['aroon_up'] = aroon['aroonup'] dataframe['aroon_down'] = aroon['aroondown'] dataframe['aroon_osc'] = dataframe['aroon_up'] - dataframe['aroon_down'] # ========== INDICADORES CUÁNTICOS V4 ========== # Coherencia cuántica mejorada dataframe['quantum_coherence'] = self.calculate_quantum_coherence_v4(dataframe) # Entrelazamiento cuántico avanzado dataframe['quantum_entanglement'] = self.calculate_quantum_entanglement_v4(dataframe) # Momentum cuántico multidimensional dataframe['quantum_momentum'] = self.calculate_quantum_momentum_v4(dataframe) # Dimensión fractal dataframe['fractal_dimension'] = dataframe['close'].rolling(55).apply( lambda x: self.calculate_fractal_dimension(x), raw=False ) # ========== NIVELES DE FIBONACCI DINÁMICOS ========== # Múltiples períodos para Fibonacci for period in [34, 55, 89, 144]: high_col = f'high_{period}' low_col = f'low_{period}' dataframe[high_col] = dataframe['high'].rolling(period).max() dataframe[low_col] = dataframe['low'].rolling(period).min() # Niveles principales dataframe[f'fib_236_{period}'] = dataframe[high_col] - ((dataframe[high_col] - dataframe[low_col]) * 0.236) dataframe[f'fib_382_{period}'] = dataframe[high_col] - ((dataframe[high_col] - dataframe[low_col]) * 0.382) dataframe[f'fib_618_{period}'] = dataframe[high_col] - ((dataframe[high_col] - dataframe[low_col]) * 0.618) dataframe[f'fib_786_{period}'] = dataframe[high_col] - ((dataframe[high_col] - dataframe[low_col]) * 0.786) # ========== PATRONES DE VELAS AVANZADOS ========== # Patrones alcistas bullish_patterns = ['HAMMER', 'MORNINGSTAR', 'ENGULFING', 'PIERCING', 'HARAMI'] for pattern in bullish_patterns: dataframe[f'cdl_{pattern.lower()}'] = getattr(ta, f'CDL{pattern}')(dataframe) # Patrones bajistas bearish_patterns = ['SHOOTINGSTAR', 'EVENINGSTAR', 'DARKCLOUDCOVER', 'HANGINGMAN'] for pattern in bearish_patterns: dataframe[f'cdl_{pattern.lower()}'] = getattr(ta, f'CDL{pattern}')(dataframe) # ========== INDICADORES DE MOMENTUM ========== # Stochastic stoch = ta.STOCH(dataframe, fastk_period=14, slowk_period=3, slowd_period=3) dataframe['stoch_k'] = stoch['slowk'] dataframe['stoch_d'] = stoch['slowd'] # Williams %R dataframe['willr'] = ta.WILLR(dataframe, timeperiod=14) # CCI (Commodity Channel Index) dataframe['cci'] = ta.CCI(dataframe, timeperiod=20) # Ultimate Oscillator dataframe['uo'] = ta.ULTOSC(dataframe, timeperiod1=7, timeperiod2=14, timeperiod3=28) # ========== INTEGRACIÓN CON SERVICIOS QBTC V4 ========== # Multiplicadores temporales dataframe['lunar_multiplier'] = self.calculate_lunar_phase_multiplier() # Multiplicador por tier del símbolo pair = metadata.get('pair', '') tier_data = self.get_symbol_tier_data(pair) dataframe['tier_multiplier'] = tier_data['weight'] dataframe['consciousness_requirement'] = tier_data['consciousness_req'] # ========== SEÑALES COMPUESTAS V4 ========== # Señal de tendencia multi-timeframe mejorada dataframe['trend_signal'] = ( np.where(dataframe['ema21'] > dataframe['ema55'], 1, -1) * 0.25 + np.where(dataframe['ema55'] > dataframe['ema89'], 1, -1) * 0.20 + np.where(dataframe['close'] > dataframe['sma200'], 1, -1) * 0.20 + np.where(dataframe['adx'] > 25, np.where(dataframe['plus_di'] > dataframe['minus_di'], 1, -1), 0) * 0.20 + np.where(dataframe['aroon_up'] > dataframe['aroon_down'], 1, -1) * 0.15 ) # Señal de momentum avanzada dataframe['momentum_signal'] = ( np.where(dataframe['rsi'] > 50, 1, -1) * 0.25 + np.where(dataframe['macd'] > dataframe['macdsignal'], 1, -1) * 0.30 + np.where(dataframe['stoch_k'] > dataframe['stoch_d'], 1, -1) * 0.20 + np.where(dataframe['mfi'] > 50, 1, -1) * 0.15 + np.where(dataframe['cci'] > 0, 1, -1) * 0.10 ) # Señal de volumen dataframe['volume_signal'] = ( np.where(dataframe['volume_ratio'] > 1.2, 1, 0) * 0.4 + np.where(dataframe['mfi'] > 50, 1, -1) * 0.3 + np.where(dataframe['ad'] > dataframe['ad'].shift(5), 1, -1) * 0.3 ) # ========== SCORE CUÁNTICO PRINCIPAL ========== # Calcular el score cuántico multidimensional dataframe['quantum_score'] = 0.0 # Placeholder # El score se calculará en las funciones de entrada/salida para optimización return dataframe def calculate_quantum_coherence_v4(self, dataframe: DataFrame) -> pd.Series: """ Calcula coherencia cuántica V4 con múltiples dimensiones """ try: # Coherencia de precio price_ma = dataframe['close'].rolling(21).mean() price_std = dataframe['close'].rolling(21).std() price_coherence = 1 - (price_std / price_ma).fillna(0) price_coherence = np.clip(price_coherence, 0, 1) # Coherencia de volumen vol_ma = dataframe['volume'].rolling(21).mean() vol_std = dataframe['volume'].rolling(21).std() volume_coherence = 1 - (vol_std / vol_ma).fillna(0) volume_coherence = np.clip(volume_coherence, 0, 1) # Coherencia técnica (correlación indicadores) rsi_norm = dataframe['rsi'] / 100 rsi_coherence = 1 - abs(rsi_norm - 0.5) * 2 macd_coherence = np.where( dataframe['macd'] * dataframe['macdsignal'] > 0, 0.8, 0.2 ) # Coherencia combinada con pesos adaptativos coherence = ( price_coherence * 0.35 + volume_coherence * 0.25 + rsi_coherence * 0.20 + macd_coherence * 0.20 ) # Suavizado exponencial return coherence.ewm(span=13, adjust=False).mean() except Exception: return pd.Series([0.75] * len(dataframe), index=dataframe.index) def calculate_quantum_entanglement_v4(self, dataframe: DataFrame) -> pd.Series: """ Calcula entrelazamiento cuántico V4 avanzado """ try: # Entrelazamiento precio-volumen price_volume_corr = dataframe['close'].rolling(21).corr(dataframe['volume']) pv_entanglement = abs(price_volume_corr).fillna(0.5) # Entrelazamiento multi-timeframe fast_ma = dataframe['ema8'] medium_ma = dataframe['ema21'] slow_ma = dataframe['ema55'] # Sincronización de medias móviles fm_sync = 1 - abs(fast_ma - medium_ma) / medium_ma ms_sync = 1 - abs(medium_ma - slow_ma) / slow_ma ma_entanglement = (fm_sync + ms_sync) / 2 ma_entanglement = np.clip(ma_entanglement.fillna(0.5), 0, 1) # Entrelazamiento de momentum rsi_macd_corr = dataframe['rsi'].rolling(21).corr(dataframe['macd']) momentum_entanglement = abs(rsi_macd_corr).fillna(0.5) # Entrelazamiento cuántico combinado entanglement = ( pv_entanglement * 0.4 + ma_entanglement * 0.35 + momentum_entanglement * 0.25 ) return entanglement.ewm(span=13, adjust=False).mean() except Exception: return pd.Series([0.5] * len(dataframe), index=dataframe.index) def calculate_quantum_momentum_v4(self, dataframe: DataFrame) -> pd.Series: """ Calcula momentum cuántico V4 multidimensional """ try: # Momentum de precio price_momentum = (dataframe['close'] / dataframe['ema21'] - 1) # Momentum técnico rsi_momentum = (dataframe['rsi'] - 50) / 50 macd_momentum = np.where(dataframe['macd'] > dataframe['macdsignal'], 1, -1) # Momentum de volumen volume_momentum = (dataframe['volume_ratio'] - 1) # Momentum de tendencia trend_momentum = np.where(dataframe['trend_signal'] > 0.2, 1, np.where(dataframe['trend_signal'] < -0.2, -1, 0)) # Momentum cuántico combinado quantum_momentum = ( price_momentum * 0.3 + rsi_momentum * 0.25 + macd_momentum * 0.20 + volume_momentum * 0.15 + trend_momentum * 0.10 ) # Normalizar y suavizar return quantum_momentum.ewm(span=13, adjust=False).mean() except Exception: return pd.Series([0.0] * len(dataframe), index=dataframe.index) def get_symbol_tier_data(self, pair: str) -> Dict: """Obtiene datos del tier del símbolo""" for tier, data in self.SYMBOL_TIERS_V4.items(): if pair in data['symbols']: return data # Default para símbolos no encontrados return { 'consciousness_req': 0.75, 'weight': 0.05, 'max_risk': 0.03 } # ==================== LÓGICA DE ENTRADA V4 ==================== def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Lógica de entrada V4 basada en scoring cuántico multidimensional """ # Calcular quantum score para cada vela quantum_scores = [] for i in range(len(dataframe)): if i < 50: # No suficientes datos para score quantum_scores.append(0.5) continue # Crear subset para cálculo de score subset = dataframe.iloc[:i+1] score = self.calculate_quantum_score(subset, metadata) quantum_scores.append(score) dataframe['quantum_score'] = quantum_scores # ========== CONDICIONES DE ENTRADA PRINCIPALES V4 ========== conditions_entry = [] # 1. Score cuántico alto (condición principal) quantum_condition = dataframe['quantum_score'] >= self.quantum_score_threshold.value conditions_entry.append(quantum_condition) # 2. Confirmación técnica básica technical_conditions = ( (dataframe['rsi'] >= 25) & (dataframe['rsi'] <= self.rsi_entry.value) & (dataframe['macd'] > dataframe['macdsignal']) & (dataframe['trend_signal'] > 0.1) ) conditions_entry.append(technical_conditions) # 3. Confirmación de momentum momentum_conditions = ( (dataframe['momentum_signal'] > 0.2) & (dataframe['quantum_momentum'] > -0.1) ) conditions_entry.append(momentum_conditions) # 4. Confirmación de volumen volume_conditions = ( (dataframe['volume_ratio'] >= self.volume_multiplier.value) & (dataframe['volume_signal'] > 0.3) ) conditions_entry.append(volume_conditions) # 5. Confirmación de coherencia cuántica coherence_conditions = ( (dataframe['quantum_coherence'] >= self.coherence_threshold.value) & (dataframe['quantum_entanglement'] >= 0.5) ) conditions_entry.append(coherence_conditions) # 6. No en zona de sobrecompra extrema not_overbought = ( (dataframe['rsi'] < 85) & (dataframe['mfi'] < 90) & (dataframe['stoch_k'] < 85) ) conditions_entry.append(not_overbought) # ========== CONDICIONES ESPECIALES POR TIER Y RÉGIMEN ========== pair = metadata.get('pair', '') tier_data = self.get_symbol_tier_data(pair) # Obtener régimen actual current_regime = self.detect_market_regime(dataframe) # Ajuste por régimen de mercado regime_multiplier = 1.0 if current_regime == 'BULL': regime_multiplier = self.regime_adaptation.value * 1.1 elif current_regime == 'BEAR': regime_multiplier = self.regime_adaptation.value * 0.9 # Condición de consciencia por tier consciousness_condition = True qbtc_consciousness = self.get_qbtc_consciousness() if qbtc_consciousness < tier_data['consciousness_req']: consciousness_condition = False # ========== INTEGRACIÓN CON SERVICIOS QBTC ========== # Guardian approval guardian_approved = True guardian_data = self.get_qbtc_service_data('guardian', '/pre_trade_check') if guardian_data and not guardian_data.get('approved', True): guardian_approved = False # MetaConsciencia decision meta_decision = 'NEUTRAL' meta_data = self.get_qbtc_service_data('metaconsciencia', '/decision') if meta_data: meta_decision = meta_data.get('action', 'NEUTRAL') meta_condition = meta_decision in ['BUY', 'HOLD', 'NEUTRAL'] # ========== MULTIPLICADORES LUNARES Y FRACTALES ========== lunar_boost = dataframe['lunar_multiplier'] > 1.25 # Fractal condition (si está habilitado) fractal_condition = True if self.enable_fractal_analysis.value: fractal_condition = ( (dataframe['fractal_dimension'] >= self.fractal_dimension_min.value) & (dataframe['fractal_dimension'] <= 1.8) ) # ========== APLICACIÓN FINAL DE CONDICIONES ========== # Combinar condiciones principales main_conditions = conditions_entry[0] # Quantum score for condition in conditions_entry[1:]: main_conditions = main_conditions & condition # Condiciones adicionales additional_conditions = ( (dataframe['volume'] > 0) & consciousness_condition & guardian_approved & meta_condition & fractal_condition ) # Aplicar multiplicador de régimen al quantum score regime_adjusted_quantum = dataframe['quantum_score'] * regime_multiplier regime_condition = regime_adjusted_quantum >= self.quantum_score_threshold.value # Condición final con boost lunar opcional final_entry_condition = ( main_conditions & additional_conditions & regime_condition ) | ( # Condiciones relajadas durante luna muy favorable lunar_boost & quantum_condition & technical_conditions & (dataframe['volume_ratio'] >= self.volume_multiplier.value * 0.8) & consciousness_condition & guardian_approved ) dataframe.loc[final_entry_condition, 'enter_long'] = 1 return dataframe # ==================== LÓGICA DE SALIDA V4 ==================== def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Lógica de salida V4 con inteligencia adaptativa """ # ========== CONDICIONES DE SALIDA PRINCIPALES ========== exit_conditions = [] # 1. Quantum score bajo (pérdida de oportunidad) quantum_exit = dataframe['quantum_score'] < (self.quantum_score_threshold.value - 0.15) exit_conditions.append(quantum_exit) # 2. RSI sobrecompra rsi_exit = dataframe['rsi'] >= self.rsi_exit.value exit_conditions.append(rsi_exit) # 3. MACD bajista con confirmación macd_exit = ( (dataframe['macd'] < dataframe['macdsignal']) & (dataframe['macd_momentum'] < -0.01) ) exit_conditions.append(macd_exit) # 4. Pérdida de tendencia trend_break = ( (dataframe['trend_signal'] < -0.2) & (dataframe['close'] < dataframe['ema21']) ) exit_conditions.append(trend_break) # 5. Divergencia de momentum momentum_divergence = ( (dataframe['momentum_signal'] < -0.3) & (dataframe['quantum_momentum'] < -0.2) ) exit_conditions.append(momentum_divergence) # 6. Pérdida de coherencia cuántica coherence_loss = ( (dataframe['quantum_coherence'] < (self.coherence_threshold.value - 0.15)) & (dataframe['quantum_entanglement'] < 0.3) ) exit_conditions.append(coherence_loss) # 7. Volumen débil con precio alto volume_divergence = ( (dataframe['volume_ratio'] < 0.6) & (dataframe['close'] > dataframe['close'].shift(10)) & (dataframe['volume_signal'] < 0.2) ) exit_conditions.append(volume_divergence) # 8. Patrones bajistas confirmados bearish_patterns = ( (dataframe['cdl_shootingstar'] < 0) | (dataframe['cdl_eveningstar'] < 0) | (dataframe['cdl_hangingman'] < 0) ) exit_conditions.append(bearish_patterns) # 9. Múltiples indicadores de momentum bajista multi_momentum_exit = ( (dataframe['stoch_k'] > 80) & (dataframe['willr'] > -20) & (dataframe['cci'] > 100) ) exit_conditions.append(multi_momentum_exit) # ========== SALIDAS DE EMERGENCIA Y SERVICIOS QBTC ========== # Guardian emergency exit guardian_exit = False guardian_data = self.get_qbtc_service_data('guardian', '/risk_assessment') if guardian_data: risk_level = guardian_data.get('risk_level', 'LOW') guardian_exit = risk_level in ['HIGH', 'CRITICAL', 'EMERGENCY'] # MetaConsciencia forced exit meta_exit = False meta_data = self.get_qbtc_service_data('metaconsciencia', '/decision') if meta_data: decision = meta_data.get('action', 'HOLD') meta_exit = decision in ['SELL', 'SELL_ALL', 'EMERGENCY_EXIT'] # ========== DETECCIÓN DE RÉGIMEN PARA SALIDAS ADAPTATIVAS ========== current_regime = self.detect_market_regime(dataframe) # Ajustar condiciones según régimen if current_regime == 'BEAR': # En mercado bajista, salir más agresivamente aggressive_exit = ( (dataframe['quantum_score'] < 0.6) | (dataframe['rsi'] > 70) | (dataframe['trend_signal'] < 0) ) exit_conditions.append(aggressive_exit) elif current_regime == 'BULL': # En mercado alcista, ser más paciente pero proteger ganancias profit_protection = ( (dataframe['rsi'] > 85) & (dataframe['quantum_score'] < 0.65) ) exit_conditions.append(profit_protection) # ========== APLICACIÓN DE LÓGICA DE SALIDA ========== # Salida por cualquier condición principal standard_exit = exit_conditions[0] for condition in exit_conditions[1:]: standard_exit = standard_exit | condition # Salida por múltiples confirmaciones (más conservadora) multiple_confirmations = ( (quantum_exit & (rsi_exit | macd_exit)) | (trend_break & coherence_loss) | (momentum_divergence & volume_divergence) ) # Condición final de salida final_exit_condition = ( standard_exit | multiple_confirmations | guardian_exit | meta_exit ) dataframe.loc[final_exit_condition, 'exit_long'] = 1 return dataframe # ==================== GESTIÓN AVANZADA DE ÓRDENES V4 ==================== def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Stop loss dinámico V4 con inteligencia cuántica """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) < 1: return self.stoploss last_candle = dataframe.iloc[-1] # ATR dinámico base base_atr_multiplier = 2.5 # Ajustar por coherencia cuántica if 'quantum_coherence' in dataframe.columns: coherence = last_candle['quantum_coherence'] if coherence > 0.8: base_atr_multiplier = 1.8 # Stop más ajustado con alta coherencia elif coherence < 0.5: base_atr_multiplier = 3.2 # Stop más amplio con baja coherencia # Ajustar por régimen de mercado current_regime = self.detect_market_regime(dataframe) if current_regime == 'BULL': base_atr_multiplier *= 0.9 # Más agresivo en bull market elif current_regime == 'BEAR': base_atr_multiplier *= 1.2 # Más conservador en bear market # Ajustar por tier del símbolo tier_data = self.get_symbol_tier_data(pair) risk_multiplier = 1.0 / tier_data['max_risk'] * 0.02 # Normalizar base_atr_multiplier *= risk_multiplier # Aplicar ATR stop loss if 'atr_percent' in dataframe.columns: dynamic_stoploss = -last_candle['atr_percent'] * base_atr_multiplier # Límites adaptativos por tier max_stop = -tier_data['max_risk'] * 3 # Máximo 3x el riesgo del tier min_stop = -0.04 # Mínimo 4% dynamic_stoploss = max(dynamic_stoploss, max_stop) dynamic_stoploss = min(dynamic_stoploss, min_stop) return dynamic_stoploss return self.stoploss def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]: """ Lógica de salida personalizada V4 con IA avanzada """ # Consultar MetaConsciencia para decisión inteligente meta_data = self.get_qbtc_service_data('metaconsciencia', '/trade_decision') if meta_data: decision = meta_data.get('action', '') confidence = meta_data.get('confidence', 0.5) if decision in ['SELL_ALL', 'EMERGENCY_EXIT'] and confidence > 0.7: return 'metaconsciencia_emergency' elif decision == 'SELL' and confidence > 0.8: return 'metaconsciencia_sell' # Guardian risk assessment guardian_data = self.get_qbtc_service_data('guardian', '/current_risk') if guardian_data: risk_level = guardian_data.get('risk_level', 'LOW') if risk_level == 'CRITICAL': return 'guardian_emergency' elif risk_level == 'HIGH' and current_profit < 0.05: return 'guardian_risk_management' # Duración del trade trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 # Obtener quantum score actual dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) > 0: current_quantum_score = dataframe['quantum_score'].iloc[-1] if 'quantum_score' in dataframe.columns else 0.75 else: current_quantum_score = 0.75 # Toma de beneficios adaptativa por quantum score if current_quantum_score > 0.85: # Score alto = mantener posición más tiempo if current_profit > 0.30 and trade_duration > 1.0: # 30% en 1h return 'quantum_high_score_profit_30' elif current_profit > 0.20 and trade_duration > 2.0: # 20% en 2h return 'quantum_high_score_profit_20' else: # Score bajo = tomar beneficios más rápido if current_profit > 0.15 and trade_duration > 0.5: # 15% en 30min return 'quantum_low_score_profit_15' elif current_profit > 0.10 and trade_duration > 1.0: # 10% en 1h return 'quantum_low_score_profit_10' # Toma de beneficios estándar con régimen current_regime = self.detect_market_regime(dataframe) if current_regime == 'BULL': # En bull market, ser más paciente if current_profit > 0.25 and trade_duration > 1.5: return 'bull_market_profit_25' elif current_profit > 0.15 and trade_duration > 3.0: return 'bull_market_profit_15' elif current_regime == 'BEAR': # En bear market, tomar beneficios rápido if current_profit > 0.12 and trade_duration > 0.5: return 'bear_market_profit_12' elif current_profit > 0.08 and trade_duration > 1.5: return 'bear_market_profit_8' # Salida por tiempo extendido sin beneficios significativos if trade_duration > 8 and current_profit < 0.05: return 'extended_time_low_profit' return None # ==================== CONFIGURACIÓN DE PLOTS V4 ==================== plot_config = { 'main_plot': { # EMAs principales 'ema21': {'color': 'orange', 'type': 'line'}, 'ema55': {'color': 'cyan', 'type': 'line'}, 'ema89': {'color': 'purple', 'type': 'line'}, 'sma200': {'color': 'white', 'type': 'line'}, # Bandas de Bollinger 'bb_upper': {'color': 'gray', 'type': 'line'}, 'bb_lower': {'color': 'gray', 'type': 'line'}, 'bb_middle': {'color': 'lightgray', 'type': 'line'}, # Fibonacci 'fib_382_55': {'color': 'yellow', 'type': 'line'}, 'fib_618_55': {'color': 'gold', 'type': 'line'}, # SAR 'sar': {'color': 'blue', 'type': 'scatter'}, }, 'subplots': { 'RSI Multi': { 'rsi': {'color': 'red'}, 'rsi_fast': {'color': 'orange'}, 'rsi_slow': {'color': 'blue'}, }, 'MACD': { 'macd': {'color': 'blue'}, 'macdsignal': {'color': 'orange'}, 'macdhist': {'color': 'gray'}, }, 'Quantum Metrics V4': { 'quantum_coherence': {'color': 'purple'}, 'quantum_entanglement': {'color': 'magenta'}, 'quantum_momentum': {'color': 'cyan'}, 'quantum_score': {'color': 'gold'}, }, 'Volume Analysis': { 'volume_ratio': {'color': 'green'}, 'mfi': {'color': 'blue'}, }, 'Trend & Momentum': { 'trend_signal': {'color': 'orange'}, 'momentum_signal': {'color': 'cyan'}, 'adx': {'color': 'purple'}, }, 'Fractal & Lunar': { 'fractal_dimension': {'color': 'brown'}, 'lunar_multiplier': {'color': 'silver'}, } } } # ==================== MÉTODOS DE INFORMACIÓN V4 ==================== def version(self) -> str: """Retorna la versión de la estrategia""" return "QBTC Quantum Strategy V4.0 REVOLUTIONARY" def bot_start(self, **kwargs) -> None: """ Método llamado al iniciar el bot - configuración inicial V4 """ logger = logging.getLogger(__name__) logger.info("🌌 ========================================== 🌌") logger.info("🚀 INICIANDO QBTC QUANTUM STRATEGY V4.0 🚀") logger.info("🌌 ========================================== 🌌") logger.info(f"🔮 Quantum Score Threshold: {self.quantum_score_threshold.value}") logger.info(f"🌊 Coherence Threshold: {self.coherence_threshold.value}") logger.info(f"🧠 Consciousness Threshold: {self.consciousness_threshold.value}") logger.info(f"⚡ Lambda Sensitivity: {self.lambda_sensitivity.value}") # Mostrar módulos habilitados modules = { 'ML Prediction': self.enable_ml_prediction.value, 'Regime Detection': self.enable_regime_detection.value, 'Fractal Analysis': self.enable_fractal_analysis.value, 'Liquidity Analysis': self.enable_liquidity_analysis.value, 'Sentiment Analysis': self.enable_sentiment_analysis.value, 'Quantum Scoring': self.enable_quantum_scoring.value, } logger.info("📊 MÓDULOS V4 HABILITADOS:") for module, enabled in modules.items(): status = "✅ ENABLED" if enabled else "❌ DISABLED" logger.info(f" {module}: {status}") # Verificar conectividad QBTC logger.info("🔗 VERIFICANDO SERVICIOS QBTC:") for service_name, url in self.QBTC_URLS.items(): try: response = requests.get(f"{url}/health", timeout=1) if response.status_code == 200: logger.info(f" ✅ {service_name.upper()}: ONLINE") else: logger.warning(f" ⚠️ {service_name.upper()}: NO RESPONSE") except Exception: logger.warning(f" ❌ {service_name.upper()}: OFFLINE") # Información sobre tiers logger.info("🏛️ SYMBOL TIERS V4 CONFIGURADOS:") total_symbols = 0 for tier, data in self.SYMBOL_TIERS_V4.items(): symbol_count = len(data['symbols']) total_symbols += symbol_count logger.info(f" Tier {tier}: {symbol_count} symbols, Weight: {data['weight']*100:.1f}%, Risk: {data['max_risk']*100:.1f}%") logger.info(f"📈 TOTAL SYMBOLS SUPPORTED: {total_symbols}") # Inicializar caches self._ml_models = {} self._regime_history = [] self._quantum_score_history = [] self._fractal_cache = {} logger.info("🎯 QBTC QUANTUM STRATEGY V4.0 LISTO PARA TRADING") logger.info("🌌 ========================================== 🌌") def check_entry_timeout(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> bool: """ Verifica timeout de órdenes de entrada con lógica cuántica V4 """ tier_data = self.get_symbol_tier_data(pair) base_timeout = 300 # 5 minutos base # Timeout más largo para tiers de alta volatilidad tier_multiplier = 1.0 + (6 - list(self.SYMBOL_TIERS_V4.keys()).index( next(tier for tier, data in self.SYMBOL_TIERS_V4.items() if pair in data['symbols']) )) * 0.2 timeout_seconds = base_timeout * tier_multiplier # Ajustar por régimen de mercado dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) > 0: current_regime = self.detect_market_regime(dataframe) if current_regime == 'BULL': timeout_seconds *= 1.2 # Más tiempo en bull market elif current_regime == 'BEAR': timeout_seconds *= 0.8 # Menos tiempo en bear market return (current_time - order.order_date_utc).total_seconds() > timeout_seconds def check_exit_timeout(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> bool: """ Verifica timeout de órdenes de salida V4 """ # Timeout base más corto para salidas base_timeout = 600 # 10 minutos # Ajustar por profit actual current_profit = trade.calc_profit_ratio(trade.close_rate or order.price or 0) if current_profit > 0.10: # Más de 10% profit base_timeout *= 1.5 # Más tiempo para salidas con ganancia elif current_profit < -0.05: # Pérdida base_timeout *= 0.7 # Menos tiempo para salidas con pérdida return (current_time - order.order_date_utc).total_seconds() > base_timeout def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: """ Confirmación final V4 con Guardian y MetaConsciencia """ # Guardian confirmation guardian_data = self.get_qbtc_service_data('guardian', '/validate_trade') if guardian_data: if not guardian_data.get('approved', True): logger = logging.getLogger(__name__) reason = guardian_data.get('reason', 'Unknown') logger.info(f"❌ GUARDIAN REJECTED: {pair} - {reason}") return False # MetaConsciencia confirmation meta_data = self.get_qbtc_service_data('metaconsciencia', '/pre_trade_analysis') if meta_data: confidence = meta_data.get('confidence', 0.5) recommendation = meta_data.get('recommendation', 'HOLD') if confidence < 0.6 or recommendation == 'AVOID': logger = logging.getLogger(__name__) logger.info(f"⚠️ METACONSCIENCIA LOW CONFIDENCE: {pair} - {confidence:.2f}") return False # Verificar límites por tier tier_data = self.get_symbol_tier_data(pair) trade_value = amount * rate max_trade_value = tier_data['max_risk'] * 50000 # Base de $50k if trade_value > max_trade_value: logger = logging.getLogger(__name__) logger.info(f"⚠️ TRADE SIZE LIMIT: {pair} ${trade_value:.2f} > ${max_trade_value:.2f}") return False # Verificar quantum score mínimo dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) > 0: current_quantum_score = dataframe['quantum_score'].iloc[-1] if 'quantum_score' in dataframe.columns else 0.5 if current_quantum_score < self.quantum_score_threshold.value: logger = logging.getLogger(__name__) logger.info(f"⚠️ LOW QUANTUM SCORE: {pair} - {current_quantum_score:.3f}") return False logger = logging.getLogger(__name__) logger.info(f"✅ TRADE CONFIRMED: {pair} ${trade_value:.2f} @ {rate} | Quantum V4") return True def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime, **kwargs) -> bool: """ Confirmación de salida del trade V4 """ profit_percent = trade.calc_profit_ratio(rate) * 100 trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 logger = logging.getLogger(__name__) logger.info(f"💰 EXIT CONFIRMED: {pair}") logger.info(f" 📊 Reason: {exit_reason}") logger.info(f" 💎 Profit: {profit_percent:.2f}%") logger.info(f" ⏱️ Duration: {trade_duration:.1f}h") logger.info(f" 🎯 Rate: {rate}") # Siempre permitir salidas de emergencia emergency_reasons = ['emergency', 'guardian', 'metaconsciencia', 'risk', 'critical'] if any(reason in exit_reason.lower() for reason in emergency_reasons): logger.info("🚨 EMERGENCY EXIT APPROVED") return True return True # ==================== FIN DE LA ESTRATEGIA V4 ====================