# source: https://raw.githubusercontent.com/xRetr00/Freqtradex/479e46c35cd0bde6dba4eb34797081ae6ca9af3c/templates/FreqaiExampleHybridStrategy.py import logging import numpy as np # noqa import pandas as pd # noqa import talib.abstract as ta from pandas import DataFrame from technical import qtpylib from freqtrade.strategy import IntParameter, IStrategy, merge_informative_pair # noqa from typing import Dict, List from datetime import datetime, timedelta import json logger = logging.getLogger(__name__) class Github_xRetr00_Freqtradex__FreqaiExampleHybridStrategy__20241129_193154(IStrategy): """ Example of a hybrid FreqAI strat, designed to illustrate how a user may employ FreqAI to bolster a typical Freqtrade strategy. Launching this strategy would be: freqtrade trade --strategy Github_xRetr00_Freqtradex__FreqaiExampleHybridStrategy__20241129_193154 --strategy-path freqtrade/templates --freqaimodel CatboostClassifier --config config_examples/config_freqai.example.json or the user simply adds this to their config: "freqai": { "enabled": true, "purge_old_models": 2, "train_period_days": 15, "identifier": "unique-id", "feature_parameters": { "include_timeframes": [ "3m", "15m", "1h" ], "include_corr_pairlist": [ "BTC/USDT", "ETH/USDT" ], "label_period_candles": 20, "include_shifted_candles": 2, "DI_threshold": 0.9, "weight_factor": 0.9, "principal_component_analysis": false, "use_SVM_to_remove_outliers": true, "indicator_periods_candles": [10, 20] }, "data_split_parameters": { "test_size": 0, "random_state": 1 }, "model_training_parameters": { "n_estimators": 800 } }, Thanks to @smarmau and @johanvulgt for developing and sharing the strategy. """ minimal_roi = { # "120": 0.0, # exit after 120 minutes at break even "60": 0.01, "30": 0.02, "0": 0.04, } plot_config = { "main_plot": { "tema": {}, }, "subplots": { "MACD": { "macd": {"color": "blue"}, "macdsignal": {"color": "orange"}, }, "RSI": { "rsi": {"color": "red"}, }, "Up_or_down": { "&s-up_or_down": {"color": "green"}, }, }, } process_only_new_candles = True stoploss = -0.05 use_exit_signal = True startup_candle_count: int = 30 can_short = True # Hyperoptable parameters buy_rsi = IntParameter(low=1, high=50, default=30, space="buy", optimize=True, load=True) sell_rsi = IntParameter(low=50, high=100, default=70, space="sell", optimize=True, load=True) short_rsi = IntParameter(low=51, high=100, default=70, space="sell", optimize=True, load=True) exit_short_rsi = IntParameter(low=1, high=50, default=30, space="buy", optimize=True, load=True) # Add new class variables for adaptive features market_state: str = "unknown" # tracks current market state last_state_change: datetime = datetime.now() state_parameters: Dict = {} adaptation_threshold: float = 0.15 # 15% change triggers adaptation def __init__(self, config: dict) -> None: super().__init__(config) # Initialize state parameters for different market conditions self.state_parameters = { "trending": { "buy_rsi": 35, "sell_rsi": 75, "short_rsi": 75, "exit_short_rsi": 35, }, "ranging": { "buy_rsi": 25, "sell_rsi": 65, "short_rsi": 65, "exit_short_rsi": 25, }, "volatile": { "buy_rsi": 45, "sell_rsi": 85, "short_rsi": 85, "exit_short_rsi": 45, } } # Load evolved parameters if they exist self._load_evolved_parameters() def _load_evolved_parameters(self) -> None: """Load evolved strategy parameters from disk""" try: with open('evolved_parameters.json', 'r') as f: self.state_parameters = json.load(f) except FileNotFoundError: logger.info("No evolved parameters found, using defaults") def _save_evolved_parameters(self) -> None: """Save evolved strategy parameters to disk""" with open('evolved_parameters.json', 'w') as f: json.dump(self.state_parameters, f) def detect_market_state(self, dataframe: DataFrame) -> str: """ Detect the current market state based on technical indicators """ volatility = dataframe['bb_width'].mean() trend_strength = abs(dataframe['tema'].pct_change(20).mean()) if volatility > self.adaptation_threshold: return "volatile" elif trend_strength > self.adaptation_threshold: return "trending" else: return "ranging" def adapt_parameters(self, new_state: str) -> None: """ Adapt strategy parameters based on market state """ if new_state != self.market_state: logger.info(f"Market state changed from {self.market_state} to {new_state}") self.market_state = new_state self.last_state_change = datetime.now() # Update parameters based on market state state_params = self.state_parameters[new_state] self.buy_rsi.value = state_params["buy_rsi"] self.sell_rsi.value = state_params["sell_rsi"] self.short_rsi.value = state_params["short_rsi"] self.exit_short_rsi.value = state_params["exit_short_rsi"] def evolve_parameters(self, dataframe: DataFrame) -> None: """ Evolve strategy parameters based on recent performance """ # Calculate success rate of recent trades recent_trades = dataframe.tail(100) success_rate = ( len(recent_trades[recent_trades['profit_ratio'] > 0]) / len(recent_trades) ) # Adjust parameters based on success rate if success_rate < 0.4: # Poor performance self.state_parameters[self.market_state] = { k: v * 1.1 for k, v in # Increase thresholds by 10% self.state_parameters[self.market_state].items() } elif success_rate > 0.6: # Good performance self.state_parameters[self.market_state] = { k: v * 0.9 for k, v in # Decrease thresholds by 10% self.state_parameters[self.market_state].items() } self._save_evolved_parameters() 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 feature_engineering_expand_basic( self, dataframe: DataFrame, metadata: dict, **kwargs ) -> DataFrame: """ *Only functional with FreqAI enabled strategies* This function will automatically expand the defined features on the config defined `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 `include_timeframes` * `include_shifted_candles` * `include_corr_pairs` numbers of features added to the model. Features defined here will *not* be automatically duplicated on user defined `indicator_periods_candles` 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 metadata: metadata of current pair dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-ema-200"] = ta.EMA(dataframe, timeperiod=200) """ dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-raw_volume"] = dataframe["volume"] dataframe["%-raw_price"] = dataframe["close"] return dataframe def feature_engineering_standard( self, dataframe: DataFrame, metadata: dict, **kwargs ) -> DataFrame: """ *Only functional with FreqAI enabled strategies* This optional function will be called once with the dataframe of the base timeframe. This is the final function to be called, which means that the dataframe entering this function will contain all the features and columns created by all other freqai_feature_engineering_* functions. This function is a good place to do custom exotic feature extractions (e.g. tsfresh). This function is a good place for any feature that should not be auto-expanded upon (e.g. day of the week). All features must be prepended with `%` to be recognized by FreqAI internals. More details about feature engineering available: https://www.freqtrade.io/en/latest/freqai-feature-engineering :param dataframe: strategy dataframe which will receive the features :param metadata: metadata of current pair usage example: dataframe["%-day_of_week"] = (dataframe["date"].dt.dayofweek + 1) / 7 """ 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: """ *Only functional with FreqAI enabled strategies* Required function to set the targets for the model. All targets must be prepended with `&` to be recognized by the FreqAI internals. More details about feature engineering available: https://www.freqtrade.io/en/latest/freqai-feature-engineering :param dataframe: strategy dataframe which will receive the targets :param metadata: metadata of current pair usage example: dataframe["&-target"] = dataframe["close"].shift(-1) / dataframe["close"] """ self.freqai.class_names = ["down", "up"] dataframe["&s-up_or_down"] = np.where( dataframe["close"].shift(-50) > dataframe["close"], "up", "down" ) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # noqa: C901 # User creates their own custom strat here. Present example is a supertrend # based strategy. dataframe = self.freqai.start(dataframe, metadata, self) # TA indicators to combine with the Freqai targets # RSI dataframe["rsi"] = ta.RSI(dataframe) # Bollinger Bands bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2) dataframe["bb_lowerband"] = bollinger["lower"] dataframe["bb_middleband"] = bollinger["mid"] dataframe["bb_upperband"] = bollinger["upper"] dataframe["bb_percent"] = (dataframe["close"] - dataframe["bb_lowerband"]) / ( dataframe["bb_upperband"] - dataframe["bb_lowerband"] ) dataframe["bb_width"] = (dataframe["bb_upperband"] - dataframe["bb_lowerband"]) / dataframe[ "bb_middleband" ] # TEMA - Triple Exponential Moving Average dataframe["tema"] = ta.TEMA(dataframe, timeperiod=9) # Detect and adapt to market state new_state = self.detect_market_state(dataframe) self.adapt_parameters(new_state) # Evolve parameters periodically (every 24 hours) if datetime.now() - self.last_state_change > timedelta(hours=24): self.evolve_parameters(dataframe) return dataframe def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[ ( # Signal: RSI crosses above 30 (qtpylib.crossed_above(df["rsi"], self.buy_rsi.value)) & (df["tema"] <= df["bb_middleband"]) # Guard: tema below BB middle & (df["tema"] > df["tema"].shift(1)) # Guard: tema is raising & (df["volume"] > 0) # Make sure Volume is not 0 & (df["do_predict"] == 1) # Make sure Freqai is confident in the prediction & # Only enter trade if Freqai thinks the trend is in this direction (df["&s-up_or_down"] == "up") # Add market state specific conditions & ( (self.market_state == "trending") | ( (self.market_state == "ranging") & (df["bb_percent"] < 0.2) # Only buy near bottom of range ) | ( (self.market_state == "volatile") & (df["volume"] > df["volume"].rolling(24).mean()) # Higher volume requirement ) ) ), "enter_long", ] = 1 df.loc[ ( # Signal: RSI crosses above 70 (qtpylib.crossed_above(df["rsi"], self.short_rsi.value)) & (df["tema"] > df["bb_middleband"]) # Guard: tema above BB middle & (df["tema"] < df["tema"].shift(1)) # Guard: tema is falling & (df["volume"] > 0) # Make sure Volume is not 0 & (df["do_predict"] == 1) # Make sure Freqai is confident in the prediction & # Only enter trade if Freqai thinks the trend is in this direction (df["&s-up_or_down"] == "down") ), "enter_short", ] = 1 return df def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[ ( # Signal: RSI crosses above 70 (qtpylib.crossed_above(df["rsi"], self.sell_rsi.value)) & (df["tema"] > df["bb_middleband"]) # Guard: tema above BB middle & (df["tema"] < df["tema"].shift(1)) # Guard: tema is falling & (df["volume"] > 0) # Make sure Volume is not 0 ), "exit_long", ] = 1 df.loc[ ( # Signal: RSI crosses above 30 (qtpylib.crossed_above(df["rsi"], self.exit_short_rsi.value)) & # Guard: tema below BB middle (df["tema"] <= df["bb_middleband"]) & (df["tema"] > df["tema"].shift(1)) # Guard: tema is raising & (df["volume"] > 0) # Make sure Volume is not 0 ), "exit_short", ] = 1 return df