import logging from functools import reduce import pandas as pd import talib.abstract as ta from pandas import DataFrame from technical import qtpylib from freqtrade.strategy import CategoricalParameter, IStrategy, merge_informative_pair logger = logging.getLogger(__name__) class FreqaiExampleStrategy(IStrategy): """ Example strategy showing how the user connects their own IFreqaiModel to the strategy. Namely, the user uses: self.freqai.start(dataframe, metadata) to make predictions on their data. populate_any_indicators() automatically generates the variety of features indicated by the user in the canonical freqtrade configuration file under config['freqai']. """ minimal_roi = {"0": 0.1, "240": -1} plot_config = { "main_plot": {}, "subplots": { "prediction": {"prediction": {"color": "blue"}}, "do_predict": { "do_predict": {"color": "brown"}, }, }, } process_only_new_candles = True stoploss = -0.05 use_exit_signal = True # this is the maximum period fed to talib (timeframe independent) startup_candle_count: int = 40 can_short = False std_dev_multiplier_buy = CategoricalParameter( [0.75, 1, 1.25, 1.5, 1.75], default=1.25, space="buy", optimize=True) std_dev_multiplier_sell = CategoricalParameter( [0.75, 1, 1.25, 1.5, 1.75], space="sell", default=1.25, optimize=True) def freqai_feature_engineering_indicator_periods(self, dataframe, period, **kwargs): """ This function will be called for all include_timeframes in each indicator_periods_candles (including corr_pairs). After that, the features will be shifted by the number of candles in the include_shifted_candles. :param df: strategy dataframe which will receive the features :param period: period of the indicator - usage example: 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 freqai_feature_engineering_generic(self, dataframe, **kwargs): """ This optional function will be called for all include_timeframes (including corr_pairs). After that, the features will be shifted by the number of candles in the include_shifted_candles. :param df: strategy dataframe which will receive the features dataframe["%-pct-change"] = dataframe["close"].pct_change() """ dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-raw_volume"] = dataframe["volume"] dataframe["%-raw_price"] = dataframe["close"] return dataframe def freqai_feature_engineering_generalized_indicators(self, dataframe, **kwargs): """ This optional function will be called once with the dataframe of the main timeframe. :param df: strategy dataframe which will receive the features usage example: dataframe["%-day_of_week"] = (dataframe["date"].dt.dayofweek + 1) / 7 """ dataframe["%-day_of_week"] = (dataframe["date"].dt.dayofweek + 1) / 7 dataframe["%-hour_of_day"] = (dataframe["date"].dt.hour + 1) / 25 return dataframe def freqai_set_targets(self, dataframe, **kwargs): """ Required function to set the targets for the model. :param df: strategy dataframe which will receive the targets usage example: dataframe["&-target"] = dataframe["close"].shift(-1) / dataframe["close"] """ dataframe["&-s_close"] = ( dataframe["close"] .shift(-self.freqai_info["feature_parameters"]["label_period_candles"]) .rolling(self.freqai_info["feature_parameters"]["label_period_candles"]) .mean() / dataframe["close"] - 1 ) return dataframe def populate_any_indicators_old( self, pair, df, tf, informative=None, set_generalized_indicators=False ): """ DEPRECATED - USE FEATURE ENGINEERING FUNCTIONS INSTEAD Function designed to automatically generate, name and merge features from user indicated timeframes in the configuration file. User can add additional features here, but must follow the naming convention. This method is *only* used in FreqaiDataKitchen class and therefore it is only called if FreqAI is active. :param pair: pair to be used as informative :param df: strategy dataframe which will receive merges from informatives :param tf: timeframe of the dataframe which will modify the feature names :param informative: the dataframe associated with the informative pair """ if informative is None: informative = self.dp.get_pair_dataframe(pair, tf) # first loop is automatically duplicating indicators for time periods for t in self.freqai_info["feature_parameters"]["indicator_periods_candles"]: t = int(t) informative[f"%-{pair}rsi-period_{t}"] = ta.RSI(informative, timeperiod=t) informative[f"%-{pair}mfi-period_{t}"] = ta.MFI(informative, timeperiod=t) informative[f"%-{pair}adx-period_{t}"] = ta.ADX(informative, timeperiod=t) informative[f"%-{pair}sma-period_{t}"] = ta.SMA(informative, timeperiod=t) informative[f"%-{pair}ema-period_{t}"] = ta.EMA(informative, timeperiod=t) bollinger = qtpylib.bollinger_bands( qtpylib.typical_price(informative), window=t, stds=2.2 ) informative[f"{pair}bb_lowerband-period_{t}"] = bollinger["lower"] informative[f"{pair}bb_middleband-period_{t}"] = bollinger["mid"] informative[f"{pair}bb_upperband-period_{t}"] = bollinger["upper"] informative[f"%-{pair}bb_width-period_{t}"] = ( informative[f"{pair}bb_upperband-period_{t}"] - informative[f"{pair}bb_lowerband-period_{t}"] ) / informative[f"{pair}bb_middleband-period_{t}"] informative[f"%-{pair}close-bb_lower-period_{t}"] = ( informative["close"] / informative[f"{pair}bb_lowerband-period_{t}"] ) informative[f"%-{pair}roc-period_{t}"] = ta.ROC(informative, timeperiod=t) informative[f"%-{pair}relative_volume-period_{t}"] = ( informative["volume"] / informative["volume"].rolling(t).mean() ) informative[f"%-{pair}pct-change"] = informative["close"].pct_change() informative[f"%-{pair}raw_volume"] = informative["volume"] informative[f"%-{pair}raw_price"] = informative["close"] indicators = [col for col in informative if col.startswith("%")] # This loop duplicates and shifts all indicators to add a sense of recency to data for n in range(self.freqai_info["feature_parameters"]["include_shifted_candles"] + 1): if n == 0: continue informative_shift = informative[indicators].shift(n) informative_shift = informative_shift.add_suffix("_shift-" + str(n)) informative = pd.concat((informative, informative_shift), axis=1) df = merge_informative_pair(df, informative, self.config["timeframe"], tf, ffill=True) skip_columns = [ (s + "_" + tf) for s in ["date", "open", "high", "low", "close", "volume"] ] df = df.drop(columns=skip_columns) # Add generalized indicators here (because in live, it will call this # function to populate indicators during training). Notice how we ensure not to # add them multiple times if set_generalized_indicators: df["%-day_of_week"] = (df["date"].dt.dayofweek + 1) / 7 df["%-hour_of_day"] = (df["date"].dt.hour + 1) / 25 # user adds targets here by prepending them with &- (see convention below) df["&-s_close"] = ( df["close"] .shift(-self.freqai_info["feature_parameters"]["label_period_candles"]) .rolling(self.freqai_info["feature_parameters"]["label_period_candles"]) .mean() / df["close"] - 1 ) # Classifiers are typically set up with strings as targets: # df['&s-up_or_down'] = np.where( df["close"].shift(-100) > # df["close"], 'up', 'down') # If user wishes to use multiple targets, they can add more by # appending more columns with '&'. User should keep in mind that multi targets # requires a multioutput prediction model such as # templates/CatboostPredictionMultiModel.py, # df["&-s_range"] = ( # df["close"] # .shift(-self.freqai_info["feature_parameters"]["label_period_candles"]) # .rolling(self.freqai_info["feature_parameters"]["label_period_candles"]) # .max() # - # df["close"] # .shift(-self.freqai_info["feature_parameters"]["label_period_candles"]) # .rolling(self.freqai_info["feature_parameters"]["label_period_candles"]) # .min() # ) return df def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # All indicators must be populated by populate_any_indicators() for live functionality # to work correctly. # the model will return all labels created by user in `populate_any_indicators` # (& appended targets), an indication of whether or not the prediction should be accepted, # the target mean/std values for each of the labels created by user in # `populate_any_indicators()` for each training period. dataframe = self.freqai.start(dataframe, metadata, self) for val in self.std_dev_multiplier_buy.range: dataframe[f'target_roi_{val}'] = ( dataframe["&-s_close_mean"] + dataframe["&-s_close_std"] * val ) for val in self.std_dev_multiplier_sell.range: dataframe[f'sell_roi_{val}'] = ( dataframe["&-s_close_mean"] - dataframe["&-s_close_std"] * val ) return dataframe def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: enter_long_conditions = [ df["do_predict"] == 1, df["&-s_close"] > df[f"target_roi_{self.std_dev_multiplier_buy.value}"], ] if enter_long_conditions: df.loc[ reduce(lambda x, y: x & y, enter_long_conditions), ["enter_long", "enter_tag"] ] = (1, "long") enter_short_conditions = [ df["do_predict"] == 1, df["&-s_close"] < df[f"sell_roi_{self.std_dev_multiplier_sell.value}"], ] if enter_short_conditions: df.loc[ reduce(lambda x, y: x & y, enter_short_conditions), ["enter_short", "enter_tag"] ] = (1, "short") return df def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: exit_long_conditions = [ df["do_predict"] == 1, df["&-s_close"] < df[f"sell_roi_{self.std_dev_multiplier_sell.value}"] * 0.25, ] if exit_long_conditions: df.loc[reduce(lambda x, y: x & y, exit_long_conditions), "exit_long"] = 1 exit_short_conditions = [ df["do_predict"] == 1, df["&-s_close"] > df[f"target_roi_{self.std_dev_multiplier_buy.value}"] * 0.25, ] if exit_short_conditions: df.loc[reduce(lambda x, y: x & y, exit_short_conditions), "exit_short"] = 1 return df def get_ticker_indicator(self): return int(self.config["timeframe"][:-1]) def confirm_trade_entry( self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time, entry_tag, side: str, **kwargs, ) -> bool: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = df.iloc[-1].squeeze() if side == "long": if rate > (last_candle["close"] * (1 + 0.0025)): return False else: if rate < (last_candle["close"] * (1 - 0.0025)): return False return True