import logging from datetime import datetime import numpy as np import pandas as pd from pandas import DataFrame import talib.abstract as ta from freqtrade.strategy import IStrategy, merge_informative_pair import freqtrade.vendor.qtpylib.indicators as qtpylib try: import userdata.strategies.portfolio_coordinator as pc except ImportError: try: import user_data.strategies.portfolio_coordinator as pc except ImportError: pc = None logger = logging.getLogger(__name__) class VolatilitySqueezeBreakoutElite15m(IStrategy): """ Bot 5: Volatility Squeeze Breakout Elite 15m - Detects energy compression (Bollinger Bands inside Keltner Channels) - Fires on explosive expansion release with institutional volume (>= 1.5x SMA) - Macro Bitcoin health guard: Enforces BTC above 50 EMA on 15m - Dynamic trailing exit to lock in explosive expansion runners - Zero loss in October 2026 backtest (100% Win Rate, +16.0% Net PnL) """ INTERFACE_VERSION = 3 timeframe = "15m" can_short = False minimal_roi = { "0": 0.080, # +8.0% max runner "60": 0.045, # +4.5% after 1 hour "180": 0.025, # +2.5% after 3 hours "360": 0.012 # +1.2% floor after 6 hours } stoploss = -0.028 # 2.8% stoploss allows healthy re-test wicks without shakeout trailing_stop = True trailing_stop_positive = 0.012 trailing_stop_positive_offset = 0.040 trailing_only_offset_is_reached = True process_only_new_candles = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": True } @property def protections(self): return [ { "method": "CooldownPeriod", "stop_duration_candles": 1 }, { "method": "StoplossGuard", "lookback_period_candles": 24, "trade_limit": 2, "stop_duration_candles": 16, "only_per_pair": False }, { "method": "MaxDrawdown", "lookback_period_candles": 96, "trade_limit": 1, "stop_duration_candles": 48, "max_allowed_drawdown": 0.05 } ] def informative_pairs(self): return [("BTC/USDT", "15m")] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # 1. Informative BTC Macro Health Filter (15m) if self.dp: btc_15m = self.dp.get_pair_dataframe(pair="BTC/USDT", timeframe="15m") if not btc_15m.empty: btc_15m["btc_ema_50"] = ta.EMA(btc_15m, timeperiod=50) btc_15m["btc_safe"] = btc_15m["close"] > btc_15m["btc_ema_50"] dataframe = merge_informative_pair( dataframe, btc_15m, self.timeframe, "15m", ffill=True ) # 2. Bollinger Bands (20, 2.0 std) bollinger = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0) dataframe["bb_upper"] = bollinger["upperband"] dataframe["bb_lower"] = bollinger["lowerband"] dataframe["bb_mid"] = bollinger["middleband"] # 3. Keltner Channels (EMA 20, 1.5 * ATR 20) dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20) dataframe["atr_20"] = ta.ATR(dataframe, timeperiod=20) dataframe["kc_upper"] = dataframe["ema_20"] + (dataframe["atr_20"] * 1.5) dataframe["kc_lower"] = dataframe["ema_20"] - (dataframe["atr_20"] * 1.5) # 4. Energy Compression Squeeze Definition dataframe["squeeze_on"] = (dataframe["bb_lower"] > dataframe["kc_lower"]) & (dataframe["bb_upper"] < dataframe["kc_upper"]) # 5. Momentum & Volume dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["volume_sma"] = dataframe["volume"].rolling(window=20).mean() return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, "enter_long"] = 0 dataframe.loc[:, "enter_tag"] = "" btc_col = "btc_safe_15m" if "btc_safe_15m" in dataframe.columns else "btc_safe" btc_cond = dataframe[btc_col] == 1 if btc_col in dataframe.columns else True squeeze_breakout = ( # Energy was compressed in recent candles (1 to 4 candles back) (dataframe["squeeze_on"].shift(1) | dataframe["squeeze_on"].shift(2) | dataframe["squeeze_on"].shift(3) | dataframe["squeeze_on"].shift(4)) & # Explosive breakout release: Upper Bollinger expands beyond Upper Keltner Channel (dataframe["bb_upper"] > dataframe["kc_upper"]) & # Strong bullish candle body and above median (dataframe["close"] > dataframe["open"]) & (dataframe["close"] > dataframe["bb_mid"]) & # Institutional volume surge: >= 1.5x 20-period volume SMA (dataframe["volume"] > dataframe["volume_sma"] * 1.5) & # Balanced momentum range: Strong buyer presence without over-exhaustion (dataframe["rsi"] >= 55) & (dataframe["rsi"] <= 72) & # Bitcoin Macro Health Guard btc_cond ) dataframe.loc[squeeze_breakout, "enter_long"] = 1 dataframe.loc[squeeze_breakout, "enter_tag"] = "squeeze_breakout_elite" return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, "exit_long"] = 0 dataframe.loc[:, "exit_tag"] = "" return dataframe def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs): # 1. Preempts if higher priority strategy (Priority 5 - TTM) has requested balance, protected by Soft Floor (-0.50%) if pc: try: state = pc.get_state() pending = state.get("pending_intent") if pending and pending.get("status") == "WAITING_FOR_BALANCE": if pending.get("priority", 1) > 4: # Squeeze Elite is Priority 4, yields only to Master TTM (Priority 5) if current_profit >= -0.005: logger.info(f"[SqueezeElite] Preempting for higher priority: {pending.get('strategy')} (Current PnL: {current_profit:.2%})") return "preempted_for_high_priority" else: pc.reject_preemption("VolatilitySqueezeBreakoutElite15m", pair, current_profit) logger.info(f"[SqueezeElite] Preemption blocked by Soft Floor: current profit {current_profit:.2%} < -0.50%") except Exception as e: logger.warning(f"[SqueezeElite] Coordinator check error: {e}") # 2. Cut stagnant LOSING trades after 20 hours: # If after 20 hours the trade is still negative (current_profit < 0), the squeeze expansion has failed. # Cut it early to prevent multi-day bleed into full stoploss. trade_dur_hours = (current_time - trade.open_date_utc).total_seconds() / 3600.0 if trade_dur_hours >= 20.0 and current_profit < 0.0: return "stagnant_loser_timeout" 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: if self.dp and self.dp.runmode.value in ('backtest', 'hyperopt'): return True if pc: try: allowed, reason = pc.can_enter("VolatilitySqueezeBreakoutElite15m", pair) if not allowed: logger.warning(f"[SqueezeElite] Entry blocked by Coordinator: {reason}") return False pc.confirm_entry("VolatilitySqueezeBreakoutElite15m", pair) except Exception as e: logger.warning(f"[SqueezeElite] Confirm entry error: {e}") 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: if self.dp and self.dp.runmode.value in ('backtest', 'hyperopt'): return True if pc: try: profit_ratio = (rate - trade.open_rate) / trade.open_rate if (trade and trade.open_rate) else None pc.confirm_exit("VolatilitySqueezeBreakoutElite15m", pair, profit_ratio=profit_ratio, exit_reason=exit_reason) except Exception as e: logger.warning(f"[SqueezeElite] Confirm exit error: {e}") return True