# source: https://raw.githubusercontent.com/hankniel/freqtrade/83ba4265808f598d5e1f810daa10ebea1726a645/user_data/strategies/freqai/prod/ProducerBaseSwingStrategy4h.py # TODO: add features related to funding rates # TODO: add mathematical features (fourier transform, etc.) import pandas as pd import numpy as np import logging from datetime import datetime import talib.abstract as ta from pandas import DataFrame from technical import qtpylib from user_data.utils.indicators.indicator_helpers import ( CMF, macd_ultimate, VWAPB, EWO_EMA, EWO_SMA, fisher_cg, breakouts, SMI, hurst_exponent, detect_pullback, supertrend, wavetrend ) from user_data.utils.constants import DEFAULT_LABEL_NAME from user_data.utils.helpers import clean_memory from user_data.utils.label_helpers import tripple_barrier from user_data.utils.market_data.binance import DataConfig, load_multiple_data_types from user_data.utils.market_data.fear_and_greed import FearGreedIndex from freqtrade.constants import Config from freqtrade.strategy import IStrategy import warnings warnings.simplefilter(action='ignore', category=FutureWarning) logger = logging.getLogger(__name__) class Github_hankniel_freqtrade__ProducerBaseSwingStrategy4h__20251014_102948(IStrategy): """ Example strategy showing how the user connects their own IFreqaiModel to the strategy. Warning! This is a showcase of functionality, which means that it is designed to show various functions of FreqAI and it runs on all computers. We use this showcase to help users understand how to build a strategy, and we use it as a benchmark to help debug possible problems. This means this is *not* meant to be run live in production. """ # =========================================================================== # STRATEGY CONFIG # =========================================================================== timeframe = '4h' process_only_new_candles = True stoploss = -0.99 use_exit_signal = True # this is the maximum period fed to talib (timeframe independent) startup_candle_count: int = 200 can_short = True #* Custom params lag_period = 30 def __init__(self, config: Config) -> None: super().__init__(config) self.identifier = self.config['freqai']['identifier'] # self.minimal_roi = {"0": self.config.get('minimal_roi', {}).get('0', None)} 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. Access metadata such as the current pair/timeframe with: `metadata["pair"]` `metadata["tf"]` 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) """ # #* New pairs # if (metadata['pair'].split('/')[0] == 'ONDO' # and metadata['tf'] == '1d' and period > 50 # ): # logger.warning(f"Skipping {metadata['pair']} {metadata['tf']} for period {period} due to insufficient data") # return dataframe #* New pairs if (metadata['pair'].split('/')[0] in ['ENA', 'TAO', 'TON'] and ((metadata['tf'] == '1d' and period >= 50) or period > 50) ): logger.warning(f"Skipping {metadata['pair']} {metadata['tf']} for period {period} due to insufficient data") return dataframe #* New pairs if (metadata['pair'].split('/')[0] in ['RENDER', 'S'] and (metadata['tf'] == '1d' or period > 20) ): logger.warning(f"Skipping {metadata['pair']} {metadata['tf']} for period {period} due to insufficient data") return dataframe #* Momentum Indicators # ------------------------------------ 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["%-rsi_pct_change-period"] = dataframe["%-rsi-period"].pct_change(periods=period) dataframe["%-mfi_pct_change-period"] = dataframe["%-mfi-period"].pct_change(periods=period) dataframe["%-adx_pct_change-period"] = dataframe["%-adx-period"].pct_change(periods=period) #* Plus Directional Indicator / Movement dataframe["%-plus_dm-period"] = ta.PLUS_DM(dataframe, timeperiod=period) dataframe["%-plus_di-period"] = ta.PLUS_DI(dataframe, timeperiod=period) #* Minus Directional Indicator / Movement dataframe["%-minus_dm-period"] = ta.MINUS_DM(dataframe, timeperiod=period) dataframe["%-minus_di-period"] = ta.MINUS_DI(dataframe, timeperiod=period) #* Aroon, Aroon Oscillator aroon = ta.AROON(dataframe, timeperiod=period) dataframe["%-aroonup-period"] = aroon["aroonup"] dataframe["%-aroondown-period"] = aroon["aroondown"] dataframe["%-aroonosc-period"] = ta.AROONOSC(dataframe, timeperiod=period) del aroon clean_memory() #* Awesome Oscillator try: if period <= 20: if metadata['tf'] == '15m': fast_period = max(3, int(period//3)) dataframe["%-ao-period"] = qtpylib.awesome_oscillator(dataframe, fast=fast_period, slow=period*2) elif metadata['tf'] in ['1h', '4h']: dataframe["%-ao-period"] = qtpylib.awesome_oscillator(dataframe, fast=int(period//2), slow=period*3) elif metadata['tf'] in ['1d', '1w']: dataframe["%-ao-period"] = qtpylib.awesome_oscillator(dataframe, fast=int(period//1.25), slow=period*3) except: dataframe["%-ao-period"] = None #* Keltner Channel keltner = qtpylib.keltner_channel(dataframe, window=period) dataframe["%-kc_upperband-period"] = keltner["upper"] dataframe["%-kc_lowerband-period"] = keltner["lower"] dataframe["%-kc_middleband-period"] = keltner["mid"] dataframe["%-kc_percent-period"] = ( (dataframe["close"] - dataframe["%-kc_lowerband-period"]) / (dataframe["%-kc_upperband-period"] - dataframe["%-kc_lowerband-period"]) ) dataframe["%-kc_width-period"] = ( (dataframe["%-kc_upperband-period"] - dataframe["%-kc_lowerband-period"]) / dataframe["%-kc_middleband-period"] ) del keltner clean_memory() #* Ultimate Oscillator dataframe["%-uo-period"] = ta.ULTOSC(dataframe, timeperiod=period) #* Commodity Channel Index: values [Oversold:-100, Overbought:100] dataframe["%-cci-period"] = ta.CCI(dataframe, timeperiod=period) #* Stochastic Slow stoch = ta.STOCH(dataframe, timeperiod=period) dataframe["%-slowd-period"] = stoch["slowd"] dataframe["%-slowk-period"] = stoch["slowk"] del stoch clean_memory() #* Stochastic Fast stoch_fast = ta.STOCHF(dataframe, timeperiod=period) # dataframe["%-fastd-period"] = stoch_fast["fastd"] dataframe["%-fastk-period"] = stoch_fast["fastk"] del stoch_fast clean_memory() #* Stochastic RSI stoch_rsi = ta.STOCHRSI(dataframe, timeperiod=period) dataframe["%-fastd_rsi-period"] = stoch_rsi["fastd"] dataframe["%-fastk_rsi-period"] = stoch_rsi["fastk"] del stoch_rsi clean_memory() #* ROC dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period) # Overlap Studies # ------------------------------------ #* Simple Moving Average Percentage #! Use sma pct change, don't use sma raw as feature dataframe["sma-period"] = ta.SMA(dataframe, timeperiod=period) dataframe["%-sma_pct_change-period"] = dataframe["sma-period"].pct_change(periods=period) dataframe = dataframe.drop(columns=['sma-period']) #* Exponential Moving Average Percentage #! Use ema pct change, don't use ema raw as feature dataframe["ema-period"] = ta.EMA(dataframe, timeperiod=period) dataframe["%-ema_pct_change-period"] = dataframe["ema-period"].pct_change(periods=period) dataframe = dataframe.drop(columns=['ema-period']) #* Bollinger Bands 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"] #? Not using as feature 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["%-bb_percent-period"] = ( (dataframe["close"] - dataframe["%-bb_lowerband-period"]) / (dataframe["%-bb_upperband-period"] - dataframe["%-bb_lowerband-period"]) ) del bollinger clean_memory() #* Bollinger Bands - Weighted (EMA based instead of SMA) weighted_bollinger = qtpylib.weighted_bollinger_bands( qtpylib.typical_price(dataframe), window=period, stds=2.2 ) dataframe["%-wbb_upperband-period"] = weighted_bollinger["upper"] dataframe["%-wbb_lowerband-period"] = weighted_bollinger["lower"] dataframe["%-wbb_middleband-period"] = weighted_bollinger["mid"] dataframe["%-wbb_percent-period"] = ( (dataframe["close"] - dataframe["%-wbb_lowerband-period"]) / (dataframe["%-wbb_upperband-period"] - dataframe["%-wbb_lowerband-period"]) ) dataframe["%-wbb_width-period"] = ( (dataframe["%-wbb_upperband-period"] - dataframe["%-wbb_lowerband-period"]) / dataframe["%-wbb_middleband-period"] ) del weighted_bollinger clean_memory() #* TEMA - Triple Exponential Moving Average dataframe["%-tema-period"] = ta.TEMA(dataframe, timeperiod=period) dataframe["%-tema_pct_change-period"] = dataframe["%-tema-period"].pct_change(periods=period) #* Volume # ------------------------------------ dataframe["%-relative_volume-period"] = ( dataframe["volume"] / dataframe["volume"].rolling(period).mean() ) # =========================================================================== # CUSTOM INDICATORS # =========================================================================== #* CMF - Chaikin Money Flow dataframe["%-cmf-period"] = CMF(dataframe, timeperiod=period) #* MACD - SMA fast_period = period slow_period = period * 2 signal_period = int(period * 0.75) sma_macd = macd_ultimate( dataframe, fast_period=fast_period, slow_period=slow_period, signal_period=signal_period, ma_type='SMA' ) dataframe["%-sma_macd-period"] = sma_macd["macd"] dataframe["%-sma_macd_signal-period"] = sma_macd["signal"] dataframe["%-sma_macd_histogram-period"] = sma_macd["histogram"] # dataframe["%-sma_macd_trend_up-period"] = sma_macd["trend_up"] # dataframe["%-sma_macd_trend_down-period"] = sma_macd["trend_down"] # dataframe["%-sma_macd_cross_up-period"] = sma_macd["macd_cross_up"] # dataframe["%-sma_macd_cross_down-period"] = sma_macd["macd_cross_down"] # dataframe["%-sma_macd_cross_up_a-period"] = sma_macd["macd_cross_up_a"] # dataframe["%-sma_macd_cross_down_b-period"] = sma_macd["macd_cross_down_b"] # for col in ['%-sma_macd_trend_up-period', '%-sma_macd_trend_down-period', # '%-sma_macd_cross_up-period', '%-sma_macd_cross_down-period', # '%-sma_macd_cross_up_a-period', '%-sma_macd_cross_down_b-period']: # dataframe[col] = dataframe[col].astype(bool) del sma_macd clean_memory() #* VWAPB vwapb = VWAPB(dataframe, window_size=period) dataframe["%-vwap_high-period"] = vwapb["vwap_high"] dataframe["%-vwap_low-period"] = vwapb["vwap_low"] # dataframe["%-vwap-period"] = vwapb["vwap"] dataframe["%-vwap_high_low_pct-period"] = 100 * (vwapb["vwap_high"] / vwapb["vwap_low"]) - 1 # dataframe["%-vwap_middle_low_pct-period"] = 100 * (vwapb["vwap"] / vwapb["vwap_low"]) - 1 # dataframe["%-vwap_high_middle_pct-period"] = 100 * (vwapb["vwap_high"] / vwapb["vwap"]) - 1 dataframe["%-vwap_high_close_pct-period"] = 100 * (vwapb["vwap_high"] / dataframe["close"]) - 1 del vwapb clean_memory() #* EWO_EMA ewo_ema = EWO_EMA(dataframe, slow=int(period//2), fast=int(period*3), signal_period=int(period//2)) dataframe["%-ewo_ema-period"] = ewo_ema['ewo'] dataframe["%-ewo_ema_signal-period"] = ewo_ema['ewo_signal'] del ewo_ema clean_memory() #* EWO_SMA ewo_sma = EWO_SMA(dataframe, slow=int(period//2), fast=int(period*3), signal_period=int(period//2)) dataframe["%-ewo_sma-period"] = ewo_sma['ewo'] dataframe["%-ewo_sma_signal-period"] = ewo_sma['ewo_signal'] del ewo_sma clean_memory() #* Fisher CG fisher_cg_df = fisher_cg(dataframe, periods=period) dataframe["%-fisher_cg-period"] = fisher_cg_df['fisher_cg'] del fisher_cg_df clean_memory() #* Breakouts breakouts_df = breakouts(dataframe, periods=period) dataframe["%-support_level-period"] = breakouts_df['support_level'] dataframe["%-resistance_level-period"] = breakouts_df['resistance_level'] # dataframe["%-support_breakout-period"] = breakouts_df['support_breakout'] # dataframe["%-resistance_breakout-period"] = breakouts_df['resistance_breakout'] # dataframe["%-support_retest-period"] = breakouts_df['support_retest'] dataframe["%-potential_support_retest-period"] = breakouts_df['potential_support_retest'] # dataframe["%-resistance_retest-period"] = breakouts_df['resistance_retest'] dataframe["%-potential_resistance_retest-period"] = breakouts_df['potential_resistance_retest'] del breakouts_df clean_memory() #* Pinbar # pinbar_df = pinbar(dataframe, overbought_threshold=period, oversold_threshold=-period) # dataframe["%-pinbar_buy-period"] = pinbar_df['pinbar_buy'] # dataframe["%-pinbar_sell-period"] = pinbar_df['pinbar_sell'] # del pinbar_df # clean_memory() #* SMI dataframe["%-smi-period"] = SMI(dataframe, k_length=period, d_length=int(period//3)) #* Hurst Exponent dataframe["%-hurst_exponent-period"] = hurst_exponent(dataframe, lookback=period) #* Supertrend supertrend_df = supertrend(dataframe, period=period) dataframe["%-supertrend-period"] = supertrend_df["supertrend"] dataframe["%-trend-period"] = supertrend_df["trend"] #* Wavetrend wavetrend_df = wavetrend(dataframe, n1=period, n2=period*2) dataframe["%-wt1-period"] = wavetrend_df["wt1"] dataframe["%-wt2-period"] = wavetrend_df["wt2"] # #* Pullback # try: # dataframe["%-pullback_1_flag-period"] = detect_pullback( # dataframe, periods=period, outlier_threshold=1.0 # ) # except: # dataframe["%-pullback_1_flag-period"] = None # dataframe["%-pullback_consecutive_1_flag-period"] = np.where( # dataframe["%-pullback_1_flag-period"] == dataframe["%-pullback_1_flag-period"].shift(1), # 1, # 0 # ) # try: # dataframe["%-pullback_2_flag-period"] = detect_pullback( # dataframe, periods=period, outlier_threshold=2.0 # ) # except: # dataframe["%-pullback_2_flag-period"] = None # dataframe["%-pullback_consecutive_2_flag-period"] = np.where( # dataframe["%-pullback_2_flag-period"] == dataframe["%-pullback_2_flag-period"].shift(1), # 1, # 0 # ) 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. Access metadata such as the current pair/timeframe with: `metadata["pair"]` `metadata["tf"]` 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["volume_change"] = np.log(dataframe['volume'] / dataframe['volume'].shift()) dataframe["%-volatility_volume"] = dataframe["volume_change"].rolling(20).std() * np.sqrt(20) # dataframe["%-pct_open_close"] = (dataframe["open"] - dataframe["close"]) / dataframe["close"] dataframe["%-pct_high_close"] = (dataframe["high"] - dataframe["close"]) / dataframe["close"] dataframe["%-pct_low_close"] = (dataframe["low"] - dataframe["close"]) / dataframe["close"] # Non-period indicators # ------------------------------------ #* Parabolic SAR dataframe["%-sar"] = ta.SAR(dataframe) #* Cycle Indicator # ------------------------------------ #* Hilbert Transform Indicator - SineWave if metadata['tf'] == '15m': hilbert = ta.HT_SINE(dataframe) dataframe["%-htsine"] = hilbert["sine"] dataframe["%-htleadsine"] = hilbert["leadsine"] # #* Pattern Recognition - Bullish candlestick patterns # # ------------------------------------ # dataframe["%-CDLHAMMER"] = ta.CDLHAMMER(dataframe) # dataframe["%-CDLINVERTEDHAMMER"] = ta.CDLINVERTEDHAMMER(dataframe) # dataframe["%-CDLDRAGONFLYDOJI"] = ta.CDLDRAGONFLYDOJI(dataframe) # dataframe["%-CDLPIERCING"] = ta.CDLPIERCING(dataframe) # dataframe["%-CDLMORNINGSTAR"] = ta.CDLMORNINGSTAR(dataframe) # # dataframe["%-CDL3WHITESOLDIERS"] = ta.CDL3WHITESOLDIERS(dataframe) # #* Pattern Recognition - Bearish candlestick patterns # # ------------------------------------ # dataframe["%-CDLHANGINGMAN"] = ta.CDLHANGINGMAN(dataframe) # dataframe["%-CDLSHOOTINGSTAR"] = ta.CDLSHOOTINGSTAR(dataframe) # dataframe["%-CDLGRAVESTONEDOJI"] = ta.CDLGRAVESTONEDOJI(dataframe) # # dataframe["%-CDLDARKCLOUDCOVER"] = ta.CDLDARKCLOUDCOVER(dataframe) # # dataframe["%-CDLEVENINGDOJISTAR"] = ta.CDLEVENINGDOJISTAR(dataframe) # # dataframe["%-CDLEVENINGSTAR"] = ta.CDLEVENINGSTAR(dataframe) # #* Pattern Recognition - Bullish/Bearish candlestick patterns # # ------------------------------------ # # dataframe["%-CDL3LINESTRIKE"] = ta.CDL3LINESTRIKE(dataframe) # dataframe["%-CDLSPINNINGTOP"] = ta.CDLSPINNINGTOP(dataframe) # dataframe["%-CDLENGULFING"] = ta.CDLENGULFING(dataframe) # dataframe["%-CDLHARAMI"] = ta.CDLHARAMI(dataframe) # dataframe["%-CDL3OUTSIDE"] = ta.CDL3OUTSIDE(dataframe) # dataframe["%-CDL3INSIDE"] = ta.CDL3INSIDE(dataframe) #* Chart type # ------------------------------------ # #* Heikin Ashi Strategy # heikinashi = qtpylib.heikinashi(dataframe) # dataframe["%-ha_open"] = heikinashi["open"] # dataframe["%-ha_close"] = heikinashi["close"] # dataframe["%-ha_high"] = heikinashi["high"] # dataframe["%-ha_low"] = heikinashi["low"] 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. Access metadata such as the current pair with: `metadata["pair"]` 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 """ lag_columns = {} for lag in range(1, self.lag_period + 1): # if lag == self.freqai_info['feature_parameters']['label_period_candles']: # break lag_columns[f'%-close_pct_change_lag_{lag}'] = dataframe['close'].pct_change(periods=lag) # lag_columns[f'%-open_pct_change_lag_{lag}'] = dataframe['open'].pct_change(periods=lag) lag_columns[f'%-high_pct_change_lag_{lag}'] = dataframe['high'].pct_change(periods=lag) lag_columns[f'%-low_pct_change_lag_{lag}'] = dataframe['low'].pct_change(periods=lag) lag_columns[f'%-volume_pct_change_lag_{lag}'] = dataframe['volume'].pct_change(periods=lag) new_df = pd.DataFrame(lag_columns, index=dataframe.index) dataframe = pd.concat([dataframe, new_df], axis=1) # Position within recent ranges # dataframe['%-close_position_20'] = ( # dataframe['close'] - dataframe['low'].rolling(20).min() # ) / ( # dataframe['high'].rolling(20).max() - dataframe['low'].rolling(20).min() # ) # dataframe['%-volume_position_20'] = ( # dataframe['volume'] - dataframe['volume'].rolling(20).min() # ) / ( # dataframe['volume'].rolling(20).max() - dataframe['volume'].rolling(20).min() # ) dataframe['%-day_of_week'] = dataframe['date'].dt.dayofweek # =========================================================================== # MARKET DATA # =========================================================================== metrics_config = DataConfig( symbol=metadata['pair'].split('/')[0] + '/USDT', data_type='metrics', base_frequency='5m', target_frequency=self.timeframe, data_dir='data/custom_data', wait=True ) klines_config = DataConfig( symbol=metadata['pair'].split('/')[0] + '/USDT', data_type='klines', base_frequency=self.timeframe, target_frequency=self.timeframe, data_dir='data/custom_data', wait=True ) premium_klines_config = DataConfig( symbol=metadata['pair'].split('/')[0] + '/USDT', data_type='premiumIndexKlines', base_frequency=self.timeframe, target_frequency=self.timeframe, data_dir='data/custom_data', wait=True ) configs = [metrics_config, klines_config, premium_klines_config] start_date = dataframe['date'].iloc[0] # Add 1 candles buffer to start date if self.timeframe.endswith('m'): start_date = start_date - pd.Timedelta(minutes=1 * int(self.timeframe[:-1])) elif self.timeframe.endswith('h'): start_date = start_date - pd.Timedelta(hours=1 * int(self.timeframe[:-1])) elif self.timeframe.endswith('d'): start_date = start_date - pd.Timedelta(days=1 * int(self.timeframe[:-1]))\ end_date = dataframe['date'].iloc[-1] # Add 1 candles buffer to end date if self.timeframe.endswith('m'): end_date = end_date + pd.Timedelta(minutes=1 * int(self.timeframe[:-1])) elif self.timeframe.endswith('h'): end_date = end_date + pd.Timedelta(hours=1 * int(self.timeframe[:-1])) elif self.timeframe.endswith('d'): end_date = end_date + pd.Timedelta(days=1 * int(self.timeframe[:-1])) #* Multithreading load results = load_multiple_data_types(configs, start_date=start_date, end_date=end_date) #* Metrics data dataframe = dataframe.merge(results['metrics'], on='date', how='left') for col in results['metrics'].columns: if col != 'date': dataframe.rename(columns={col: f'%-{col}'}, inplace=True) #* Klines data results['klines'].rename( columns={ 'count': 'klines_count' }, inplace=True ) dataframe = dataframe.merge( results['klines'][['date', 'quote_volume', 'klines_count', 'taker_buy_volume', 'taker_buy_quote_volume']], on='date', how='left' ) for col in ['quote_volume', 'klines_count', 'taker_buy_volume', 'taker_buy_quote_volume']: dataframe.rename(columns={col: f'%-{col}'}, inplace=True) #* PremiumIndexKlines data results['premiumIndexKlines'].rename( columns={ 'open': 'premium_open', 'high': 'premium_high', 'low': 'premium_low', 'close': 'premium_close', }, inplace=True ) dataframe = dataframe.merge( results['premiumIndexKlines'][['date', 'premium_open', 'premium_high', 'premium_low', 'premium_close']], on='date', how='left' ) for col in ['premium_open', 'premium_high', 'premium_low', 'premium_close']: dataframe.rename(columns={col: f'%-{col}'}, inplace=True) del results clean_memory() #* Fear and greed data fg = FearGreedIndex(data_dir='data/custom_data') fg_data = fg.get_fear_greed_data(start_date, end_date, timeframe=self.timeframe) dataframe = dataframe.merge(fg_data, on='date', how='left') for col in fg_data.columns: if col != 'date': dataframe.rename(columns={col: f'%-{col}'}, inplace=True) del fg_data clean_memory() #! Check duplicate if dataframe.duplicated(subset=['date']).any(): dataframe = dataframe.drop_duplicates(subset=['date'], keep='first') dataframe = dataframe.reset_index(drop=True) self.log.error(f"Duplicate dates found and removed for {metadata['pair']} {self.timeframe}") #! Don't remove this dataframe["%-index"] = dataframe.index 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. Access metadata such as the current pair with: `metadata["pair"]` 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"] """ # 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 # freqai/prediction_models/CatboostRegressorMultiTarget.py, # freqtrade trade --freqaimodel CatboostRegressorMultiTarget # dataframe['&-s_range'] = ( # dataframe['close'] # .shift(-self.freqai_info['feature_parameters']['label_period_candles']) # .rolling(self.freqai_info['feature_parameters']['label_period_candles']) # .max() # - # dataframe['close'] # .shift(-self.freqai_info['feature_parameters']['label_period_candles']) # .rolling(self.freqai_info['feature_parameters']['label_period_candles']) # .min() # ) # Get label name from the config file label = f'&-s_{self.freqai_info.get('label_config', {}).get('label_name', DEFAULT_LABEL_NAME)}' #* Tripple barrier method dataframe[label] = ( dataframe["close"] .shift(-self.freqai_info['label_config']['window']) .rolling(self.freqai_info['label_config']['window'] + self.freqai_info['label_config']['bias']) .apply(tripple_barrier, kwargs={ 'upper_pct': self.freqai_info['label_config']['upper_pct'], 'lower_pct': self.freqai_info['label_config']['lower_pct'] } ) ) dataframe[label] = dataframe[label].map({-1.0: 'short', 0.0: 'hold', 1.0: 'long'}) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # All indicators must be populated by feature_engineering_*() functions # the model will return all labels created by user in `set_freqai_targets()` # (& 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 # `set_freqai_targets()` for each training period. dataframe = self.freqai.start(dataframe, metadata, self) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['enter_long'] = 0 dataframe['enter_tag'] = None return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['exit_long'] = 0 dataframe['exit_tag'] = None return dataframe 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 # def custom_stake_amount(self, pair: str, 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: # # Using static stake amount for now, 10% available balance each entry # return self.wallets.get_total_stake_amount() / self.config["max_open_trades"] 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: return self.config['leverage'] # def calculate_trade_return( # self, dataframe: DataFrame, metadata: dict, **kwargs # ): # # Get entry/exit pair # entry = dataframe[dataframe['enter_long'] == 1]['close'] # exit = dataframe[dataframe['exit_long'] == 1]['close'] # returns = [] # leverage = 20 # Static leverage # for entry, exit in zip(entry, exit): # returns.append((exit - entry) / entry) # TODO: Implement custom_exit(), custom_stake_amount(), leverage(), custom_stoploss(), adjust_trade_position()