522 lines
21 KiB
Python
522 lines
21 KiB
Python
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
|