FreqAI-driven crypto strategy — LightGBM predictions drive entries directly.
Timeframe
30m
Direction
Long Only
Stoploss
-5.5%
Trailing Stop
Yes
ROI
0m: 8.0%, 60m: 5.0%, 180m: 3.0%, 1440m: 1.5%
Interface Version
3
Startup Candles
400
Indicators
11
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
CryptoMeanReversion — FreqAI-driven crypto strategy.
Entry: LightGBM regressor predicts positive 4-candle (2h) forward return,
gated by regime detection (BTC 4h SMA50 slope) and BTC macro filter
(BTC 1h EMA200).
Exit: Model predicts negative return (populate_exit_trend) + profit-aware
custom_exit (RSI exhaustion, time stop, regime stop, FreqAI bearish).
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from functools import reduce
import talib.abstract as ta
from pandas import DataFrame
from freqtrade.persistence import Trade
from freqtrade.strategy import (
DecimalParameter,
IntParameter,
IStrategy,
merge_informative_pair,
)
logger = logging.getLogger(__name__)
class CryptoMeanReversion(IStrategy):
"""
FreqAI-driven crypto strategy — LightGBM predictions drive entries directly.
"""
# === Strategy metadata ===
INTERFACE_VERSION = 3
timeframe = "30m"
can_short = False
process_only_new_candles = True
startup_candle_count = 400 # need 200 1h candles for BTC EMA(200)
# === Risk parameters (hyperopt-optimizable) ===
minimal_roi = {
"0": 0.08, # 8% profit target (wider for crypto vol)
"60": 0.05, # 5% after 60 min
"180": 0.03, # 3% after 3 hours
"1440": 0.015, # 1.5% after 24 hours
}
stoploss = -0.055 # 5.5% hard stop (crypto needs room to breathe)
trailing_stop = True
trailing_stop_positive = 0.02 # 2% trail distance
trailing_stop_positive_offset = 0.04 # activate at +4%
trailing_only_offset_is_reached = True
# === Position management ===
position_adjustment_enable = False
max_entry_position_adjustment = 0
# === Order configuration ===
order_types = {
"entry": "limit",
"exit": "limit",
"stoploss": "market",
"stoploss_on_exchange": False,
}
order_time_in_force = {
"entry": "GTC",
"exit": "GTC",
}
# === FreqAI entry/exit thresholds (hyperopt-optimizable) ===
entry_threshold = DecimalParameter(0.001, 0.02, default=0.005, decimals=4, space="buy", optimize=True)
# Entry guardrails — lightweight technical filters
entry_rsi_max = IntParameter(55, 80, default=68, space="buy", optimize=True)
# Exit parameters
rsi_exhaustion = IntParameter(65, 85, default=72, space="sell", optimize=True)
max_hold_candles = IntParameter(24, 96, default=48, space="sell", optimize=True)
time_stop_min_gain = DecimalParameter(0.0, 0.02, default=0.01, decimals=3, space="sell", optimize=True)
min_regime_stop_candles = IntParameter(6, 24, default=12, space="sell", optimize=True)
# FreqAI exit thresholds (previously hardcoded)
freqai_bearish_min_gain = DecimalParameter(0.001, 0.02, default=0.005, decimals=3, space="sell", optimize=True)
freqai_model_threshold = DecimalParameter(-0.015, -0.001, default=-0.005, decimals=3, space="sell", optimize=True)
freqai_cut_loss = DecimalParameter(-0.05, -0.005, default=-0.02, decimals=3, space="sell", optimize=True)
freqai_cut_min_candles = IntParameter(4, 16, default=8, space="sell", optimize=True)
regime_stop_loss = DecimalParameter(-0.03, 0.0, default=-0.005, decimals=3, space="sell", optimize=True)
# Progressive loss cap — catch bleeds before time_stop
early_cut_candles = IntParameter(10, 24, default=16, space="sell", optimize=True)
early_cut_loss = DecimalParameter(-0.03, -0.005, default=-0.015, decimals=3, space="sell", optimize=True)
# === Informative pairs (for regime detection via BTC and market context) ===
def informative_pairs(self) -> list[tuple[str, str]]:
return [
("BTC/USD", "1h"),
("BTC/USD", "4h"),
]
def feature_engineering_expand_all(
self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
) -> DataFrame:
"""FreqAI feature engineering — expanded features across multiple periods."""
dataframe[f"%-rsi-period_{period}"] = ta.RSI(dataframe, timeperiod=period)
dataframe[f"%-mfi-period_{period}"] = ta.MFI(dataframe, timeperiod=period)
dataframe[f"%-adx-period_{period}"] = ta.ADX(dataframe, timeperiod=period)
dataframe[f"%-cci-period_{period}"] = ta.CCI(dataframe, timeperiod=period)
dataframe[f"%-roc-period_{period}"] = ta.ROC(dataframe, timeperiod=period)
# Bollinger Band width
bollinger = ta.BBANDS(dataframe, timeperiod=period)
dataframe[f"%-bb_width-period_{period}"] = (
(bollinger["upperband"] - bollinger["lowerband"]) / bollinger["middleband"]
)
dataframe[f"%-bb_pct-period_{period}"] = (
(dataframe["close"] - bollinger["lowerband"])
/ (bollinger["upperband"] - bollinger["lowerband"])
)
# Volume features
dataframe[f"%-volume_ratio-period_{period}"] = (
dataframe["volume"] / dataframe["volume"].rolling(period).mean()
)
return dataframe
def feature_engineering_expand_basic(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
"""FreqAI basic features — single-period indicators."""
# MACD
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
dataframe["%-macd_hist"] = macd["macdhist"]
dataframe["%-macd_line"] = macd["macd"]
# Stochastic
stoch = ta.STOCH(dataframe, fastk_period=14, slowk_period=3, slowd_period=3)
dataframe["%-stoch_k"] = stoch["slowk"]
dataframe["%-stoch_d"] = stoch["slowd"]
# ATR normalized by price
dataframe["%-atr_pct"] = ta.ATR(dataframe, timeperiod=14) / dataframe["close"]
# Price vs moving averages
dataframe["%-close_sma20_ratio"] = dataframe["close"] / ta.SMA(dataframe, timeperiod=20)
dataframe["%-close_sma50_ratio"] = dataframe["close"] / ta.SMA(dataframe, timeperiod=50)
# Candle features
dataframe["%-body_pct"] = abs(dataframe["close"] - dataframe["open"]) / dataframe["close"]
dataframe["%-upper_wick_pct"] = (
(dataframe["high"] - dataframe[["close", "open"]].max(axis=1)) / dataframe["close"]
)
dataframe["%-lower_wick_pct"] = (
(dataframe[["close", "open"]].min(axis=1) - dataframe["low"]) / dataframe["close"]
)
return dataframe
def feature_engineering_standard(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
"""FreqAI standard features — target variable for regression."""
# Target: next-N-candle return (predict 4 candles ahead = 2 hours on 30m)
dataframe["%-day_of_week"] = dataframe["date"].dt.dayofweek
dataframe["%-hour_of_day"] = dataframe["date"].dt.hour
return dataframe
def set_freqai_targets(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
"""Define the prediction target for FreqAI."""
# Predict 4-candle (2h) forward return
dataframe["&-target"] = (
dataframe["close"].shift(-4) / dataframe["close"] - 1
)
return dataframe
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Compute regime detection indicators and run FreqAI."""
# RSI for custom_exit profit-taking
dataframe["rsi_14"] = ta.RSI(dataframe, timeperiod=14)
# === Higher-timeframe BTC data for macro regime detection ===
# 4h BTC: SMA(50) slope — ~8 days of trend context
btc_4h = self.dp.get_pair_dataframe("BTC/USD", "4h")
if not btc_4h.empty:
btc_4h["btc_sma_50"] = ta.SMA(btc_4h, timeperiod=50)
btc_4h["btc_sma_50_slope"] = btc_4h["btc_sma_50"].pct_change(5)
btc_4h["btc_close"] = btc_4h["close"]
btc_4h_inf = btc_4h[["date", "btc_sma_50", "btc_sma_50_slope", "btc_close"]].copy()
dataframe = merge_informative_pair(
dataframe, btc_4h_inf, self.timeframe, "4h", ffill=True
)
# 1h BTC: EMA(200) for macro trend filter
btc_1h = self.dp.get_pair_dataframe("BTC/USD", "1h")
if not btc_1h.empty:
btc_1h["btc_ema_200"] = ta.EMA(btc_1h, timeperiod=200)
btc_1h["btc_close"] = btc_1h["close"]
btc_1h_inf = btc_1h[["date", "btc_ema_200", "btc_close"]].copy()
dataframe = merge_informative_pair(
dataframe, btc_1h_inf, self.timeframe, "1h", ffill=True
)
# === Regime detection — HTF-based ===
if "btc_sma_50_slope_4h" in dataframe.columns:
dataframe["regime"] = "ranging"
dataframe.loc[dataframe["btc_sma_50_slope_4h"] > 0.002, "regime"] = "bullish"
dataframe.loc[dataframe["btc_sma_50_slope_4h"] < -0.002, "regime"] = "bearish"
else:
sma_50 = ta.SMA(dataframe, timeperiod=50)
sma_50_slope = sma_50.pct_change(5)
dataframe["regime"] = "ranging"
dataframe.loc[sma_50_slope > 0.002, "regime"] = "bullish"
dataframe.loc[sma_50_slope < -0.002, "regime"] = "bearish"
# BTC macro trend: is BTC above its 1h EMA(200)?
if "btc_close_1h" in dataframe.columns and "btc_ema_200_1h" in dataframe.columns:
dataframe["btc_macro_bullish"] = dataframe["btc_close_1h"] > dataframe["btc_ema_200_1h"]
else:
dataframe["btc_macro_bullish"] = True # fallback: don't filter
# === FreqAI ===
dataframe = self.freqai.start(dataframe, metadata, self)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""FreqAI-driven entries: model predicts positive return, gated by regime + technical filters."""
conditions = []
# FreqAI: model must be confident (DI threshold passed)
conditions.append(dataframe["do_predict"] == 1)
# FreqAI: predicted return exceeds entry threshold
conditions.append(dataframe["&-target"] > self.entry_threshold.value)
# Regime filter: NEVER enter in bearish regime
conditions.append(dataframe["regime"] != "bearish")
# BTC macro trend: only enter when BTC is above 1h EMA(200)
conditions.append(dataframe["btc_macro_bullish"])
# RSI guardrail: don't buy into overbought conditions
conditions.append(dataframe["rsi_14"] < self.entry_rsi_max.value)
# Volume sanity: skip zero-volume candles
conditions.append(dataframe["volume"] > 0)
if conditions:
dataframe.loc[reduce(lambda a, b: a & b, conditions), "enter_long"] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Exits handled by ROI, trailing stop, and custom_exit — not candle signals."""
# Disabled: FreqAI exit_signal caused 707/785 noise exits at 49.4% win rate.
# ROI + trailing + custom_exit are profit-aware and far more selective.
dataframe["exit_long"] = 0
return dataframe
def custom_exit(
self,
pair: str,
trade: Trade,
current_time: datetime,
current_rate: float,
current_profit: float,
**kwargs,
) -> str | bool | None:
"""
Custom exit logic — all profit-aware exits live here.
Handles:
- RSI exhaustion: take profits on strength (only when winning)
- FreqAI bearish: protect gains or cut confirmed losers
- Time stop: held > max_hold_candles with insufficient gain
- Regime stop: exit on bearish regime flip while in loss
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe.empty:
return None
last_candle = dataframe.iloc[-1]
trade_duration_candles = (current_time - trade.open_date_utc).total_seconds() / (30 * 60)
# RSI exhaustion — profit-taking only (don't lock in losses)
if (
last_candle["rsi_14"] >= self.rsi_exhaustion.value
and current_profit > 0
):
return f"rsi_exhaustion (RSI {last_candle['rsi_14']:.0f}, gain {current_profit:.2%})"
# FreqAI bearish exit — profit-aware
if "&-target" in last_candle.index and "do_predict" in last_candle.index:
if last_candle["do_predict"] == 1 and last_candle["&-target"] < self.freqai_model_threshold.value:
# Protect any meaningful gain when model turns bearish
if current_profit > self.freqai_bearish_min_gain.value:
return f"freqai_bearish (gain {current_profit:.2%})"
# Cut losses on model conviction after minimum hold
if current_profit < self.freqai_cut_loss.value and trade_duration_candles >= self.freqai_cut_min_candles.value:
return f"freqai_cut (loss {current_profit:.2%})"
# Progressive loss cap — exit if bleeding beyond threshold after N candles
if trade_duration_candles >= self.early_cut_candles.value and current_profit < self.early_cut_loss.value:
return f"early_cut ({trade_duration_candles:.0f} candles, loss {current_profit:.2%})"
# Time stop (Sauce condition #7)
if trade_duration_candles > self.max_hold_candles.value:
if current_profit < self.time_stop_min_gain.value:
return f"time_stop ({trade_duration_candles:.0f} candles, gain {current_profit:.2%})"
# Regime stop (Sauce condition #8) — exit on bearish flip while losing
# Only after minimum hold to avoid immediate exit from spread
if (
trade_duration_candles >= self.min_regime_stop_candles.value
and last_candle.get("regime") == "bearish"
and current_profit < self.regime_stop_loss.value
):
return f"regime_stop (bearish, loss {current_profit:.2%})"
return None
def custom_stake_amount(
self,
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:
"""
Position sizing from Sauce seed tier: 20% of equity per position.
Freqtrade handles this via stake_amount / tradable_balance_ratio,
but this gives us explicit control matching Sauce's TierParams.
"""
# Use 20% of total balance (Sauce SEED_PARAMS.max_position_pct = 0.20)
# Freqtrade already handles max via config, but we enforce floor/ceiling
return proposed_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:
"""No leverage — spot only (matching Sauce)."""
return 1.0