v0.3.1
Full Changelog: v0.2.20...v0.3.1
FreqAI
class Fibbo(Istrategy):
def feature_engineering_expand_all(
self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
) -> DataFrame:
"""
*Only functional with FreqAI enabled strategies*
This function will automatically expand the defined features on the config defined
`indicator_periods_candles`, `include_timeframes`, `include_shifted_candles`, and
`include_corr_pairs`. In other words, a single feature defined in this function
will automatically expand to a total of
`indicator_periods_candles` * `include_timeframes` * `include_shifted_candles` *
`include_corr_pairs` numbers of features added to the model.
All features must be prepended with `%` to be recognized by FreqAI internals.
More details on how these config defined parameters accelerate feature engineering
in the documentation at:
https://www.freqtrade.io/en/latest/freqai-parameter-table/#feature-parameters
https://www.freqtrade.io/en/latest/freqai-feature-engineering/#defining-the-features
:param dataframe: strategy dataframe which will receive the features
:param period: period of the indicator - usage example:
:param metadata: metadata of current pair
dataframe["%-ema-period"] = ta.EMA(dataframe, timeperiod=period)
"""
dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period)
dataframe["%-mfi-period"] = ta.MFI(dataframe, timeperiod=period)
dataframe["%-adx-period"] = ta.ADX(dataframe, timeperiod=period)
dataframe["%-sma-period"] = ta.SMA(dataframe, timeperiod=period)
dataframe["%-ema-period"] = ta.EMA(dataframe, timeperiod=period)
bollinger = qtpylib.bollinger_bands(
qtpylib.typical_price(dataframe), window=period, stds=2.2
)
dataframe["bb_lowerband-period"] = bollinger["lower"]
dataframe["bb_middleband-period"] = bollinger["mid"]
dataframe["bb_upperband-period"] = bollinger["upper"]
dataframe["%-bb_width-period"] = (
dataframe["bb_upperband-period"] - dataframe["bb_lowerband-period"]
) / dataframe["bb_middleband-period"]
dataframe["%-close-bb_lower-period"] = dataframe["close"] / dataframe["bb_lowerband-period"]
dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period)
dataframe["%-relative_volume-period"] = (
dataframe["volume"] / dataframe["volume"].rolling(period).mean()
)
return dataframe
def set_freqai_targets(
self,
dataframe: DataFrame,
metadata: dict,
**kwargs
) -> DataFrame:
"""
FreqAI target definition for:
- Classifier
- ClassifierMultiTarget
- Regressor
- RegressorMultiTarget
"""
model_name = self.model_name.lower()
is_classifier = "classifier" in model_name
is_multi_target = "multitarget" in model_name
label_period = self.freqai_info["feature_parameters"]["label_period_candles"]
if is_classifier:
# ==================================================
# CLASSIFIERS
# ==================================================
if is_multi_target:
# CatboostClassifierMultiTarget
# IMPORTANT:
# - class labels must be UNIQUE across targets
# - target 1 uses {0, 1}
# - target 2 uses {2, 3}
self.freqai.class_names = [0, 1, 2, 3]
# Target 1: direction (0 = down, 1 = up)
dataframe["&s-up_or_down"] = (
dataframe["close"].shift(-label_period) > dataframe["close"]
).astype(int)
# Target 2: volatility (2 = low, 3 = high)
dataframe["&s-volatility"] = (
(
dataframe["close"].rolling(label_period).std()
> dataframe["close"].rolling(label_period).std().median()
).astype(int)
+ 2
)
else:
# CatboostClassifier (single target)
self.freqai.class_names = [0, 1]
dataframe["&s-up_or_down"] = (
dataframe["close"].shift(-label_period) > dataframe["close"]
).astype(int)
else:
# ==================================================
# REGRESSORS
# ==================================================
if is_multi_target:
# CatboostRegressorMultiTarget
dataframe["&-s_close"] = (
dataframe["close"]
.shift(-label_period)
.rolling(label_period)
.mean()
/ dataframe["close"]
- 1
)
dataframe["&-s_range"] = (
dataframe["close"]
.shift(-label_period)
.rolling(label_period)
.max()
-
dataframe["close"]
.shift(-label_period)
.rolling(label_period)
.min()
)
else:
# CatboostRegressor
dataframe["&-s_close"] = (
dataframe["close"]
.shift(-label_period)
.rolling(label_period)
.mean()
/ dataframe["close"]
- 1
)
return dataframe
Fibbo Indicator
class Fibbo(Istrategy):
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
pair = metadata["pair"]
# --- FreqAI (robust for dynamic pairs) ---
if self.freqai is not None and self.freqai_enabled:
try:
# Start FreqAI
dataframe = self.freqai.start(dataframe, metadata, self)
# Process DI_values if available
if 'DI_values' in dataframe.columns:
# Check if we have enough data for meaningful percentile
if len(dataframe) >= self.di_rolling_window:
dataframe['di_percentile'] = (dataframe['DI_values']
.rolling(self.di_rolling_window)
.rank(pct=True))
logger.debug(f"FreqAI DI_percentile calculated for {pair}")
else:
# Not enough data yet, use neutral value
dataframe['di_percentile'] = 0.5
logger.debug(f"FreqAI: Insufficient data for {pair}, using neutral confidence")
# Log DI_values stats for debugging
logger.debug(f"DI_values - min: {dataframe['DI_values'].min():.3f}, "
f"max: {dataframe['DI_values'].max():.3f}, "
f"mean: {dataframe['DI_values'].mean():.3f}")
# Also log do_predict stats
if 'do_predict' in dataframe.columns:
buy_signals = (dataframe['do_predict'] == 1).sum()
sell_signals = (dataframe['do_predict'] == -1).sum()
logger.debug(f"FreqAI signals for {pair}: {buy_signals} buy, {sell_signals} sell")
except KeyError:
# Pair introduced dynamically without FreqAI history/model
logger.debug(f"FreqAI model not ready for {pair} - skipping AI signals")
except Exception as e:
# Extra safety: never let AI crash the strategy
logger.warning(f"FreqAI error for {pair}: {e}")
else:
if self.freqai is None:
logger.debug("FreqAI not initialized for this strategy")
# --- Classical indicators (always run) ---
# SWING high/low for Fibonacci levels
dataframe['swing_high'] = dataframe['high'].rolling(self.buy_swing_period.value).max()
dataframe['swing_low'] = dataframe['low'].rolling(self.buy_swing_period.value).min()
swing_range = dataframe['swing_high'] - dataframe['swing_low']
# LONG (retracement in uptrend)
dataframe['fib_long_0236'] = dataframe['swing_high'] - swing_range * 0.236
dataframe['fib_long_0382'] = dataframe['swing_high'] - swing_range * 0.382
dataframe['fib_long_0618'] = dataframe['swing_high'] - swing_range * 0.618
dataframe['fib_long_0786'] = dataframe['swing_high'] - swing_range * 0.786
# SHORT (retracement in downtrend)
dataframe['fib_short_0236'] = dataframe['swing_low'] + swing_range * 0.236
dataframe['fib_short_0382'] = dataframe['swing_low'] + swing_range * 0.382
dataframe['fib_short_0618'] = dataframe['swing_low'] + swing_range * 0.618
dataframe['fib_short_0786'] = dataframe['swing_low'] + swing_range * 0.786
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Combine your Fibbo strategy with FreqAI predictions.
FreqAI columns are now available in the dataframe.
"""
logger.debug(f"Generating entry signals for {metadata['pair']}")
entry_long_conditions = []
entry_short_conditions = []
# === Your existing Fibbo conditions ===
FIBBO_LONG_ENTRY = (
(dataframe['close'] >= (dataframe[f'fib_long_{str(self.buy_fib_level.value).replace(".", "")}'] * (1 - dataframe['atr_pct']))) &
(dataframe['close'] <= (dataframe[f'fib_long_{str(self.buy_fib_level.value).replace(".", "")}'] * (1 + dataframe['atr_pct'])))
)
FIBBO_SHORT_ENTRY = (
(dataframe['close'] >= (dataframe[f'fib_short_{str(self.buy_fib_level.value).replace(".", "")}'] * (1 - dataframe['atr_pct']))) &
(dataframe['close'] <= (dataframe[f'fib_short_{str(self.buy_fib_level.value).replace(".", "")}'] * (1 + dataframe['atr_pct'])))
)
# Always include FIBBO
entry_long_conditions.append(FIBBO_LONG_ENTRY)
entry_short_conditions.append(FIBBO_SHORT_ENTRY)
# === FreqAI Entry Signals ===
if 'do_predict' in dataframe.columns:
freqai_bullish = (dataframe['do_predict'] == 1)
freqai_bearish = (dataframe['do_predict'] == -1)
if 'di_percentile' in dataframe.columns:
long_conf = dataframe['di_percentile'] > float(self.buy_freqai.value)
short_conf = dataframe['di_percentile'] < float(self.sell_freqai.value)
# Enter LONG when bullish, Enter SHORT when bearish
entry_long_conditions.append(freqai_bullish & long_conf)
entry_short_conditions.append(freqai_bearish & short_conf)
else:
entry_long_conditions.append(freqai_bullish)
entry_short_conditions.append(freqai_bearish)
# Combine entry conditions with AND logic
# Enter if ALL conditions are met
if entry_long_conditions:
dataframe.loc[
reduce(lambda x, y: x & y, entry_long_conditions),
'enter_long'
] = 1
if entry_short_conditions:
dataframe.loc[
reduce(lambda x, y: x & y, entry_short_conditions),
'enter_short'
] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Exit logic combining Fibbo strategy with FreqAI sell signals.
"""
logger.debug(f"Generating exit signals for {metadata['pair']}")
exit_long_conditions = []
exit_short_conditions = []
# === Your existing Fibbo exit conditions ===
FIBBO_LONG_EXIT = (
(dataframe['close'] >= (dataframe[f'fib_long_{str(self.sell_fib_level.value).replace(".", "")}'] * (1 - dataframe['atr_pct']))) &
(dataframe['close'].shift(1) < (dataframe[f'fib_long_{str(self.sell_fib_level.value).replace(".", "")}'] * (1 - dataframe['atr_pct'])))
)
FIBBO_SHORT_EXIT = (
(dataframe['close'] <= (dataframe[f'fib_short_{str(self.sell_fib_level.value).replace(".", "")}'] * (1 + dataframe['atr_pct']))) &
(dataframe['close'].shift(1) > (dataframe[f'fib_short_{str(self.sell_fib_level.value).replace(".", "")}'] * (1 + dataframe['atr_pct'])))
)
# Always include FIBBO
exit_long_conditions.append(FIBBO_LONG_EXIT)
exit_short_conditions.append(FIBBO_SHORT_EXIT)
# === FreqAI Exit Signals ===
if 'do_predict' in dataframe.columns:
freqai_bullish = (dataframe['do_predict'] == 1)
freqai_bearish = (dataframe['do_predict'] == -1)
if 'di_percentile' in dataframe.columns:
long_conf = dataframe['di_percentile'] > float(self.buy_freqai.value)
short_conf = dataframe['di_percentile'] < float(self.sell_freqai.value)
# Exit LONG when bearish, Exit SHORT when bullish
exit_long_conditions.append(freqai_bearish & short_conf)
exit_short_conditions.append(freqai_bullish & long_conf)
else:
exit_long_conditions.append(freqai_bearish)
exit_short_conditions.append(freqai_bullish)
# Combine exit conditions with AND logic
# Exit if ALL condition are met
if exit_long_conditions:
dataframe.loc[
reduce(lambda x, y: x & y, exit_long_conditions),
'exit_long'
] = 1
if exit_short_conditions:
dataframe.loc[
reduce(lambda x, y: x & y, exit_short_conditions),
'exit_short'
] = 1
return dataframe