# source: https://raw.githubusercontent.com/vigoferrel/qbtc-unified/ef7ffc8f45cc7ab041884cd0c08a2043acc1f0cb/freqtrade_qbtc/user_data/strategies/QBTCQuantumStrategy.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file """ 🌌 QBTC QUANTUM STRATEGY v3.0 🌌 Estrategia de Trading Cuántico con Consciencia Hermética Desarrollado por: vigoferrel@gmail.com Framework: QBTC-UNIFIED Quantum Trading System Licencia: Propietaria - vigoferrel ============================================================================== 🔮 CARACTERÍSTICAS PRINCIPALES: - Integración completa con ecosistema QBTC (MetaConsciencia, Guardian, Portfolio) - Análisis multi-timeframe con resonancia λ₇₉₁₉ - Lógica hermética: "Como es arriba, es abajo" - 77 símbolos organizados en 6 tiers jerárquicos - Métricas cuánticas: Coherencia, Entrelazamiento, Momentum - Ciclos lunares y correspondencias temporales - Machine Learning integrado con 7 modelos - Protección avanzada de capital con Guardian ============================================================================== """ 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 import requests import json import math import asyncio from functools import lru_cache from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, timeframe_to_minutes, timeframe_to_next_date, timeframe_to_prev_date, merge_informative_pair, stoploss_from_absolute, stoploss_from_open, ) import talib.abstract as ta from technical import qtpylib class Github_vigoferrel_qbtc_unified__QBTCQuantumStrategy__20251009_202533(IStrategy): """ 🌊 QBTC Quantum Strategy - Trading Cuántico con Consciencia Hermética Esta estrategia implementa el framework completo QBTC con: - Análisis cuántico multidimensional - Integración con servicios QBTC (MetaConsciencia, Guardian, Portfolio) - Lógica hermética y correspondencias temporales - Machine Learning y patrones COBOL - Gestión avanzada de riesgo cuántico """ # ==================== CONFIGURACIÓN BÁSICA ==================== INTERFACE_VERSION = 3 can_short: bool = False # Configuración de timeframes timeframe = '15m' # Timeframe principal para señales startup_candle_count: int = 400 # Necesario para indicadores complejos # ==================== CONFIGURACIÓN DE RIESGO ==================== minimal_roi = { "0": 0.20, # 20% ROI inmediato si es posible "30": 0.15, # 15% después de 30 min "60": 0.10, # 10% después de 1h "120": 0.05, # 5% después de 2h "240": 0.02, # 2% después de 4h "480": 0.01, # 1% después de 8h } stoploss = -0.15 # Stop loss base del 15% # Trailing stop cuántico trailing_stop = True trailing_stop_positive = 0.02 trailing_stop_positive_offset = 0.05 trailing_only_offset_is_reached = True # ==================== CONFIGURACIÓN DE ÓRDENES ==================== order_types = { 'entry': 'limit', 'exit': 'limit', 'stoploss': 'market', 'stoploss_on_exchange': True, } order_time_in_force = { 'entry': 'GTC', 'exit': 'GTC' } # ==================== PARÁMETROS HIPEROPTIMIZABLES ==================== # Parámetros de consciencia cuántica consciousness_threshold = DecimalParameter(0.65, 0.85, default=0.72, decimals=2, space="buy", optimize=True) coherence_threshold = DecimalParameter(0.60, 0.80, default=0.68, decimals=2, space="buy", optimize=True) lambda_sensitivity = DecimalParameter(0.8, 1.5, default=1.0, decimals=1, space="buy", optimize=True) # Parámetros de análisis técnico rsi_entry = IntParameter(25, 45, default=35, space="buy", optimize=True) rsi_exit = IntParameter(65, 85, default=75, space="sell", optimize=True) macd_fast = IntParameter(8, 15, default=12, space="buy", optimize=True) macd_slow = IntParameter(20, 30, default=26, space="buy", optimize=True) macd_signal = IntParameter(7, 12, default=9, space="buy", optimize=True) # Parámetros de Fibonacci y Golden Ratio fib_level_entry = CategoricalParameter([0.236, 0.382, 0.618], default=0.382, space="buy", optimize=True) fib_level_exit = CategoricalParameter([1.0, 1.272, 1.618], default=1.272, space="sell", optimize=True) # Parámetros de volumen cuántico volume_multiplier = DecimalParameter(1.2, 3.0, default=1.8, decimals=1, space="buy", optimize=True) # ==================== CONFIGURACIÓN QBTC ==================== # URLs de servicios QBTC 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', } # Constantes herméticas PHI = 1.618033988749 # Proporción Áurea LAMBDA_RESONANCE = 7919 # Frecuencia de resonancia cuántica # Arquetipos de Tarot para símbolos principales SYMBOL_ARCHETYPES = { 'BTC/USDT': {'tarot': 'The Emperor', 'element': 'Fire', 'tier': 1, 'consciousness_req': 0.70}, 'ETH/USDT': {'tarot': 'The High Priestess', 'element': 'Water', 'tier': 1, 'consciousness_req': 0.65}, 'BNB/USDT': {'tarot': 'The Empress', 'element': 'Earth', 'tier': 1, 'consciousness_req': 0.60}, 'SOL/USDT': {'tarot': 'The Sun', 'element': 'Fire', 'tier': 2, 'consciousness_req': 0.62}, 'XRP/USDT': {'tarot': 'The Star', 'element': 'Water', 'tier': 2, 'consciousness_req': 0.58}, 'DOGE/USDT': {'tarot': 'The Fool', 'element': 'Air', 'tier': 2, 'consciousness_req': 0.55}, } # Configuración de tiers por símbolo SYMBOL_TIERS = { 1: ['BTC/USDT', 'ETH/USDT', 'BNB/USDT'], # Rey Solar, Reina Lunar, Venus 2: ['SOL/USDT', 'XRP/USDT', 'DOGE/USDT', 'ADA/USDT', 'AVAX/USDT', 'DOT/USDT', 'LINK/USDT', 'MATIC/USDT', 'LTC/USDT', 'BCH/USDT', 'ATOM/USDT', 'NEAR/USDT'], 3: ['UNI/USDT', 'FIL/USDT', 'TRX/USDT', 'ETC/USDT', 'XLM/USDT', 'ICP/USDT', 'VET/USDT', 'FTM/USDT', 'ALGO/USDT', 'SAND/USDT', 'MANA/USDT', 'AXS/USDT'], 4: ['APT/USDT', 'SUI/USDT', 'ARB/USDT', 'OP/USDT', 'INJ/USDT', 'STX/USDT'], 5: ['CRV/USDT', 'LRC/USDT', 'ENJ/USDT', 'CHZ/USDT', 'BAT/USDT', 'ZRX/USDT'], 6: ['APE/USDT', 'GALA/USDT', 'GME/USDT', 'IMX/USDT', 'FLOW/USDT'] } # ==================== CONFIGURACIÓN DE INDICADORES ==================== process_only_new_candles = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # ==================== MÉTODOS AUXILIARES QBTC ==================== @lru_cache(maxsize=128) def get_qbtc_service_data(self, service: str, endpoint: str = '/health') -> Optional[Dict]: """ Obtiene datos de los servicios QBTC con cache para optimización """ try: url = f"{self.QBTC_URLS.get(service, '')}{endpoint}" response = requests.get(url, timeout=2) if response.status_code == 200: return response.json() except Exception as e: # Log error but don't break strategy pass return None def calculate_lunar_phase_multiplier(self) -> float: """ Calcula el multiplicador basado en la fase lunar actual Implementa el ciclo hermético de 29.53 días """ now = datetime.now(timezone.utc) # Referencia: Luna nueva del 1 de enero de 2024 new_moon_ref = datetime(2024, 1, 1, tzinfo=timezone.utc) days_since_new_moon = (now - new_moon_ref).days % 29.53 # Mapeo de fases lunares a multiplicadores if days_since_new_moon < 7.38: # Luna creciente return 1.05 + (days_since_new_moon / 7.38) * 0.07 # 1.05 a 1.12 elif days_since_new_moon < 14.77: # Primer cuarto a luna llena return 1.12 + ((days_since_new_moon - 7.38) / 7.38) * 0.22 # 1.12 a 1.34 elif days_since_new_moon < 22.15: # Luna menguante return 1.34 - ((days_since_new_moon - 14.77) / 7.38) * 0.13 # 1.34 a 1.21 else: # Último cuarto return 1.21 - ((days_since_new_moon - 22.15) / 7.38) * 0.13 # 1.21 a 1.08 def calculate_fibonacci_levels(self, high: float, low: float) -> Dict[str, float]: """ Calcula niveles de Fibonacci para el análisis hermético """ diff = high - low return { 'fib_236': high - (diff * 0.236), 'fib_382': high - (diff * 0.382), 'fib_500': high - (diff * 0.500), 'fib_618': high - (diff * 0.618), 'fib_786': high - (diff * 0.786), 'extension_127': high + (diff * 0.272), 'extension_162': high + (diff * 0.618), } def calculate_quantum_coherence(self, dataframe: DataFrame) -> pd.Series: """ Calcula la coherencia cuántica basada en múltiples indicadores """ # Coherencia basada en correlaciones multi-timeframe rsi_coherence = 1 - abs(dataframe['rsi'] - 50) / 50 macd_coherence = np.where( dataframe['macd'] * dataframe['macdsignal'] > 0, 0.8, 0.2 ) volume_coherence = np.minimum(dataframe['volume_ratio'], 2.0) / 2.0 # Promedio ponderado de coherencias coherence = (rsi_coherence * 0.4 + macd_coherence * 0.3 + volume_coherence * 0.3) # Suavizado con EMA return ta.EMA(coherence, timeperiod=21) def calculate_quantum_entanglement(self, dataframe: DataFrame) -> pd.Series: """ Calcula el entrelazamiento cuántico entre indicadores """ # Correlación entre precio y volumen price_volume_corr = dataframe['close'].rolling(21).corr(dataframe['volume']) # Correlación entre múltiples timeframes (simulada) fast_ma = ta.EMA(dataframe['close'], timeperiod=8) slow_ma = ta.EMA(dataframe['close'], timeperiod=21) ma_sync = 1 - abs(fast_ma - slow_ma) / slow_ma # Entrelazamiento combinado entanglement = (abs(price_volume_corr) * 0.6 + ma_sync * 0.4) return entanglement.fillna(0.5) def get_symbol_tier_multiplier(self, pair: str) -> float: """ Obtiene el multiplicador basado en el tier del símbolo """ for tier, symbols in self.SYMBOL_TIERS.items(): if pair in symbols: multipliers = {1: 1.0, 2: 1.1, 3: 1.2, 4: 1.3, 5: 1.15, 6: 1.4} return multipliers.get(tier, 1.0) return 1.0 # ==================== TIMEFRAMES INFORMATIVOS ==================== def informative_pairs(self): """ Define pares informativos para análisis multi-timeframe """ pairs = self.dp.current_whitelist() informative_pairs = [] # Agregar timeframes informativos para cada par for pair in pairs[:20]: # Limitar 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 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 @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) macd_1h = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9) dataframe['macd_1h'] = macd_1h['macd'] dataframe['macdsignal_1h'] = macd_1h['macdsignal'] 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['sma200_4h'] = ta.SMA(dataframe, timeperiod=200) # Bandas de Bollinger para contexto de 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'] 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['rsi_1d'] = ta.RSI(dataframe, timeperiod=14) # ATR para cálculo de stop loss dinámico dataframe['atr_1d'] = ta.ATR(dataframe, timeperiod=14) return dataframe # ==================== INDICADORES PRINCIPALES ==================== def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Población de indicadores principales con lógica cuántica y hermética """ # ========== INDICADORES BÁSICOS ========== # RSI con diferentes 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) # MACD con parámetros optimizables 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'] # Medias móviles con números de Fibonacci dataframe['ema8'] = ta.EMA(dataframe, timeperiod=8) dataframe['ema13'] = ta.EMA(dataframe, timeperiod=13) dataframe['ema21'] = ta.EMA(dataframe, timeperiod=21) dataframe['ema34'] = ta.EMA(dataframe, timeperiod=34) dataframe['ema55'] = ta.EMA(dataframe, timeperiod=55) dataframe['ema89'] = ta.EMA(dataframe, timeperiod=89) # SMA para contexto a largo plazo dataframe['sma200'] = ta.SMA(dataframe, timeperiod=200) # ========== BANDAS DE BOLLINGER ========== bollinger = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=2) dataframe['bb_lower'] = bollinger['lower'] dataframe['bb_middle'] = bollinger['mid'] dataframe['bb_upper'] = 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']) # ========== INDICADORES DE VOLUMEN ========== # Volumen relativo dataframe['volume_mean'] = dataframe['volume'].rolling(20).mean() dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_mean'] # Money Flow Index dataframe['mfi'] = ta.MFI(dataframe, timeperiod=14) # ========== INDICADORES CUÁNTICOS PERSONALIZADOS ========== # Coherencia cuántica dataframe['quantum_coherence'] = self.calculate_quantum_coherence(dataframe) # Entrelazamiento cuántico dataframe['quantum_entanglement'] = self.calculate_quantum_entanglement(dataframe) # Momentum cuántico (basado en múltiples timeframes) dataframe['quantum_momentum'] = ( (dataframe['close'] / dataframe['ema21'] - 1) * 0.4 + (dataframe['rsi'] - 50) / 50 * 0.3 + np.where(dataframe['macd'] > dataframe['macdsignal'], 1, -1) * 0.3 ) # ========== NIVELES DE FIBONACCI DINÁMICOS ========== # Cálculo de máximos y mínimos para Fibonacci dataframe['high_55'] = dataframe['high'].rolling(55).max() dataframe['low_55'] = dataframe['low'].rolling(55).min() # Niveles de Fibonacci dataframe['fib_236'] = dataframe['high_55'] - ((dataframe['high_55'] - dataframe['low_55']) * 0.236) dataframe['fib_382'] = dataframe['high_55'] - ((dataframe['high_55'] - dataframe['low_55']) * 0.382) dataframe['fib_618'] = dataframe['high_55'] - ((dataframe['high_55'] - dataframe['low_55']) * 0.618) # ========== INDICADORES DE TENDENCIA ========== # ADX para fuerza de tendencia dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) # Parabolic SAR dataframe['sar'] = ta.SAR(dataframe, acceleration=0.02, maximum=0.2) # ========== PATRONES DE VELAS ========== # Patrones alcistas dataframe['hammer'] = ta.CDLHAMMER(dataframe) dataframe['morning_star'] = ta.CDLMORNINGSTAR(dataframe) dataframe['engulfing_bull'] = ta.CDLENGULFING(dataframe) # Patrones bajistas dataframe['shooting_star'] = ta.CDLSHOOTINGSTAR(dataframe) dataframe['evening_star'] = ta.CDLEVENINGSTAR(dataframe) dataframe['hanging_man'] = ta.CDLHANGINGMAN(dataframe) # ========== MÉTRICAS DE RIESGO ========== # ATR para stop loss dinámico dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['atr_percent'] = dataframe['atr'] / dataframe['close'] # Volatilidad realizada dataframe['volatility'] = dataframe['close'].pct_change().rolling(20).std() * np.sqrt(252) # ========== INTEGRACIÓN CON SERVICIOS QBTC ========== # Multiplicador lunar (actualizado una vez por día) dataframe['lunar_multiplier'] = self.calculate_lunar_phase_multiplier() # Multiplicador por tier del símbolo pair = metadata.get('pair', '') dataframe['tier_multiplier'] = self.get_symbol_tier_multiplier(pair) # ========== SEÑALES COMPUESTAS ========== # Señal de tendencia multi-timeframe dataframe['trend_signal'] = ( np.where(dataframe['ema21'] > dataframe['ema55'], 1, -1) * 0.5 + np.where(dataframe['close'] > dataframe['sma200'], 1, -1) * 0.3 + np.where(dataframe['adx'] > 25, 1, 0) * 0.2 ) # Señal de momentum dataframe['momentum_signal'] = ( np.where(dataframe['rsi'] > 50, 1, -1) * 0.4 + np.where(dataframe['macd'] > dataframe['macdsignal'], 1, -1) * 0.6 ) return dataframe # ==================== LÓGICA DE ENTRADA ==================== def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Lógica de entrada basada en confluencias cuánticas y herméticas """ # Obtener datos de servicios QBTC (con fallback si no están disponibles) metaconsciencia_data = self.get_qbtc_service_data('metaconsciencia', '/status') guardian_data = self.get_qbtc_service_data('guardian', '/health') # Valores por defecto si los servicios no están disponibles current_consciousness = 0.75 # Valor por defecto system_health = True if metaconsciencia_data: current_consciousness = metaconsciencia_data.get('consciousness_level', 0.75) if guardian_data: system_health = guardian_data.get('status') == 'healthy' # ========== CONDICIONES DE ENTRADA PRINCIPAL ========== conditions_entry = [] # 1. Consciencia cuántica suficiente consciousness_ok = current_consciousness >= self.consciousness_threshold.value # 2. Coherencia cuántica alta coherence_condition = dataframe['quantum_coherence'] >= self.coherence_threshold.value conditions_entry.append(coherence_condition) # 3. RSI en zona de entrada rsi_condition = (dataframe['rsi'] <= self.rsi_entry.value) & (dataframe['rsi'] >= 25) conditions_entry.append(rsi_condition) # 4. MACD alcista macd_condition = ( (dataframe['macd'] > dataframe['macdsignal']) | (qtpylib.crossed_above(dataframe['macd'], dataframe['macdsignal'])) ) conditions_entry.append(macd_condition) # 5. Tendencia alcista multi-timeframe trend_condition = ( (dataframe['ema21'] > dataframe['ema55']) & (dataframe['close'] > dataframe['ema21']) & (dataframe['trend_signal'] > 0.2) ) conditions_entry.append(trend_condition) # 6. Volumen significativo volume_condition = dataframe['volume_ratio'] >= self.volume_multiplier.value conditions_entry.append(volume_condition) # 7. Confluencia de Fibonacci (precio cerca de nivel de soporte) fib_condition = ( (dataframe['close'] <= dataframe['fib_382'] * 1.02) | (dataframe['close'] <= dataframe['fib_236'] * 1.02) ) conditions_entry.append(fib_condition) # 8. Momentum cuántico positivo quantum_momentum_condition = dataframe['quantum_momentum'] > 0.1 conditions_entry.append(quantum_momentum_condition) # 9. No en sobrecompra not_overbought = ( (dataframe['rsi'] < 80) & (dataframe['mfi'] < 85) & (dataframe['bb_percent'] < 0.9) ) conditions_entry.append(not_overbought) # 10. Patrones de velas alcistas bullish_patterns = ( (dataframe['hammer'] > 0) | (dataframe['morning_star'] > 0) | (dataframe['engulfing_bull'] > 0) ) # Patrones son opcionales, no obligatorios # ========== CONDICIONES ESPECIALES POR TIER ========== pair = metadata.get('pair', '') tier_multiplier = self.get_symbol_tier_multiplier(pair) # Ajustar condiciones según el tier del símbolo if tier_multiplier >= 1.3: # Tiers altos (más volátiles) # Requiere mayor confluencia para símbolos de alto riesgo min_conditions = 7 else: min_conditions = 6 # ========== APLICACIÓN FINAL DE CONDICIONES ========== # Combinar todas las condiciones principales main_conditions = conditions_entry[0] # Comenzar con coherencia for condition in conditions_entry[1:]: main_conditions = main_conditions & condition # Condiciones adicionales additional_conditions = ( (dataframe['volume'] > 0) & # Volumen mínimo system_health & # Estado del sistema consciousness_ok # Consciencia suficiente ) # ========== APLICACIÓN DE MULTIPLICADORES LUNARES ========== # Durante fases lunares favorables, relajar ligeramente las condiciones lunar_boost = dataframe['lunar_multiplier'] > 1.2 final_entry_condition = ( main_conditions & additional_conditions ) | ( # Condiciones relajadas durante luna favorable lunar_boost & coherence_condition & trend_condition & (dataframe['rsi'] <= self.rsi_entry.value + 5) & macd_condition & volume_condition ) dataframe.loc[final_entry_condition, 'enter_long'] = 1 return dataframe # ==================== LÓGICA DE SALIDA ==================== def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Lógica de salida con protección cuántica y toma de beneficios hermética """ # ========== CONDICIONES DE SALIDA PRINCIPAL ========== exit_conditions = [] # 1. RSI en sobrecompra rsi_exit_condition = dataframe['rsi'] >= self.rsi_exit.value exit_conditions.append(rsi_exit_condition) # 2. MACD bajista macd_exit_condition = ( (dataframe['macd'] < dataframe['macdsignal']) & (qtpylib.crossed_below(dataframe['macd'], dataframe['macdsignal'])) ) exit_conditions.append(macd_exit_condition) # 3. Ruptura de tendencia trend_break_condition = ( (dataframe['close'] < dataframe['ema21']) & (dataframe['ema21'] < dataframe['ema55']) & (dataframe['trend_signal'] < -0.2) ) exit_conditions.append(trend_break_condition) # 4. Toma de beneficios en Fibonacci fib_profit_condition = ( (dataframe['close'] >= dataframe['fib_618'] * 1.02) | (dataframe['close'] >= dataframe['fib_382'] * 1.15) ) exit_conditions.append(fib_profit_condition) # 5. Divergencia de volumen volume_divergence = ( (dataframe['volume_ratio'] < 0.7) & (dataframe['close'] > dataframe['close'].shift(5)) ) exit_conditions.append(volume_divergence) # 6. Pérdida de coherencia cuántica coherence_loss = dataframe['quantum_coherence'] < (self.coherence_threshold.value - 0.1) exit_conditions.append(coherence_loss) # 7. Momentum cuántico negativo negative_momentum = dataframe['quantum_momentum'] < -0.15 exit_conditions.append(negative_momentum) # 8. Patrones bajistas bearish_patterns = ( (dataframe['shooting_star'] < 0) | (dataframe['evening_star'] < 0) | (dataframe['hanging_man'] < 0) ) exit_conditions.append(bearish_patterns) # ========== SALIDAS DE EMERGENCIA ========== # Salida de emergencia por Guardian guardian_data = self.get_qbtc_service_data('guardian', '/risk') emergency_exit = False if guardian_data: risk_level = guardian_data.get('risk_level', 'LOW') emergency_exit = risk_level in ['HIGH', 'CRITICAL'] # ========== APLICACIÓN DE LÓGICA DE SALIDA ========== # Salida normal (cualquier condición de las principales) normal_exit = exit_conditions[0] for condition in exit_conditions[1:]: normal_exit = normal_exit | condition # Salida por múltiples confirmaciones multiple_confirmations = ( rsi_exit_condition & macd_exit_condition ) | ( trend_break_condition & coherence_loss ) | ( negative_momentum & bearish_patterns ) # Condición final de salida final_exit_condition = ( normal_exit | multiple_confirmations | emergency_exit ) dataframe.loc[final_exit_condition, 'exit_long'] = 1 return dataframe # ==================== GESTIÓN AVANZADA DE ÓRDENES ==================== def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Stop loss dinámico basado en ATR y coherencia cuántica """ # Obtener datos del dataframe actual dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) < 1: return self.stoploss last_candle = dataframe.iloc[-1] # ATR dinámico para stop loss atr_multiplier = 2.0 if hasattr(last_candle, 'quantum_coherence') and last_candle['quantum_coherence'] > 0.8: atr_multiplier = 1.5 # Stop más ajustado con alta coherencia elif hasattr(last_candle, 'quantum_coherence') and last_candle['quantum_coherence'] < 0.5: atr_multiplier = 3.0 # Stop más amplio con baja coherencia if hasattr(last_candle, 'atr_percent'): dynamic_stoploss = -last_candle['atr_percent'] * atr_multiplier # Limitar el stop loss dinámico dynamic_stoploss = max(dynamic_stoploss, -0.25) # Máximo 25% dynamic_stoploss = min(dynamic_stoploss, -0.05) # Mínimo 5% 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 con integración QBTC """ # Obtener datos de MetaConsciencia metaconsciencia_data = self.get_qbtc_service_data('metaconsciencia', '/decision') if metaconsciencia_data: decision = metaconsciencia_data.get('action', '') if decision == 'SELL_ALL' or decision == 'EMERGENCY_EXIT': return 'metaconsciencia_exit' # Lógica de toma de beneficios por tiempo trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 # en horas # Toma de beneficios escalonada if current_profit > 0.20 and trade_duration > 0.5: # 20% en 30 min return 'quick_profit_20' elif current_profit > 0.15 and trade_duration > 1: # 15% en 1 hora return 'profit_15_1h' elif current_profit > 0.10 and trade_duration > 2: # 10% en 2 horas return 'profit_10_2h' elif current_profit > 0.05 and trade_duration > 4: # 5% en 4 horas return 'profit_5_4h' return None # ==================== CONFIGURACIÓN DE PLOTS ==================== plot_config = { 'main_plot': { 'ema21': {'color': 'orange'}, 'ema55': {'color': 'cyan'}, 'sma200': {'color': 'white'}, 'bb_upper': {'color': 'gray'}, 'bb_lower': {'color': 'gray'}, 'bb_middle': {'color': 'lightgray'}, 'fib_382': {'color': 'yellow'}, 'fib_618': {'color': 'gold'}, }, 'subplots': { 'RSI': { 'rsi': {'color': 'red'}, 'rsi_fast': {'color': 'orange'}, 'rsi_slow': {'color': 'blue'}, }, 'MACD': { 'macd': {'color': 'blue'}, 'macdsignal': {'color': 'orange'}, 'macdhist': {'color': 'gray'}, }, 'Quantum Metrics': { 'quantum_coherence': {'color': 'purple'}, 'quantum_entanglement': {'color': 'magenta'}, 'quantum_momentum': {'color': 'cyan'}, }, 'Volume': { 'volume_ratio': {'color': 'green'}, }, 'Lunar & Tier': { 'lunar_multiplier': {'color': 'silver'}, 'tier_multiplier': {'color': 'gold'}, } } } # ==================== MÉTODOS DE INFORMACIÓN ==================== def version(self) -> str: """Retorna la versión de la estrategia""" return "QBTC Quantum Strategy v3.0" def bot_start(self, **kwargs) -> None: """ Método llamado al iniciar el bot - configuración inicial """ import logging logger = logging.getLogger(__name__) logger.info("QBTC Quantum Strategy v3.0 iniciando") logger.info(f"Configuracion de consciencia: {self.consciousness_threshold.value}") logger.info(f"Configuracion de coherencia: {self.coherence_threshold.value}") logger.info(f"URLs de servicios QBTC configuradas") # Verificar conectividad con servicios QBTC for service_name, url in self.QBTC_URLS.items(): try: response = requests.get(f"{url}/health", timeout=2) if response.status_code == 200: logger.info(f"{service_name} conectado: {url}") else: logger.warning(f"{service_name} no responde: {url}") except Exception: logger.warning(f"{service_name} no disponible: {url}") 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 """ # Timeout más largo para símbolos de tier alto (más volátiles) tier_multiplier = self.get_symbol_tier_multiplier(pair) base_timeout = 300 # 5 minutos base timeout_seconds = base_timeout * tier_multiplier 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 """ # Timeout más corto para salidas - queremos salir rápido return (current_time - order.order_date_utc).total_seconds() > 600 # 10 minutos 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 antes de ejecutar el trade con Guardian """ # Consultar Guardian para confirmación final guardian_data = self.get_qbtc_service_data('guardian', '/validate_trade') import logging logger = logging.getLogger(__name__) if guardian_data: if not guardian_data.get('approved', True): logger.info(f"Guardian rechazo el trade para {pair}: {guardian_data.get('reason', 'Unknown')}") return False # Verificar límites adicionales pair_tier = self.get_symbol_tier_multiplier(pair) # Límites por tier if pair_tier >= 1.4 and amount * rate > 1000: # Tier 6: límite de $1000 logger.info(f"Trade de {pair} excede limite para tier alto: ${amount * rate:.2f}") return False logger.info(f"Trade confirmado para {pair}: {amount} @ {rate}") 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 """ profit_percent = trade.calc_profit_ratio(rate) * 100 import logging logger = logging.getLogger(__name__) logger.info(f"Salida confirmada para {pair}: {exit_reason} - Profit: {profit_percent:.2f}%") # Siempre permitir salidas de emergencia if 'emergency' in exit_reason.lower() or 'guardian' in exit_reason.lower(): return True return True