import numpy as np import pandas as pd import logging from datetime import datetime, timedelta, timezone from pandas import DataFrame from typing import Optional from freqtrade.enums import CandleType from freqtrade.strategy import IStrategy, IntParameter, RealParameter from freqtrade.strategy import stoploss_from_absolute from freqtrade.persistence import Trade logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # N-value constants (mirrors quantdinger strategy.py) # --------------------------------------------------------------------------- SPREAD_SPAN = 6 N_VALUE_SPAN = 10 # Regime enum REGIME_COMPRESSION = 0 REGIME_EXPANSION = 1 REGIME_TREND = 2 REGIME_RANGE = 3 def _ema_of(values): """EMAp of a list of values, matching the quantdinger _ema function.""" if not values: return 0.0 result = None n = float(len(values)) for v in values: v = float(v) if result is None: result = v else: result = 2.0 * v / (n + 1.0) + (n - 1.0) / (n + 1.0) * result return result or 0.0 class QuantDingerStrategy(IStrategy): """ Freqtrade port of the quantdinger live strategy. Original strategy: ~/agents/quantdinger/live/strategy.py Trading: USDT-M perpetual futures (long only) Timeframe: 5m (OKX does not support 10m; parameters scaled from 10m original) Exchange: OKX isolated futures Core approach: - N-value (custom ATR) for dynamic position sizing and stop placement - Market regime detection (compression/expansion/trend/range) with hysteresis - Regime-gated entries: breakout chase (trend), compression breakout, pullback reentry (range) - Priority-ordered exits: protective → trailing → lock-profit → breakeven → time """ INTERFACE_VERSION = 3 timeframe = "5m" can_short = False use_custom_stoploss = True process_only_new_candles = True # OKX limits 5m candles to 300 per request, 5 calls max = 1499 candles. # We use 1400 (passes validation: ceil(1401/300)=5) and pre-fetch the # remaining required history in bot_start() by bumping _startup_candle_count # on the exchange, so retention keeps 300 + 5000 = 5300 candles. startup_candle_count = 1400 # Bars per day for 5m candles: 24 * 60 / 5 = 288 _BARS_PER_DAY = 288 # ROI disabled — exits are driven entirely by custom_stoploss / custom_exit minimal_roi = {"0": 1.0} # Hard stop-loss floor; custom_stoploss tightens from here stoploss = -0.10 # Approximate cooldown (7 h × 12 candles/h at 5m = 84 candles) ignore_buying_expired_candle_after = 84 # ----------------------------------------------------------------------- # Strategy parameters (scaled from original 10m defaults to 5m ×2) # ----------------------------------------------------------------------- strategy_lever_rate = RealParameter(0.5, 3.0, default=1.0, space="buy", load=True) profit_line = RealParameter(0.02, 0.20, default=0.08, space="sell", load=True) lock_profit_rate = RealParameter(0.10, 0.50, default=0.33, space="sell", load=True) open_time_interval = RealParameter(1.0, 24.0, default=7.0, space="buy", load=True) up_line_span = IntParameter(200, 1600, default=1008, space="buy", load=True) up_line_offset = RealParameter(0.5, 3.0, default=1.8, space="buy", load=True) buy_stop_profit_span = IntParameter(100, 1000, default=480, space="sell", load=True) buy_stop_profit_offset = RealParameter(0.5, 2.0, default=1.04, space="sell", load=True) ma_span_long = IntParameter(1, 10, default=1, space="buy", load=True) regime_slope_lookback = IntParameter(20, 200, default=96, space="buy", load=True) regime_slope_threshold = RealParameter(0.001, 0.05, default=0.008, space="buy", load=True) regime_displace_threshold = RealParameter(0.05, 0.30, default=0.14, space="buy", load=True) regime_vol_ema_span = IntParameter(10, 200, default=48, space="buy", load=True) regime_compression_threshold = RealParameter(0.2, 1.0, default=0.59, space="buy", load=True) regime_expansion_threshold = RealParameter(0.8, 3.0, default=1.0, space="buy", load=True) regime_hysteresis_bars = IntParameter(2, 40, default=10, space="buy", load=True) entry_up_line_span_short = IntParameter(20, 400, default=120, space="buy", load=True) entry_ma_span_short = IntParameter(1, 30, default=7, space="buy", load=True) entry_pullback_bars_min = IntParameter(4, 60, default=16, space="buy", load=True) exit_max_loss_pct = RealParameter(0.01, 0.10, default=0.03, space="sell", load=True) exit_breakeven_buffer = RealParameter(0.0005, 0.02, default=0.001, space="sell", load=True) exit_max_hold_bars = IntParameter(200, 4000, default=1440, space="sell", load=True) # ----------------------------------------------------------------------- # Pre-fetch sufficient historical data to satisfy indicator warmup needs. # OKX 5m candle limit is 300/request, validated max 5 calls → ~1500 candles. # We need 2016+ for ma_short (7 days × 288 bars/day), so we fetch further # history and raise the exchange retention limit after validation passes. # ----------------------------------------------------------------------- def bot_start(self, **kwargs) -> None: exchange = self.dp._exchange exchange._startup_candle_count = max(exchange._startup_candle_count, 5000) # Only fetch if cache is empty (first start, not restart with warm cache) candle_type = CandleType.FUTURES pairs = self.config["exchange"]["pair_whitelist"] pairs_missing = [ p for p in pairs if (p, self.timeframe, candle_type) not in exchange._klines ] if not pairs_missing: return since_ms = int((datetime.now(timezone.utc) - timedelta(days=12)).timestamp() * 1000) logger.info( f"Pre-fetching ~12 days of history for {len(pairs_missing)} pairs " f"to satisfy indicator warmup..." ) for pair in pairs_missing: try: df = exchange.get_historic_ohlcv( pair=pair, timeframe=self.timeframe, since_ms=since_ms, candle_type=candle_type, ) if not df.empty: exchange._klines[(pair, self.timeframe, candle_type)] = df exchange._pairs_last_refresh_time[ (pair, self.timeframe, candle_type) ] = int(df.iloc[-1]["date"].timestamp() * 1000) logger.info( f" {pair}: pre-loaded {len(df)} candles " f"(from {df.iloc[0]['date']} to {df.iloc[-1]['date']})" ) except Exception as e: logger.warning(f" {pair}: pre-fetch failed ({e}), " f"will rely on normal data loading") # ----------------------------------------------------------------------- # Indicator calculation # ----------------------------------------------------------------------- def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Resolve all parameter values once up_line_span = self.up_line_span.value up_line_offset = self.up_line_offset.value buy_stop_profit_span = self.buy_stop_profit_span.value buy_stop_profit_offset = self.buy_stop_profit_offset.value ma_span_long = self.ma_span_long.value regime_slope_lookback = self.regime_slope_lookback.value regime_slope_threshold = self.regime_slope_threshold.value regime_displace_threshold = self.regime_displace_threshold.value regime_vol_ema_span = self.regime_vol_ema_span.value regime_compression_threshold = self.regime_compression_threshold.value regime_expansion_threshold = self.regime_expansion_threshold.value regime_hysteresis_bars = self.regime_hysteresis_bars.value entry_up_line_span_short = self.entry_up_line_span_short.value entry_ma_span_short = self.entry_ma_span_short.value entry_pullback_bars_min = self.entry_pullback_bars_min.value # -- Vectorized channel indicators ---------------------------------- dataframe["up_line"] = ( dataframe["high"].rolling(up_line_span).max() ) dataframe["up_line_short"] = ( dataframe["high"].rolling(entry_up_line_span_short).max() ) dataframe["stop_profit_bottom"] = ( dataframe["low"].rolling(buy_stop_profit_span).min() ) ma_len = int(ma_span_long) * self._BARS_PER_DAY dataframe["ma_long"] = dataframe["close"].rolling(ma_len).mean() ma_short_len = int(entry_ma_span_short) * self._BARS_PER_DAY dataframe["ma_short"] = dataframe["close"].rolling(ma_short_len).mean() # -- Stateful N-value + regime (iterate through dataframe) ---------- n_values = [0.0] * len(dataframe) regimes = [REGIME_RANGE] * len(dataframe) vol_expanding = [False] * len(dataframe) pullback_trigger = [False] * len(dataframe) n_ema = None regime_candidate = REGIME_RANGE regime_candidate_bars = 0 effective_regime = REGIME_RANGE prev_vol_ratio = None pullback_bars_below = 0 close = dataframe["close"].values high = dataframe["high"].values low = dataframe["low"].values ma_long_arr = dataframe["ma_long"].values ma_short_arr = dataframe["ma_short"].values chunk_count = N_VALUE_SPAN n_window = SPREAD_SPAN * N_VALUE_SPAN # 60 for i in range(len(dataframe)): # --- N-value --------------------------------------------------- if i >= n_window: # Need 60 full bars of history BEFORE current spreads = [] for j in range(chunk_count): start = i - n_window + j * SPREAD_SPAN end = start + SPREAD_SPAN chunk_high = high[start:end].max() chunk_low = low[start:end].min() spreads.append(chunk_high - chunk_low) n_val = _ema_of(spreads) else: n_val = 0.0 n_values[i] = n_val # --- N-value EMA for vol_ratio -------------------------------- if n_ema is None: n_ema = n_val else: ema_alpha = 2.0 / (regime_vol_ema_span + 1.0) n_ema = ema_alpha * n_val + (1.0 - ema_alpha) * n_ema vol_ratio = n_val / n_ema if n_ema and n_ema > 0 else 1.0 vol_expanding[i] = prev_vol_ratio is not None and vol_ratio > prev_vol_ratio prev_vol_ratio = vol_ratio # --- Regime classification ------------------------------------ ma_l = ma_long_arr[i] regime = REGIME_RANGE if ma_l is not None and not np.isnan(ma_l) and n_val > 0: # Price displacement from MA if ma_l != 0: price_displacement = (close[i] - ma_l) / ma_l else: price_displacement = 0.0 # MA slope ma_slope = None if i >= int(regime_slope_lookback): past_idx = i - int(regime_slope_lookback) ma_past = ma_long_arr[past_idx] if (ma_past is not None and not np.isnan(ma_past) and ma_past != 0): ma_slope = (ma_l - ma_past) / ma_past # Classify if vol_ratio < regime_compression_threshold: regime = REGIME_COMPRESSION elif vol_ratio > regime_expansion_threshold: regime = REGIME_EXPANSION elif (ma_slope is not None and abs(ma_slope) > regime_slope_threshold and abs(price_displacement) > regime_displace_threshold): regime = REGIME_TREND else: regime = REGIME_RANGE # --- Regime hysteresis ----------------------------------------- if regime == regime_candidate: regime_candidate_bars += 1 else: regime_candidate = regime regime_candidate_bars = 1 if regime_candidate_bars >= int(regime_hysteresis_bars): effective_regime = regime_candidate regimes[i] = effective_regime # --- Pullback reentry tracking (for range entries) ------------- if ma_short_arr[i] is not None and not np.isnan(ma_short_arr[i]): if close[i] < ma_short_arr[i]: pullback_bars_below += 1 else: if (pullback_bars_below >= entry_pullback_bars_min and close[i] > ma_short_arr[i]): pullback_trigger[i] = True pullback_bars_below = 0 dataframe["n_value"] = n_values dataframe["regime"] = regimes dataframe["vol_expanding"] = vol_expanding dataframe["pullback_trigger"] = pullback_trigger dataframe["_n_ema"] = n_ema # Store reference value for diagnostics # Diagnostics: regime distribution and indicator health recent = dataframe.tail(288) # last 24h r_counts = recent["regime"].value_counts().to_dict() regime_names = {0: "COMPR", 1: "EXPAN", 2: "TREND", 3: "RANGE"} parts = [] for r, name in regime_names.items(): if r in r_counts: parts.append(f"{name}={r_counts[r]}") ma_ok = int(not dataframe["ma_short"].isna().all()) n_ok = int((dataframe["n_value"] > 0).any()) entry_count = int(dataframe["enter_long"].sum()) if "enter_long" in dataframe else 0 logger.info( f"{metadata['pair']}: candles={len(dataframe)}, " f"regime_24h=[{', '.join(parts)}], " f"ma_short_ok={ma_ok}, n_ok={n_ok}, entries={entry_count}" ) return dataframe # ----------------------------------------------------------------------- # Entry signals # ----------------------------------------------------------------------- def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: up_line_offset = self.up_line_offset.value buy_stop_profit_offset = self.buy_stop_profit_offset.value # noqa (kept for future use) n_val = dataframe["n_value"] close = dataframe["close"] regime = dataframe["regime"] # -- Trend: breakout chase ------------------------------------------ trend_cond = ( (regime == REGIME_TREND) & (close > dataframe["ma_long"]) & (close > dataframe["up_line"] + n_val * up_line_offset) ) # -- Compression / Expansion: compression breakout ------------------ ce_cond = ( ((regime == REGIME_COMPRESSION) | (regime == REGIME_EXPANSION)) & (dataframe["vol_expanding"]) & (close > dataframe["up_line_short"] + n_val * up_line_offset) ) # -- Range: pullback reentry ---------------------------------------- range_cond = ( (regime == REGIME_RANGE) & (dataframe["pullback_trigger"]) ) dataframe.loc[trend_cond | ce_cond | range_cond, "enter_long"] = 1 return dataframe # ----------------------------------------------------------------------- # Exit signals (stub — real exits in custom_stoploss / custom_exit) # ----------------------------------------------------------------------- def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe # ----------------------------------------------------------------------- # Dynamic stoploss: protective → trailing → breakeven # ----------------------------------------------------------------------- def custom_stoploss( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> Optional[float]: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return None last = dataframe.iloc[-1] n_value = last.get("n_value", 0) if not n_value or n_value <= 0: return None exit_max_loss_pct = self.exit_max_loss_pct.value exit_breakeven_buffer = self.exit_breakeven_buffer.value buy_stop_profit_offset = self.buy_stop_profit_offset.value strategy_lever_rate = self.strategy_lever_rate.value # 1. Protective stop — absolute price floor protective_price = trade.open_rate * ( 1.0 - exit_max_loss_pct * strategy_lever_rate ) # 2. Trailing stop stop_profit_bottom = last.get("stop_profit_bottom", 0) trailing_price = stop_profit_bottom + n_value * buy_stop_profit_offset # 3. Breakeven stop breakeven_price = None if current_profit > exit_breakeven_buffer: trade.set_custom_data("breakeven_armed", True) if trade.get_custom_data("breakeven_armed"): breakeven_price = trade.open_rate - n_value * 0.3 # Choose the highest (tightest) stop price among active stops candidates = [protective_price] if trailing_price and trailing_price > 0: candidates.append(trailing_price) if breakeven_price is not None: candidates.append(breakeven_price) stop_price = max(candidates) # Convert to relative stoploss and ensure it's below current_rate sl = stoploss_from_absolute( stop_price, current_rate, is_short=trade.is_short, leverage=trade.leverage, ) # Only tighten — never let stoploss go above previous return sl # ----------------------------------------------------------------------- # Custom exits: lock profit → time stop # ----------------------------------------------------------------------- def custom_exit( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> Optional[str]: profit_line = self.profit_line.value lock_profit_rate = self.lock_profit_rate.value exit_max_hold_bars = self.exit_max_hold_bars.value # --- Track peak profit -------------------------------------------- max_pp = trade.get_custom_data("max_profit_pct") or 0.0 if current_profit > max_pp: trade.set_custom_data("max_profit_pct", current_profit) max_pp = current_profit # --- Lock profit -------------------------------------------------- if (max_pp >= profit_line and current_profit > 0 and current_profit < max_pp * (1.0 - lock_profit_rate)): return "lock_profit" # --- Time stop ---------------------------------------------------- bars_held = ( (current_time.replace(tzinfo=timezone.utc) - trade.open_date_utc).total_seconds() / 600.0 ) if bars_held >= exit_max_hold_bars: return "time_stop" return None # ----------------------------------------------------------------------- # N-value based dynamic position sizing # ----------------------------------------------------------------------- def custom_stake_amount( self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float | None, max_stake: float, leverage: float, entry_tag: str | None, side: str, **kwargs, ) -> float: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return 0.0 n_value = dataframe.iloc[-1].get("n_value", 0) if n_value <= 0 or current_rate <= 0: return 0.0 strategy_lever_rate = self.strategy_lever_rate.value stop_loss_pct = n_value / current_rate pct = 0.01 * strategy_lever_rate / stop_loss_pct pct = max(0.0, min(pct, 1.0)) stake = max(float(min_stake or 0), min(max_stake, max_stake * pct)) return stake # ----------------------------------------------------------------------- # Fixed leverage for OKX USDT-M futures # ----------------------------------------------------------------------- def leverage( self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str, **kwargs, ) -> float: return 10.0