# source: https://raw.githubusercontent.com/vigoferrel/qbtc-unified/ef7ffc8f45cc7ab041884cd0c08a2043acc1f0cb/freqtrade_qbtc/strategies/QBTCMicroMacro.py """ Github_vigoferrel_qbtc_unified__QBTCMicroMacro__20251009_202533 - Operaciones Micro con Visión Macro =================================================== FILOSOFÍA CORRECTA: - EJECUCIÓN: Timeframe micro (3m) para entradas precisas - CONTEXTO: Análisis macro (1h, 4h, 1d) para filtrar direccionalidad - TRANSFORMACIONES: Prima→Markov en macro, Feynman en micro - VISIÓN: Surf micro waves riding macro tide FUNDAMENTOS: 1. Contexto Macro: Estados de mercado de largo plazo 2. Ejecución Micro: Timing preciso en movimientos cortos 3. Filtros de Coherencia: Solo trades alineados con macro 4. Probabilidades Feynman: Optimizar entradas micro Target: >65% win rate, operaciones rápidas alineadas con macro Creado: 2025-09-10 Autor: QBTC Trading System - Micro with Macro Vision """ import talib.abstract as ta import pandas as pd import numpy as np from pandas import DataFrame from freqtrade.strategy.interface import IStrategy from freqtrade.strategy import DecimalParameter, IntParameter, informative import logging from datetime import datetime class Github_vigoferrel_qbtc_unified__QBTCMicroMacro__20251009_202533(IStrategy): # MICRO EXECUTION SETUP INTERFACE_VERSION = 3 timeframe = '5m' # MICRO: Ejecución en 5 minutos can_short = False stoploss = -0.02 # 2% - Tight stop for micro trades # ROI micro - Quick profits minimal_roi = { "0": 0.04, # 4% immediate "5": 0.025, # 2.5% after 5min "15": 0.015, # 1.5% after 15min "45": 0.01, # 1% after 45min "120": 0.005 # 0.5% after 2h } # Micro trading configuration max_open_trades = 6 # Multiple micro positions stake_currency = 'USDT' stake_amount = 100 use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # Respect quick ROI # Fast execution orders order_types = { 'entry': 'market', 'exit': 'limit', 'stoploss': 'market', 'stoploss_on_exchange': True } order_time_in_force = { 'entry': 'gtc', 'exit': 'ioc' # Immediate fills for micro } # MICRO PARAMETERS micro_rsi_oversold = IntParameter(25, 40, default=32, space="buy") micro_rsi_overbought = IntParameter(65, 85, default=75, space="buy") volume_spike_micro = DecimalParameter(1.5, 3.0, default=2.0, space="buy") # MACRO FILTER PARAMETERS macro_trend_strength = DecimalParameter(0.02, 0.08, default=0.04, space="buy") macro_alignment_threshold = DecimalParameter(0.6, 0.9, default=0.75, space="buy") # EXIT PARAMETERS micro_profit_target = DecimalParameter(0.015, 0.04, default=0.025, space="sell") momentum_loss_threshold = DecimalParameter(-0.3, -0.1, default=-0.2, space="sell") def __init__(self, config: dict) -> None: super().__init__(config) self.logger = logging.getLogger(__name__) # Macro context tracking self._macro_trend = 'NEUTRAL' self._macro_strength = 0.0 self._macro_confidence = 0.0 # Micro execution stats self._micro_stats = { 'micro_aligned': 0, 'micro_counter': 0, 'quick_exits': 0, 'roi_hits': 0 } self.logger.info("🎯 MICRO-MACRO STRATEGY: Surf micro waves riding macro tide") self.logger.info("⚡ Execution: 5m micro | Context: Macro filter") def informative_pairs(self): pairs = self.dp.current_whitelist() informative_pairs = [] # Add macro timeframes for context for pair in pairs: informative_pairs.extend([ (pair, '1h'), # Macro context (pair, '4h'), # Macro structure (pair, '1d') # Macro cycle ]) return informative_pairs @informative('1h') def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """MACRO CONTEXT - 1H""" # Macro trend identification dataframe['ema_21'] = ta.EMA(dataframe, timeperiod=21) dataframe['ema_50'] = ta.EMA(dataframe, timeperiod=50) dataframe['ema_100'] = ta.EMA(dataframe, timeperiod=100) # Macro momentum dataframe['macro_momentum'] = (dataframe['close'] - dataframe['close'].shift(24)) / dataframe['close'].shift(24) # Macro RSI for overall direction dataframe['rsi_macro'] = ta.RSI(dataframe, timeperiod=14) # Macro volume profile dataframe['volume_profile'] = dataframe['volume'] / ta.SMA(dataframe['volume'], timeperiod=48) # Macro trend strength dataframe['trend_strength_macro'] = abs(dataframe['ema_21'] - dataframe['ema_50']) / dataframe['ema_50'] return dataframe @informative('4h') def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """MACRO STRUCTURE - 4H""" # Structural levels dataframe['high_structure'] = dataframe['high'].rolling(window=24).max() # 4 days dataframe['low_structure'] = dataframe['low'].rolling(window=24).min() # 4 days # Structural EMAs dataframe['ema_structure'] = ta.EMA(dataframe, timeperiod=12) # 2 days # Structural momentum dataframe['structure_momentum'] = (dataframe['close'] - dataframe['close'].shift(6)) / dataframe['close'].shift(6) # Market regime detection dataframe['volatility_4h'] = dataframe['close'].pct_change().rolling(window=12).std() dataframe['regime'] = 'NORMAL' # High volatility regime vol_threshold = dataframe['volatility_4h'].quantile(0.8) dataframe.loc[dataframe['volatility_4h'] > vol_threshold, 'regime'] = 'HIGH_VOL' # Low volatility regime vol_threshold_low = dataframe['volatility_4h'].quantile(0.2) dataframe.loc[dataframe['volatility_4h'] < vol_threshold_low, 'regime'] = 'LOW_VOL' return dataframe @informative('1d') def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """MACRO CYCLE - 1D""" # Daily cycle position dataframe['cycle_ema'] = ta.EMA(dataframe, timeperiod=7) # 1 week dataframe['cycle_position'] = dataframe['close'] / dataframe['cycle_ema'] # Long-term momentum dataframe['cycle_momentum'] = (dataframe['close'] - dataframe['close'].shift(7)) / dataframe['close'].shift(7) # Macro cycle strength dataframe['cycle_strength'] = abs(dataframe['cycle_momentum']) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """MICRO INDICATORS - 3M execution""" pair = metadata['pair'] # === MICRO RSI === dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['rsi_fast'] = ta.RSI(dataframe, timeperiod=7) # === MICRO EMAs === dataframe['ema_8'] = ta.EMA(dataframe, timeperiod=8) dataframe['ema_21'] = ta.EMA(dataframe, timeperiod=21) dataframe['ema_50'] = ta.EMA(dataframe, timeperiod=50) # === MICRO MACD === macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] dataframe['macdhist'] = macd['macdhist'] # === MICRO VOLUME === dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20) dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] # === MICRO BOLLINGER BANDS === bollinger = ta.BBANDS(dataframe, timeperiod=20) dataframe['bb_lower'] = bollinger['lowerband'] dataframe['bb_middle'] = bollinger['middleband'] dataframe['bb_upper'] = bollinger['upperband'] dataframe['bb_percent'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower']) # === MICRO MOMENTUM === dataframe['momentum_5m'] = (dataframe['close'] - dataframe['close'].shift(5)) / dataframe['close'].shift(5) dataframe['acceleration'] = dataframe['momentum_5m'].diff(3) # === MICRO PRICE ACTION === dataframe['candle_size'] = abs(dataframe['close'] - dataframe['open']) / dataframe['open'] dataframe['upper_shadow'] = (dataframe['high'] - dataframe[['close', 'open']].max(axis=1)) / dataframe['open'] dataframe['lower_shadow'] = (dataframe[['close', 'open']].min(axis=1) - dataframe['low']) / dataframe['open'] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ MICRO ENTRIES WITH MACRO FILTER Only trade micro setups aligned with macro trend """ pair = metadata['pair'] # Get macro context informative_1h = self.dp.get_pair_dataframe(pair=pair, timeframe='1h') informative_4h = self.dp.get_pair_dataframe(pair=pair, timeframe='4h') informative_1d = self.dp.get_pair_dataframe(pair=pair, timeframe='1d') # Merge macro context dataframe = merge_informative_pair(dataframe, informative_1h, self.timeframe, '1h', ffill=True) dataframe = merge_informative_pair(dataframe, informative_4h, self.timeframe, '4h', ffill=True) dataframe = merge_informative_pair(dataframe, informative_1d, self.timeframe, '1d', ffill=True) # Initialize dataframe.loc[:, 'enter_long'] = 0 dataframe.loc[:, 'enter_tag'] = '' # === MACRO ALIGNMENT CHECK === macro_bullish = ( (dataframe['ema_21_1h'] > dataframe['ema_50_1h']) & # 1H uptrend (dataframe['macro_momentum_1h'] > 0.01) & # Positive 1H momentum (dataframe['structure_momentum_4h'] > -0.02) & # 4H not breaking down (dataframe['cycle_momentum_1d'] > -0.05) & # 1D not in major decline (dataframe['close'] > dataframe['ema_structure_4h']) # Above 4H structure ) # === MICRO ENTRY 1: OVERSOLD BOUNCE (Macro Aligned) === oversold_micro = ( (dataframe['rsi'] <= self.micro_rsi_oversold.value) & # Micro oversold (dataframe['volume_ratio'] >= self.volume_spike_micro.value) & # Volume spike (dataframe['bb_percent'] <= 0.2) & # Near lower BB (dataframe['close'] > dataframe['open']) & # Green candle (dataframe['macd'] > dataframe['macdsignal'].shift(1)) & # MACD improving macro_bullish # MACRO FILTER ) dataframe.loc[oversold_micro, 'enter_long'] = 1 dataframe.loc[oversold_micro, 'enter_tag'] = 'micro_oversold_aligned' # === MICRO ENTRY 2: MOMENTUM BREAKOUT (Macro Aligned) === momentum_breakout = ( (dataframe['close'] > dataframe['ema_21']) & # Above micro trend (dataframe['rsi_fast'] > 50) & (dataframe['rsi_fast'] < 78) & # Good RSI range (dataframe['momentum_5m'] > 0.003) & # Positive micro momentum (dataframe['acceleration'] > 0) & # Accelerating (dataframe['volume_ratio'] >= 1.5) & # Volume confirmation (dataframe['candle_size'] > 0.002) & # Meaningful candle macro_bullish # MACRO FILTER ) # Don't overwrite oversold breakout_mask = momentum_breakout & (dataframe['enter_long'] == 0) dataframe.loc[breakout_mask, 'enter_long'] = 1 dataframe.loc[breakout_mask, 'enter_tag'] = 'micro_momentum_aligned' # === MICRO ENTRY 3: PULLBACK TO SUPPORT (Macro Aligned) === pullback_support = ( (dataframe['close'] <= dataframe['ema_21']) & # At micro support (dataframe['close'] > dataframe['ema_21'] * 0.998) & # Not far below (dataframe['rsi'] > 35) & (dataframe['rsi'] < 60) & # Not oversold/overbought (dataframe['bb_percent'] > 0.3) & (dataframe['bb_percent'] < 0.7) & # Mid-range (dataframe['lower_shadow'] > 0.001) & # Long lower shadow (dataframe['trend_strength_macro_1h'] > self.macro_trend_strength.value) & # Strong macro trend macro_bullish # MACRO FILTER ) # Don't overwrite other signals pullback_mask = pullback_support & (dataframe['enter_long'] == 0) dataframe.loc[pullback_mask, 'enter_long'] = 1 dataframe.loc[pullback_mask, 'enter_tag'] = 'micro_pullback_aligned' # Stats logging if oversold_micro.any(): self._micro_stats['micro_aligned'] += 1 self.logger.info(f"🎯 MICRO OVERSOLD ALIGNED: {pair} | RSI={dataframe['rsi'].iloc[-1]:.1f} | Macro: BULLISH") if breakout_mask.any(): self._micro_stats['micro_aligned'] += 1 self.logger.info(f"⚡ MICRO MOMENTUM ALIGNED: {pair} | Mom={dataframe['momentum_5m'].iloc[-1]:.3f} | Macro: BULLISH") if pullback_mask.any(): self._micro_stats['micro_aligned'] += 1 self.logger.info(f"🏄 MICRO PULLBACK ALIGNED: {pair} | Price={dataframe['close'].iloc[-1]:.4f} | Macro: STRONG") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ MICRO EXITS - Quick and precise """ dataframe.loc[:, 'exit_long'] = 0 dataframe.loc[:, 'exit_tag'] = '' # === MICRO EXIT 1: Overbought + Momentum Loss === overbought_exit = ( (dataframe['rsi'] >= self.micro_rsi_overbought.value) & # Micro overbought (dataframe['rsi_fast'] < dataframe['rsi_fast'].shift(1)) & # RSI declining (dataframe['acceleration'] < self.momentum_loss_threshold.value) & # Momentum loss (dataframe['upper_shadow'] > 0.002) # Long upper shadow ) dataframe.loc[overbought_exit, 'exit_long'] = 1 dataframe.loc[overbought_exit, 'exit_tag'] = 'micro_overbought' # === MICRO EXIT 2: Volume Exhaustion === volume_exhaustion = ( (dataframe['volume_ratio'] < 0.6) & # Low volume (dataframe['bb_percent'] > 0.8) & # Near upper BB (dataframe['momentum_5m'] < 0.001) & # Momentum fading (dataframe['close'] < dataframe['close'].shift(1)) # Price declining ) volume_mask = volume_exhaustion & (dataframe['exit_long'] == 0) dataframe.loc[volume_mask, 'exit_long'] = 1 dataframe.loc[volume_mask, 'exit_tag'] = 'micro_exhaustion' # === MICRO EXIT 3: Support Break === support_break = ( (dataframe['close'] < dataframe['ema_21']) & # Below micro support (dataframe['ema_21'] < dataframe['ema_21'].shift(2)) & # Support declining (dataframe['volume_ratio'] > 1.3) & # Volume on break (dataframe['candle_size'] > 0.002) # Meaningful break ) break_mask = support_break & (dataframe['exit_long'] == 0) dataframe.loc[break_mask, 'exit_long'] = 1 dataframe.loc[break_mask, 'exit_tag'] = 'micro_support_break' return dataframe def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> tuple: """ MICRO PROFIT MANAGEMENT """ # Quick partial exit at micro target if current_profit >= self.micro_profit_target.value: self._micro_stats['quick_exits'] += 1 self.logger.info(f"⚡ MICRO QUICK EXIT: {pair} at {current_profit:.2%}") return 0.6, 'micro_quick_profit' # Exit 60% return None def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: str, side: str, **kwargs) -> bool: """ MICRO-MACRO ALIGNMENT CONFIRMATION """ # Log alignment stats total_entries = sum(self._micro_stats.values()) if total_entries > 0 and total_entries % 20 == 0: alignment_rate = (self._micro_stats['micro_aligned'] / total_entries) * 100 self.logger.info(f"🎯 MACRO ALIGNMENT RATE: {alignment_rate:.1f}% | " f"Quick Exits: {self._micro_stats['quick_exits']}") self.logger.info(f"🎯 MICRO-MACRO ENTRY: {pair} | {entry_tag} | ${rate:.4f}") return True def merge_informative_pair(dataframe, informative, timeframe, timeframe_inf, ffill=True): """Helper function to merge informative pairs""" try: from freqtrade.strategy.merge_informative_pair import merge_informative_pair as mip return mip(dataframe, informative, timeframe, timeframe_inf, ffill=ffill) except ImportError: # Fallback manual merge informative['date'] = pd.to_datetime(informative['date']) dataframe['date'] = pd.to_datetime(dataframe['date']) # Add suffix to informative columns informative_cols = {} for col in informative.columns: if col != 'date': informative_cols[col] = f"{col}_{timeframe_inf}" informative = informative.rename(columns=informative_cols) # Merge dataframe = pd.merge(dataframe, informative, on='date', how='left') if ffill: for col in informative_cols.values(): dataframe[col] = dataframe[col].fillna(method='ffill') return dataframe