# source: https://raw.githubusercontent.com/hankniel/freqtrade/83ba4265808f598d5e1f810daa10ebea1726a645/user_data/strategies/freqai/prod/ProducerBaseSwingStrategy.py """ Producer Strategy, main strategy that generates features and indicators. """ import pandas as pd import numpy as np import logging from datetime import datetime import os 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 ) 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 feature_engine.datetime import DatetimeFeatures from freqtrade.constants import Config from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter) logger = logging.getLogger(__name__) class Github_hankniel_freqtrade__ProducerBaseSwingStrategy__20251014_102948(IStrategy): process_only_new_candles = True stoploss = -0.7395433873821076 use_exit_signal = True # this is the maximum period fed to talib (timeframe independent) startup_candle_count: int = 400 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: if metadata['tf'] != '15m' and period > 20: return dataframe else: #* 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) #* Inverse Fisher transform on RSI: values [-1.0, 1.0] rsi = 0.1 * (ta.RSI(dataframe, timeperiod=period) - 50) dataframe["%-fisher_rsi-period"] = (np.exp(2 * rsi) - 1) / (np.exp(2 * rsi) + 1) del rsi clean_memory() #* Inverse Fisher transform on RSI normalized: values [0.0, 100.0] # dataframe["%-fisher_rsi_norma-period"] = 50 * (dataframe["%-fisher_rsi-period"] + 1) #* 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"] 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'] dataframe["%-ewo_ema_long_signal-period"] = ewo_ema['ewo_long_signal'] dataframe["%-ewo_ema_short_signal-period"] = ewo_ema['ewo_short_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'] # dataframe["%-ewo_sma_long_signal-period"] = ewo_sma['ewo_long_signal'] dataframe["%-ewo_sma_short_signal-period"] = ewo_sma['ewo_short_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'] dataframe["%-fisher_cg_sig-period"] = fisher_cg_df['fisher_sig'] 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() #* 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) #* 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: 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"] #* 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"] return dataframe def feature_engineering_standard( self, dataframe: DataFrame, metadata: dict, **kwargs ) -> DataFrame: 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) # Static features dtf = DatetimeFeatures( variables='date', features_to_extract=[ 'day_of_week', 'hour', # 'minute', # 'day_of_month', # 'month', # 'quarter', # 'week', # 'weekend', # 'month_start', # 'month_end', ], drop_original=True ) dtf_features = dtf.fit_transform(dataframe[['date']]) dataframe[[f'%-{feature}' for feature in dtf.get_feature_names_out()]] = dtf_features # =========================================================================== # 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' ) 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' ) 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' ) 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] #* Multithreading load results = load_multiple_data_types(configs, start_date=start_date, end_date=end_date) #* Metrics data for col in results['metrics'].columns: if col.startswith('sum_taker_long_short_vol_ratio'): results['metrics'][col] = results['metrics'][col].shift(1) dataframe = dataframe.merge(results['metrics'], on='date', how='left') for col in results['metrics'].drop(columns=['date']).columns: 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: # 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: dataframe = self.freqai.start(dataframe, metadata, self) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: 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 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']