Ensemble Donchian V01 — Donchian-lower trailing stop (ratcheted).
Timeframe
1h
Direction
Long Only
Stoploss
-12.0%
Trailing Stop
No
ROI
0m: 10000.0%
Interface Version
N/A
Startup Candles
N/A
Indicators
2
freqtrade/freqtrade-strategies
freqtrade/freqtrade-strategies
this is an example class, implementing a PSAR based trailing stop loss you are supposed to take the `custom_stoploss()` and `populate_indicators()` parts and adapt it to your own strategy
freqtrade/freqtrade-strategies
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa
"""
Ensemble Donchian trend-following (Candidate J) — Phase 1 MVP.
Long-only: multiple Donchian lookbacks vote breakout (close > prior upper band);
ensemble score is the mean vote rate. Enter when score > threshold; exit via
trailing stop (Donchian lower on shortest active lookback at entry, or ATR
trailing) plus hard stoploss and time stop. Inverse-vol position sizing across
the futures whitelist (same pattern as Candidate G).
See user_data/info/EnsembleDonchianTrend_Deep_Dive.md and
EnsembleDonchianTrend_Dev_Plan.md.
Backtest (repo root, PowerShell one line). Requires 1h + 1d futures data for
whitelisted pairs (e.g. ``download-data --timeframes 1h 1d``).
docker compose run --rm freqtrade backtesting --config /freqtrade/config/config_donchian.json --strategy EnsembleDonchianStrategy_V01 --timerange 20220101-20250101 --timeframe 1h --fee 0.0005 --export trades
ATR trailing variant:
docker compose run --rm freqtrade backtesting --config /freqtrade/config/config_donchian.json --strategy EnsembleDonchianStrategy_V01_ATR --timerange 20220101-20250101 --timeframe 1h --fee 0.0005
"""
from __future__ import annotations
import math
from datetime import datetime, timezone
import numpy as np
import pandas as pd
from pandas import DataFrame
from freqtrade.persistence import Trade
from freqtrade.strategy import IStrategy, merge_informative_pair
class EnsembleDonchianStrategy_V01(IStrategy):
"""Ensemble Donchian V01 — Donchian-lower trailing stop (ratcheted)."""
timeframe = "1h"
# Paper uses calendar-day Donchian windows; implement on 1d and merge to 1h
# so startup limits on 1h (~2494 bars) do not cap lookbacks at ~100 days.
DONCHIAN_TIMEFRAME: str = "1d"
can_short = False
# --- Risk -----------------------------------------------------------------
minimal_roi = {"0": 100}
stoploss = -0.12
use_custom_stoploss = True
trailing_stop = False
# --- Donchian ensemble (paper: 5–360 calendar days) ----------------------
LOOKBACK_DAYS: tuple[int, ...] = (5, 10, 20, 30, 60, 90, 150, 250, 360)
ENTRY_THRESHOLD: float = 0.5
# "donchian_lower" | "atr"
TRAILING_STOP_METHOD: str = "donchian_lower"
ATR_PERIOD: int = 14
ATR_MULTIPLIER: float = 3.0
VOL_SCALING_WINDOW: int = 168 # ~7d of 1h
VOL_SCALING_FLOOR: float = 0.5
VOL_SCALING_CAP: float = 2.0
TIME_STOP_HOURS: float = 720.0 # 30d backup
process_only_new_candles = True
# Only 1h indicators (vol, ATR) need a modest warm-up; Donchian runs on 1d.
startup_candle_count: int = 400
# Per-pair state (single position per pair in futures)
_pair_entry_trail_n: dict[str, int] = {}
_pair_ratched_stop: dict[str, float] = {}
# Cached (vol mult crosses pairs on same chunk)
_vol_cache: tuple | None = None
def informative_pairs(self):
tf = self.DONCHIAN_TIMEFRAME
return [(p, tf) for p in self.dp.current_whitelist()]
def _build_vol_stake_mult(self, dates: pd.Series) -> pd.DataFrame:
"""Per-date × per-pair vol stake multiplier (median vol / pair vol), clipped."""
whitelist = list(self.dp.current_whitelist())
if not whitelist:
return pd.DataFrame()
vol_series: dict[str, pd.Series] = {}
w = int(self.VOL_SCALING_WINDOW)
for p in whitelist:
df = self.dp.get_pair_dataframe(pair=p, timeframe=self.timeframe)
if df is None or df.empty:
continue
d = df[["date", "close"]].copy()
d["date"] = pd.to_datetime(d["date"], utc=True)
d = d.sort_values("date")
rets = d["close"].pct_change(fill_method=None)
d["vol"] = rets.rolling(w, min_periods=max(w // 3, 10)).std()
vol_series[p] = d.set_index("date")["vol"]
if not vol_series:
return pd.DataFrame()
vol_df = pd.DataFrame(vol_series).sort_index()
med = vol_df.median(axis=1).replace(0.0, np.nan)
mult = vol_df.rdiv(med, axis=0) # med / vol_asset
mult = mult.clip(float(self.VOL_SCALING_FLOOR), float(self.VOL_SCALING_CAP))
mult = mult.where(vol_df.notna())
want = pd.DatetimeIndex(pd.to_datetime(dates.unique(), utc=True)).sort_values()
return mult.reindex(want).ffill().bfill()
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
pair = metadata["pair"]
if dataframe.empty:
return dataframe
dates = dataframe["date"]
cache_key = (
len(dataframe),
dates.iloc[-1],
self.LOOKBACK_DAYS,
self.DONCHIAN_TIMEFRAME,
self.VOL_SCALING_WINDOW,
)
if self._vol_cache is None or self._vol_cache[0] != cache_key:
vol_mult = self._build_vol_stake_mult(dates)
self._vol_cache = (cache_key, vol_mult)
_, vol_mult = self._vol_cache
h1 = dataframe["high"].astype(float)
lo1 = dataframe["low"].astype(float)
cl1 = dataframe["close"].astype(float)
dtf = self.DONCHIAN_TIMEFRAME
inf = self.dp.get_pair_dataframe(pair=pair, timeframe=dtf)
if inf is None or inf.empty or len(inf) < max(self.LOOKBACK_DAYS) + 2:
for N in self.LOOKBACK_DAYS:
dataframe[f"sig_{N}"] = np.nan
dataframe[f"dc_lower_{N}"] = np.nan
dataframe["ensemble_score"] = np.nan
dataframe["trail_n_active_min"] = np.nan
else:
inf = inf.copy()
h = inf["high"].astype(float)
lo = inf["low"].astype(float)
cl = inf["close"].astype(float)
valid_row = np.ones(len(inf), dtype=bool)
sig_parts: list[np.ndarray] = []
for N in self.LOOKBACK_DAYS:
upper = h.rolling(int(N), min_periods=int(N)).max().shift(1)
valid_row &= upper.notna().to_numpy()
sig = ((cl > upper) & upper.notna()).astype(np.float64)
inf[f"sig_{N}"] = sig
sig_parts.append(sig.to_numpy(dtype=float))
inf[f"dc_lower_{N}"] = lo.rolling(int(N), min_periods=int(N)).min().shift(1)
sig_mat = np.column_stack(sig_parts)
inf["ensemble_score"] = np.where(valid_row, sig_mat.mean(axis=1), np.nan)
trail_n = np.full(len(inf), np.nan)
look = np.array(self.LOOKBACK_DAYS, dtype=float)
for i in range(len(inf)):
if not valid_row[i]:
continue
active = look[sig_mat[i, :] > 0.5]
if active.size:
trail_n[i] = float(active.min())
inf["trail_n_active_min"] = trail_n
suffix = f"_{dtf}"
merged = merge_informative_pair(
dataframe,
inf,
self.timeframe,
dtf,
ffill=True,
)
dataframe = merged
dataframe["ensemble_score"] = dataframe[f"ensemble_score{suffix}"]
dataframe["trail_n_active_min"] = dataframe[f"trail_n_active_min{suffix}"]
for N in self.LOOKBACK_DAYS:
dataframe[f"sig_{N}"] = dataframe[f"sig_{N}{suffix}"]
dataframe[f"dc_lower_{N}"] = dataframe[f"dc_lower_{N}{suffix}"]
# ATR (for ATR trailing) — 1h
prev_close = cl1.shift(1)
tr = pd.concat(
[
(h1 - lo1).abs(),
(h1 - prev_close).abs(),
(lo1 - prev_close).abs(),
],
axis=1,
).max(axis=1)
dataframe["atr"] = tr.rolling(int(self.ATR_PERIOD), min_periods=self.ATR_PERIOD).mean()
d_utc = pd.to_datetime(dataframe["date"], utc=True)
idx = pd.DatetimeIndex(d_utc)
if not vol_mult.empty and pair in vol_mult.columns:
vm = vol_mult[pair].reindex(idx).ffill().bfill().fillna(1.0).to_numpy(dtype=float)
else:
vm = np.ones(len(dataframe), dtype=float)
dataframe["vol_stake_mult"] = vm
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
vol = dataframe["volume"] > 0
ok = vol & (dataframe["ensemble_score"] > float(self.ENTRY_THRESHOLD))
ok = ok & dataframe["ensemble_score"].notna()
ok = ok & dataframe["trail_n_active_min"].notna()
dataframe.loc[ok, "enter_long"] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
return dataframe
def confirm_trade_entry(
self,
pair: str,
order_type: str,
amount: float,
rate: float,
time_in_force: str,
current_time: datetime,
entry_tag: str | None,
side: str,
**kwargs,
) -> bool:
"""Record shortest active lookback for Donchian trailing stop."""
if side != "long":
return True
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if df is None or df.empty:
return True
ct = current_time.replace(tzinfo=timezone.utc) if current_time.tzinfo is None else current_time.astimezone(timezone.utc)
tcol = pd.to_datetime(df["date"], utc=True)
mask = tcol <= ct
if not mask.any():
return True
row = df.loc[mask].iloc[-1]
tn = row.get("trail_n_active_min", np.nan)
if tn is not None and not (isinstance(tn, float) and math.isnan(tn)):
self._pair_entry_trail_n[pair] = int(tn)
else:
self._pair_entry_trail_n[pair] = int(min(self.LOOKBACK_DAYS))
self._pair_ratched_stop.pop(pair, None)
return True
def _cleanup_pair_state(self) -> None:
open_pairs = {t.pair for t in Trade.get_open_trades()}
for p in list(self._pair_entry_trail_n.keys()):
if p not in open_pairs:
self._pair_entry_trail_n.pop(p, None)
self._pair_ratched_stop.pop(p, None)
def bot_loop_start(self, current_time: datetime, **kwargs) -> None:
self._cleanup_pair_state()
def custom_exit(
self,
pair: str,
trade: Trade,
current_time: datetime,
current_rate: float,
current_profit: float,
**kwargs,
) -> str | None:
open_utc = getattr(trade, "open_date_utc", None) or getattr(trade, "open_date", None)
if open_utc is None:
return None
if open_utc.tzinfo is None:
open_utc = open_utc.replace(tzinfo=timezone.utc)
hold_h = (current_time.replace(tzinfo=timezone.utc) - open_utc).total_seconds() / 3600.0
if hold_h >= float(self.TIME_STOP_HOURS):
return "time_stop_donchian"
return None
def _row_at(self, pair: str, current_time: datetime) -> pd.Series | None:
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if df is None or df.empty:
return None
ct = current_time.replace(tzinfo=timezone.utc) if current_time.tzinfo is None else current_time.astimezone(timezone.utc)
tcol = pd.to_datetime(df["date"], utc=True)
mask = tcol <= ct
if not mask.any():
return None
return df.loc[mask].iloc[-1]
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:
row = self._row_at(pair, current_time)
if row is None:
return proposed_stake
mult = float(row.get("vol_stake_mult", 1.0))
scaled = proposed_stake * mult
lo = float(min_stake) if min_stake is not None else 0.0
return max(lo, min(scaled, float(max_stake)))
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 min(2.0, float(max_leverage))
def custom_stoploss(
self,
pair: str,
trade: Trade,
current_time: datetime,
current_rate: float,
current_profit: float,
**kwargs,
) -> float:
row = self._row_at(pair, current_time)
if row is None:
return self.stoploss
if self.TRAILING_STOP_METHOD == "atr":
atr = float(row.get("atr", 0.0) or 0.0)
peak = float(getattr(trade, "max_rate", None) or current_rate)
stop_price = peak - float(self.ATR_MULTIPLIER) * atr
base_sl = (stop_price - current_rate) / max(current_rate, 1e-12)
return max(base_sl, float(self.stoploss))
n = self._pair_entry_trail_n.get(pair, int(min(self.LOOKBACK_DAYS)))
col = f"dc_lower_{n}"
raw = row.get(col, np.nan)
if raw is None or (isinstance(raw, float) and (math.isnan(raw) or np.isnan(raw))):
return self.stoploss
stop_price = float(raw)
prev = self._pair_ratched_stop.get(pair, stop_price)
stop_price = max(stop_price, prev)
self._pair_ratched_stop[pair] = stop_price
base_sl = (stop_price - current_rate) / max(current_rate, 1e-12)
# Freqtrade: negative = below price; cap to hard stoploss (more negative = wider)
return max(base_sl, float(self.stoploss))
class EnsembleDonchianStrategy_V01_ATR(EnsembleDonchianStrategy_V01):
"""Same ensemble entries; ATR trailing from peak since open."""
TRAILING_STOP_METHOD = "atr"
# --- Phase 0 threshold grid (one class per value so the strategy resolver can load them)
class EnsembleDonchianStrategy_V01_Entry030(EnsembleDonchianStrategy_V01):
ENTRY_THRESHOLD = 0.3
class EnsembleDonchianStrategy_V01_Entry070(EnsembleDonchianStrategy_V01):
ENTRY_THRESHOLD = 0.7
class EnsembleDonchianStrategy_V01_Entry090(EnsembleDonchianStrategy_V01):
ENTRY_THRESHOLD = 0.9
class EnsembleDonchianStrategy_V01_LookbackAblated(EnsembleDonchianStrategy_V01):
"""Same as V01 but drop 150/250/360d — shorter effective ensemble."""
LOOKBACK_DAYS: tuple[int, ...] = (5, 10, 20, 30, 60, 90)
class EnsembleDonchianStrategy_V01_LookbackAblated_ATR(EnsembleDonchianStrategy_V01_LookbackAblated):
TRAILING_STOP_METHOD = "atr"