Freqtrade strategy powered by DeepAlpha's ML pipeline.
Timeframe
1h
Direction
Long Only
Stoploss
-4.0%
Trailing Stop
No
ROI
0m: 6.0%, 240m: 4.0%, 480m: 2.5%, 1440m: 1.5%
Interface Version
3
Startup Candles
N/A
Indicators
11
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
DeepAlpha Example Strategy for Freqtrade
=========================================
A minimal Freqtrade strategy demonstrating how to use the DeepAlpha
FreqAI model. This strategy:
- Defines technical indicator features for FreqAI
- Uses Triple Barrier labels generated by the DeepAlpha model
- Filters entries using meta-labeling confidence scores
- Implements basic risk management via stop-loss and ROI
Usage:
freqtrade trade --strategy DeepAlphaStrategy --freqaimodel DeepAlphaModel
Author: DeepAlpha Team
License: MIT
"""
from __future__ import annotations
import logging
from typing import Dict
import numpy as np
import pandas as pd
import talib.abstract as ta
from pandas import DataFrame
from freqtrade.strategy import IStrategy, merge_informative_pair
from freqtrade.strategy.parameters import IntParameter, DecimalParameter
logger = logging.getLogger(__name__)
class DeepAlphaStrategy(IStrategy):
"""
Freqtrade strategy powered by DeepAlpha's ML pipeline.
The strategy delegates all signal generation to the FreqAI model.
Features are defined in `feature_engineering_*` methods and passed
to the DeepAlpha model, which returns predictions and a confidence
mask from meta-labeling.
"""
# ---------------------------------------------------------------
# Strategy configuration
# ---------------------------------------------------------------
INTERFACE_VERSION = 3
# Minimal ROI table tuned for 1h bars — less aggressive take-profit
# so winners have room to compound over multiple bars.
minimal_roi = {
"0": 0.06,
"240": 0.04,
"480": 0.025,
"1440": 0.015,
}
# 1h timeframe: reduces intraday noise that dominated 5m tests and
# lets Triple Barrier capture real directional moves.
stoploss = -0.04
trailing_stop = False
trailing_stop_positive = 0.015
trailing_stop_positive_offset = 0.03
timeframe = "1h"
# FreqAI configuration
freqai_info = {
"enabled": True,
}
# Strategy parameters (optimizable).
# SHORT threshold higher than LONG (0.60 vs 0.55) to compensate for
# the residual downside skew of Triple Barrier on volatile intraday data.
buy_prediction_threshold = DecimalParameter(
0.5, 0.8, default=0.55, decimals=2, space="buy",
optimize=True, load=True,
)
sell_prediction_threshold = DecimalParameter(
0.5, 0.8, default=0.60, decimals=2, space="sell",
optimize=True, load=True,
)
meta_confidence_threshold = DecimalParameter(
0.5, 0.8, default=0.55, decimals=2, space="buy",
optimize=True, load=True,
)
# Confidence level that unlocks counter-trend entries (e.g. take a
# short while EMA24 > EMA96 only if P_short exceeds this).
counter_trend_override = DecimalParameter(
0.6, 0.9, default=0.70, decimals=2, space="buy",
optimize=True, load=True,
)
# ---------------------------------------------------------------
# Feature engineering
# ---------------------------------------------------------------
def feature_engineering_expand_all(
self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
) -> DataFrame:
"""
Create features that will be expanded across multiple periods.
FreqAI automatically generates these features for each period
defined in the config (e.g., 10, 20, 50 candles).
"""
# Momentum indicators
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)
# Volatility
bollinger = ta.BBANDS(dataframe, timeperiod=period, nbdevup=2.0, nbdevdn=2.0)
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"] + 1e-10)
)
dataframe[f"%-atr-period_{period}"] = ta.ATR(dataframe, timeperiod=period)
# Trend
dataframe[f"%-sma-period_{period}"] = ta.SMA(dataframe, timeperiod=period)
dataframe[f"%-ema-period_{period}"] = ta.EMA(dataframe, timeperiod=period)
dataframe[f"%-close-sma-ratio_{period}"] = (
dataframe["close"] / (dataframe[f"%-sma-period_{period}"] + 1e-10)
)
# Volume
dataframe[f"%-volume-sma-ratio_{period}"] = (
dataframe["volume"]
/ (dataframe["volume"].rolling(period).mean() + 1e-10)
)
return dataframe
def feature_engineering_expand_basic(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
"""
Create features that are not expanded across multiple periods.
"""
# MACD
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
dataframe["%-macd"] = macd["macd"]
dataframe["%-macd-signal"] = macd["macdsignal"]
dataframe["%-macd-hist"] = macd["macdhist"]
# Stochastic
stoch = ta.STOCH(dataframe, fastk_period=14, slowk_period=3, slowd_period=3)
dataframe["%-slowk"] = stoch["slowk"]
dataframe["%-slowd"] = stoch["slowd"]
# Ichimoku
dataframe["%-tenkan"] = (
dataframe["high"].rolling(9).max() + dataframe["low"].rolling(9).min()
) / 2
dataframe["%-kijun"] = (
dataframe["high"].rolling(26).max() + dataframe["low"].rolling(26).min()
) / 2
# Price-based features
dataframe["%-pct-change"] = dataframe["close"].pct_change()
dataframe["%-log-return"] = np.log(
dataframe["close"] / dataframe["close"].shift(1)
)
# Candle patterns
dataframe["%-body-ratio"] = (
abs(dataframe["close"] - dataframe["open"])
/ (dataframe["high"] - dataframe["low"] + 1e-10)
)
dataframe["%-upper-shadow"] = (
dataframe["high"]
- dataframe[["close", "open"]].max(axis=1)
) / (dataframe["high"] - dataframe["low"] + 1e-10)
# --- Sequential / memory features (poor-man's LSTM) --------------
# Rolling log-return statistics over multiple horizons give the
# tree model temporal context without needing a recurrent network.
log_ret = dataframe["%-log-return"]
for win in (4, 12, 24, 48):
dataframe[f"%-logret-mean-{win}"] = log_ret.rolling(win).mean()
dataframe[f"%-logret-std-{win}"] = log_ret.rolling(win).std()
dataframe[f"%-logret-sum-{win}"] = log_ret.rolling(win).sum()
dataframe[f"%-logret-skew-{win}"] = log_ret.rolling(win).skew()
# Return z-score: how extreme is the current bar's move vs recent history
roll_std_24 = log_ret.rolling(24).std()
dataframe["%-logret-zscore-24"] = log_ret / (roll_std_24 + 1e-10)
# Volatility-of-volatility: regime-change signal
dataframe["%-vol-of-vol-48"] = (
roll_std_24.rolling(48).std() / (roll_std_24.rolling(48).mean() + 1e-10)
)
# Momentum-of-momentum: acceleration features
for lag in (1, 3, 6, 12):
dataframe[f"%-close-lag-{lag}"] = (
dataframe["close"] / dataframe["close"].shift(lag) - 1
)
dataframe[f"%-vol-lag-{lag}"] = (
dataframe["volume"] / (dataframe["volume"].shift(lag) + 1e-10)
)
# High-Low range momentum
hl_range = (dataframe["high"] - dataframe["low"]) / dataframe["close"]
dataframe["%-hl-range-ratio-24"] = hl_range / (hl_range.rolling(24).mean() + 1e-10)
return dataframe
def feature_engineering_standard(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
"""
Features computed after all expansion. Also defines the target
label column that the model will train on.
"""
# Day-of-week and hour features
dataframe["%-day-of-week"] = dataframe["date"].dt.dayofweek
dataframe["%-hour"] = dataframe["date"].dt.hour
# NOTE: Do NOT define a "&-label" column here. Target generation is
# handled exclusively in set_freqai_targets() with Triple Barrier
# labels. If both "&-label" and "&-target" exist, FreqAI's
# BaseClassifierModel.fit uses the first one alphabetically
# ("&-label"), which would bypass our Triple Barrier labels.
return dataframe
def set_freqai_targets(self, dataframe: DataFrame, metadata: Dict, **kwargs) -> DataFrame:
"""
Generate regime-aware Triple Barrier labels.
Class mapping (3-class, LightGBM-friendly non-negative ints):
0 = SHORT (-1), 1 = FLAT (0), 2 = LONG (+1)
The previous symmetric (1.5 / 1.5 ATR) version produced too many
SHORT labels in bull markets because intraday pullbacks hit the
lower barrier first even when the overall drift was positive.
This version detects the local regime via EMA24 vs EMA96 on the
training timeframe and adapts the barrier multipliers:
* BULL (EMA24 > EMA96 * 1.002): pt=1.2, sl=2.0 (favour LONG)
* BEAR (EMA24 < EMA96 * 0.998): pt=2.0, sl=1.2 (favour SHORT)
* SIDEWAYS: pt=1.5, sl=1.5 (symmetric)
Horizon raised 12 -> 24 candles (2h on 5m) to let directional
moves play out before the time-barrier fallback triggers.
"""
import numpy as np
closes = dataframe["close"].values
highs = dataframe["high"].values
lows = dataframe["low"].values
# Regime detection via EMA slope
close_s = pd.Series(closes)
ema_fast = close_s.ewm(span=24, adjust=False).mean().values
ema_slow = close_s.ewm(span=96, adjust=False).mean().values
regime = np.zeros(len(closes), dtype=int)
regime[ema_fast > ema_slow * 1.002] = 1 # BULL
regime[ema_fast < ema_slow * 0.998] = -1 # BEAR
# ATR for adaptive barrier sizing
atr_period = 14
tr = np.maximum.reduce([
highs - lows,
np.abs(highs - np.roll(closes, 1)),
np.abs(lows - np.roll(closes, 1))
])
atr = np.zeros(len(closes))
if len(tr) >= atr_period:
atr[atr_period:] = np.convolve(
tr, np.ones(atr_period) / atr_period, mode='valid'
)[:len(closes) - atr_period]
horizon = 12 # 12 hours on a 1h timeframe
labels = np.zeros(len(closes), dtype=int)
for i in range(len(closes) - horizon):
if atr[i] <= 0:
continue
entry = closes[i]
atr_pct = atr[i] / entry if entry > 0 else 0
if atr_pct == 0:
continue
# Regime-aware asymmetric barriers
if regime[i] == 1:
pt_mult, sl_mult = 1.2, 2.0
elif regime[i] == -1:
pt_mult, sl_mult = 2.0, 1.2
else:
pt_mult, sl_mult = 1.5, 1.5
pt_long = entry * (1 + pt_mult * atr_pct)
sl_long = entry * (1 - sl_mult * atr_pct)
end = min(i + horizon, len(closes) - 1)
hit = 0
for j in range(i + 1, end + 1):
if highs[j] >= pt_long:
hit = 1 # LONG wins
break
if lows[j] <= sl_long:
hit = -1 # SHORT wins
break
if hit == 0:
# Time barrier: use sign of return vs regime-aware dead-zone
ret = (closes[end] - entry) / entry
if ret > atr_pct * 0.3:
hit = 1
elif ret < -atr_pct * 0.3:
hit = -1
labels[i] = hit
target_3class = (labels + 1).astype(int)
dataframe["&-target"] = target_3class
return dataframe
# ---------------------------------------------------------------
# Signal generation
# ---------------------------------------------------------------
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
FreqAI handles indicator generation via feature_engineering methods.
This method is intentionally minimal.
"""
# FreqAI populates predictions automatically
dataframe = self.freqai.start(dataframe, metadata, self)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Generate entry signals from DeepAlpha 3-class model predictions,
gated by a same-timeframe EMA regime filter.
Class mapping (LightGBM multi-class, non-negative ints):
"0" -> P(SHORT)
"1" -> P(FLAT)
"2" -> P(LONG)
Entry rules:
* LONG if P_long > long_thr AND P_long > P_short
AND (uptrend OR P_long > counter_trend_override)
* SHORT if P_short > short_thr AND P_short > P_long
AND (downtrend OR P_short > counter_trend_override)
"""
n = len(dataframe)
p_short = dataframe["0"] if "0" in dataframe.columns else pd.Series([0.0] * n, index=dataframe.index)
p_long = dataframe["2"] if "2" in dataframe.columns else pd.Series([0.0] * n, index=dataframe.index)
do_predict = (
dataframe["do_predict"] if "do_predict" in dataframe.columns
else pd.Series([1] * n, index=dataframe.index)
)
# Regime filter on the trading timeframe (EMA24 vs EMA96 on 5m)
ema_fast = dataframe["close"].ewm(span=24, adjust=False).mean()
ema_slow = dataframe["close"].ewm(span=96, adjust=False).mean()
uptrend = ema_fast > ema_slow
downtrend = ema_fast < ema_slow
long_thr = self.buy_prediction_threshold.value
short_thr = self.sell_prediction_threshold.value
override = self.counter_trend_override.value
# Hard regime gate: never counter-trend. Counter-trend shorts in
# bull markets produced a 3% stop-loss carnage in our Q1 2024
# tests (112 stop-losses = -36.6% total). The override was too
# permissive even at 0.70 confidence. Force alignment with
# same-timeframe EMA regime.
long_entry = (
(p_long > long_thr)
& (p_long > p_short)
& (do_predict == 1)
& uptrend
)
short_entry = (
(p_short > short_thr)
& (p_short > p_long)
& (do_predict == 1)
& downtrend
)
_ = override # kept for hyperopt compatibility; currently unused
dataframe.loc[long_entry, "enter_long"] = 1
dataframe.loc[long_entry, "enter_tag"] = "DeepAlpha-LONG"
dataframe.loc[short_entry, "enter_short"] = 1
dataframe.loc[short_entry, "enter_tag"] = "DeepAlpha-SHORT"
return dataframe
# ---------------------------------------------------------------
# Dynamic position sizing based on prediction confidence
# ---------------------------------------------------------------
def custom_stake_amount(
self,
pair: str,
current_time,
current_rate: float,
proposed_stake: float,
min_stake,
max_stake: float,
leverage: float,
entry_tag,
side: str,
**kwargs,
) -> float:
"""
Scale the stake linearly with the winning-class probability.
A P_max of 0.55 (just above the gate) uses 50% of the default
stake; a P_max of 0.80+ uses 150%. This concentrates risk on
high-confidence signals and reduces bleed on marginal calls.
"""
try:
df, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe)
if df is None or df.empty:
return proposed_stake
last = df.iloc[-1]
p_short = float(last.get("0", 0.0) or 0.0)
p_long = float(last.get("2", 0.0) or 0.0)
p_max = p_long if side == "long" else p_short
# Linear scaling: 0.55 -> 0.5x, 0.80 -> 1.5x (clipped)
scale = 0.5 + (p_max - 0.55) * 4.0
scale = max(0.25, min(scale, 1.75))
stake = proposed_stake * scale
if min_stake is not None:
stake = max(stake, float(min_stake))
stake = min(stake, float(max_stake))
return stake
except Exception as e:
logger.warning("custom_stake_amount fallback (%s): %s", pair, e)
return proposed_stake
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
3-class exits: close longs when P_short dominates, close shorts
when P_long dominates. ROI / stoploss / trailing stop remain the
primary risk-management layer.
"""
n = len(dataframe)
p_short = dataframe["0"] if "0" in dataframe.columns else pd.Series([0.0] * n, index=dataframe.index)
p_long = dataframe["2"] if "2" in dataframe.columns else pd.Series([0.0] * n, index=dataframe.index)
dataframe.loc[(p_short > 0.50) & (p_short > p_long), "exit_long"] = 1
dataframe.loc[(p_long > 0.50) & (p_long > p_short), "exit_short"] = 1
return dataframe