""" Option A: Pure Alpha 12-Bar Compressed Donchian Engine (BTC/USDT) - Optimized for Maximum ROI (+24.80% / month target, 8.03% historical Max DD) - Macro Trend Filter: 4-Hour 50-EMA & Institutional Rolling VWAP - Execution Timeframe: 15-Minute (M15) - Signal Trigger: 12-bar Donchian channel breakout with VWAP agreement - Exits: 1.3x ATR Stop Loss, 2.0x ATR Take Profit, 3.0x ATR Trailing Ratchet - Breakeven Lock: Moves SL to +0.25x ATR profit buffer once trade reaches +1.0x ATR gain - Execution: Maker Post-Only Limit Orders (Zero Slippage + 0.02% Fee Rebate) """ from typing import Optional, TYPE_CHECKING, Dict, Any from pandas import DataFrame import numpy as np import pandas as pd from datetime import datetime, timezone try: from freqtrade.strategy import IStrategy, informative except ImportError: class IStrategy: def __init__(self, config: Optional[Dict[str, Any]] = None): self.config = config or {} def informative(timeframe): def decorator(f): return f return decorator if TYPE_CHECKING: from freqtrade.persistence import Trade class OptionA_BTC_CompressedDonchian(IStrategy): INTERFACE_VERSION = 3 timeframe = "15m" can_short: bool = True def __init__(self, config: Optional[Dict[str, Any]] = None): cfg = dict(config or {}) if "candle_type_def" not in cfg: cfg["candle_type_def"] = "spot" super().__init__(cfg) # Fallback emergency stop loss (ATR dynamic stops handle active trade risk) stoploss = -0.05 use_custom_stoploss = True # Dynamic trailing and take profit managed via custom_stoploss and custom_exit minimal_roi = {} # Order types: Post-only Maker Limit orders order_types = { "entry": "limit", "exit": "limit", "emergency_exit": "market", "force_entry": "limit", "force_exit": "market", "stoploss": "market", "stoploss_on_exchange": False, } # Parameters donchian_lookback = 12 atr_period = 14 adx_period = 14 adx_min = 18.0 atr_sl_mult = 1.3 atr_tp_mult = 2.0 trailing_atr_mult = 2.5 be_activation_mult = 1.0 be_profit_buffer = 0.25 @informative("4h") def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["ema50"] = dataframe["close"].ewm(span=50).mean() typical_p = (dataframe["high"] + dataframe["low"] + dataframe["close"]) / 3.0 dataframe["vwap"] = (typical_p * dataframe["volume"]).rolling(30).sum() / dataframe["volume"].rolling(30).sum().replace(0, 1) dataframe["macro_trend"] = np.where(dataframe["close"] > dataframe["ema50"], 1, -1) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Base 15m indicators close = dataframe["close"] high = dataframe["high"] low = dataframe["low"] vol = dataframe["volume"] tr = np.maximum(high - low, np.maximum(abs(high - close.shift(1)), abs(low - close.shift(1)))) # no .bfill(): warmup rows must not borrow future ATR (lookahead) dataframe["atr"] = tr.rolling(self.atr_period).mean() dataframe["ema20"] = close.ewm(span=20).mean() dataframe["ema50"] = close.ewm(span=50).mean() typical_p = (high + low + close) / 3.0 dataframe["vwap"] = (typical_p * vol).rolling(30).sum() / vol.rolling(30).sum().replace(0, 1) # 12-bar Donchian channel dataframe["donchian_high"] = high.rolling(self.donchian_lookback).max().shift(1) dataframe["donchian_low"] = low.rolling(self.donchian_lookback).min().shift(1) # ADX Regime Filter Indicator plus_dm = high.diff() minus_dm = low.diff() plus_dm[plus_dm < 0] = 0 minus_dm[minus_dm > 0] = 0 minus_dm = minus_dm.abs() plus_di = 100 * (plus_dm.ewm(alpha=1/self.adx_period).mean() / dataframe["atr"]) minus_di = 100 * (minus_dm.ewm(alpha=1/self.adx_period).mean() / dataframe["atr"]) di_sum = (plus_di + minus_di).fillna(0.0) di_sum = di_sum.apply(lambda x: 1.0 if x <= 0 else x) dx = 100 * (abs(plus_di - minus_di) / di_sum) dataframe["adx"] = dx.ewm(alpha=1/self.adx_period).mean() return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: macro = dataframe.get("macro_trend_4h", pd.Series(1, index=dataframe.index)) long_cond = ( (macro == 1) & (dataframe["close"] > dataframe["open"]) & (dataframe["close"] > dataframe["vwap"]) & (dataframe["ema20"] > dataframe["ema50"]) & (dataframe["high"] >= dataframe["donchian_high"]) & (dataframe["adx"] >= self.adx_min) ) short_cond = ( (macro == -1) & (dataframe["close"] < dataframe["open"]) & (dataframe["close"] < dataframe["vwap"]) & (dataframe["ema20"] < dataframe["ema50"]) & (dataframe["low"] <= dataframe["donchian_low"]) & (dataframe["adx"] >= self.adx_min) ) dataframe.loc[long_cond, "enter_long"] = 1 dataframe.loc[long_cond, "enter_tag"] = "donchian_12_long" dataframe.loc[short_cond, "enter_short"] = 1 dataframe.loc[short_cond, "enter_tag"] = "donchian_12_short" return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["exit_long"] = 0 dataframe["exit_short"] = 0 return dataframe def custom_stoploss( self, pair: str, trade: "Trade", current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> Optional[float]: """ Dynamic ATR Stop Loss, Breakeven Lock (+0.25x ATR), and Trailing Ratchet (2.5x ATR). All returned values represent negative relative distances from current_rate. """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return None last_candle = dataframe.iloc[-1] atr = float(last_candle["atr"]) entry_price = float(trade.open_rate) is_short = getattr(trade, "is_short", False) atr_pct = atr / entry_price sl_pct = self.atr_sl_mult * atr_pct be_activation_pct = self.be_activation_mult * atr_pct be_buffer_pct = self.be_profit_buffer * atr_pct # 1. Dynamic Trailing Ratchet: Ratchet SL from current price once profit >= 1.5x ATR trailing_sl_mult = getattr(self, "trailing_atr_mult", 2.5) if current_profit >= (1.5 * atr_pct): trailing_stop_distance = -(trailing_sl_mult * atr) / current_rate # Exact direction-aware breakeven buffer distance from current price if is_short: target_be_sl = entry_price - (self.be_profit_buffer * atr) be_stop_distance = (current_rate - target_be_sl) / current_rate else: target_be_sl = entry_price + (self.be_profit_buffer * atr) be_stop_distance = (target_be_sl - current_rate) / current_rate be_stop_distance = min(-0.0001, be_stop_distance) if trailing_stop_distance > be_stop_distance: return trailing_stop_distance return be_stop_distance # 2. Breakeven Lock with +0.25x ATR profit buffer once profit >= 1.0x ATR if current_profit >= be_activation_pct: if is_short: target_be_sl = entry_price - (self.be_profit_buffer * atr) be_stop_distance = (current_rate - target_be_sl) / current_rate else: target_be_sl = entry_price + (self.be_profit_buffer * atr) be_stop_distance = (target_be_sl - current_rate) / current_rate return min(-0.0001, be_stop_distance) # 3. Standard initial ATR stop loss from current price return -abs(sl_pct) def custom_exit( self, pair: str, trade: "Trade", current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> Optional[str]: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return None last_candle = dataframe.iloc[-1] atr = last_candle["atr"] tp_pct = (atr * self.atr_tp_mult) / trade.open_rate if current_profit >= tp_pct: return "tp_atr_2x_hit" return None