# source: https://raw.githubusercontent.com/kainyang/freqtrade_strategies/cc22bcf6c652b68d09f4e8dc87b3ec8c35dd0d7e/divergence.py # --- freqtrade strategy: Github_kainyang_freqtrade_strategies__divergence__20250521_161832 (Updated to use DecimalParameter) --- # Required imports import logging import talib.abstract as ta from pandas import DataFrame, Series from freqtrade.exchange import timeframe_to_prev_date from freqtrade.persistence import Trade # For type hinting in custom_exit # MODIFIED IMPORT: Replaced FloatParameter with DecimalParameter from freqtrade.strategy import (IStrategy, IntParameter, DecimalParameter, merge_informative_pair) logger = logging.getLogger(__name__) class Github_kainyang_freqtrade_strategies__divergence__20250521_161832(IStrategy): INTERFACE_VERSION = 3 # Strategy timeframe timeframe = '1m' # Informative timeframes informative_timeframes = ['3m', '5m'] # ROI table (Futures context) minimal_roi = {"0": 1.50} # Stoploss stoploss = -0.30 use_custom_exit = True process_only_new_candles = True # --- Strategy Parameters --- rsi_period = 14 div_lookback_1m = IntParameter(10, 30, default=15, space="buy sell") div_lookback_3m = IntParameter(8, 25, default=12, space="buy sell") div_lookback_5m = IntParameter(6, 20, default=10, space="buy sell") # MODIFIED PARAMETER DEFINITION: Using DecimalParameter instead of FloatParameter rsi_buffer = DecimalParameter(0.0, 3.0, default=0.5, decimals=1, space="buy sell") # --- Helper function to detect divergence --- def _check_divergence(self, dataframe: DataFrame, price_col_name: str, osc_col_name: str, lookback: int, divergence_type: str, rsi_bf: float) -> Series: if lookback <= 0: return Series([False] * len(dataframe), index=dataframe.index) if price_col_name not in dataframe.columns or osc_col_name not in dataframe.columns: logger.warning(f"Missing required columns for divergence check: {price_col_name} or {osc_col_name} in _check_divergence. Columns: {dataframe.columns.tolist()}") return Series([False] * len(dataframe), index=dataframe.index) price_shifted = dataframe[price_col_name].shift(lookback) osc_shifted = dataframe[osc_col_name].shift(lookback) if divergence_type == 'bullish': price_condition = dataframe[price_col_name] <= price_shifted osc_condition = dataframe[osc_col_name] > (osc_shifted + rsi_bf) elif divergence_type == 'bearish': price_condition = dataframe[price_col_name] >= price_shifted osc_condition = dataframe[osc_col_name] < (osc_shifted - rsi_bf) else: return Series([False] * len(dataframe), index=dataframe.index) return price_condition & osc_condition # --- Populate indicators for informative timeframes (3m, 5m) --- def informative_populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: logger.info(f"ENTERING informative_populate_indicators for {metadata['pair']}. Initial DF columns: {dataframe.columns.tolist()}") current_rsi_buffer = self.rsi_buffer.value # Pre-initialize all expected informative columns at the very beginning for tf_info_str_prefix_init in self.informative_timeframes: for signal_suffix_init in ['bullish_div', 'bearish_div']: col_name_init = f'inf_{tf_info_str_prefix_init}_{signal_suffix_init}' if col_name_init not in dataframe.columns: dataframe[col_name_init] = False for tf_info_str in self.informative_timeframes: inf_df = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=tf_info_str) if inf_df.empty: logger.info(f"Informative dataframe for {metadata['pair']} timeframe {tf_info_str} is empty. Pre-initialized columns will be used.") # Columns for this tf_info_str should already be pre-initialized to False. # No merge will happen, so we continue to the next timeframe. continue inf_df[f'rsi_{tf_info_str}'] = ta.RSI(inf_df['close'], timeperiod=self.rsi_period) lookback_val = 0 if tf_info_str == '3m': lookback_val = self.div_lookback_3m.value elif tf_info_str == '5m': lookback_val = self.div_lookback_5m.value inf_df[f'bullish_div'] = self._check_divergence(inf_df, 'low', f'rsi_{tf_info_str}', lookback_val, 'bullish', current_rsi_buffer) inf_df[f'bearish_div'] = self._check_divergence(inf_df, 'high', f'rsi_{tf_info_str}', lookback_val, 'bearish', current_rsi_buffer) columns_to_merge = [f'bullish_div', f'bearish_div'] existing_cols_in_inf_df = [ col for col in columns_to_merge if col in inf_df.columns and not inf_df[col].empty # Check if series has data points ] if existing_cols_in_inf_df and not inf_df.empty: # ensure inf_df has rows and selected columns have data informative_data_to_merge = inf_df[existing_cols_in_inf_df] if not informative_data_to_merge.empty: dataframe = merge_informative_pair(dataframe, informative_data_to_merge, self.timeframe, tf_info_str, ffill=True, append_prefix=True) # After merge, explicitly ensure the target columns for THIS timeframe exist. # This is because merge_informative_pair might return a new dataframe # and might not create a column if its source in informative_data_to_merge was all False/NaN. expected_signal_cols_for_tf = [f'inf_{tf_info_str}_bullish_div', f'inf_{tf_info_str}_bearish_div'] for col_name in expected_signal_cols_for_tf: if col_name not in dataframe.columns: logger.warning( f"Strategy Dev: Column '{col_name}' was not found in dataframe after merge for {tf_info_str} on {metadata['pair']}. " f"Adding it as False. This might indicate an issue with merge_informative_pair or source data." ) dataframe[col_name] = False else: logger.info(f"Informative data slice for {tf_info_str} on {metadata['pair']} became empty after selecting columns. Merge skipped. Pre-initialized columns used.") else: logger.info(f"No valid data in inf_df for {tf_info_str} on {metadata['pair']} to merge or inf_df was empty. Merge skipped. Pre-initialized columns used.") # Final check before returning - this is mostly for sanity checking during development. # The loop above should handle individual TFs. for tf_final_check in self.informative_timeframes: for sig_final_check in ['bullish_div', 'bearish_div']: final_col_name = f'inf_{tf_final_check}_{sig_final_check}' if final_col_name not in dataframe.columns: # This would be unexpected if the logic above is correct. logger.error(f"CRITICAL STRATEGY ERROR: Column '{final_col_name}' is MISSING from dataframe for {metadata['pair']} just before returning from informative_populate_indicators. Setting to False.") dataframe[final_col_name] = False logger.info(f"EXITING informative_populate_indicators for {metadata['pair']}. Final DF columns: {dataframe.columns.tolist()}") return dataframe # --- Populate indicators for base timeframe (1m) --- def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: logger.info(f"ENTERING populate_indicators for {metadata['pair']}. Received DF columns: {dataframe.columns.tolist()}") current_rsi_buffer = self.rsi_buffer.value lookback_1m = self.div_lookback_1m.value dataframe['rsi_1m'] = ta.RSI(dataframe['close'], timeperiod=self.rsi_period) dataframe['bullish_div_1m'] = self._check_divergence(dataframe, 'low', 'rsi_1m', lookback_1m, 'bullish', current_rsi_buffer) dataframe['bearish_div_1m'] = self._check_divergence(dataframe, 'high', 'rsi_1m', lookback_1m, 'bearish', current_rsi_buffer) logger.info(f"EXITING populate_indicators for {metadata['pair']}. Final DF columns: {dataframe.columns.tolist()}") return dataframe # --- Populate Entry Signals --- def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: col_b_1m = 'bullish_div_1m' col_b_3m = 'inf_3m_bullish_div' col_b_5m = 'inf_5m_bullish_div' col_s_1m = 'bearish_div_1m' col_s_3m = 'inf_3m_bearish_div' col_s_5m = 'inf_5m_bearish_div' required_cols = [col_b_1m, col_b_3m, col_b_5m, col_s_1m, col_s_3m, col_s_5m] for col in required_cols: if col not in dataframe.columns: logger.warning(f"Column '{col}' not found for entry check on {metadata['pair']}. Filling with False. Available: {dataframe.columns.tolist()}") dataframe[col] = False bullish_cond_1m_3m = dataframe[col_b_1m] & dataframe[col_b_3m] bullish_cond_1m_5m = dataframe[col_b_1m] & dataframe[col_b_5m] # Corrected typo from previous full version dataframe.loc[ (bullish_cond_1m_3m | bullish_cond_1m_5m) & # Corrected typo used here (dataframe['volume'] > 0), 'enter_long'] = 1 bearish_cond_1m_3m = dataframe[col_s_1m] & dataframe[col_s_3m] bearish_cond_1m_5m = dataframe[col_s_1m] & dataframe[col_s_5m] dataframe.loc[ (bearish_cond_1m_3m | bearish_cond_1m_5m) & (dataframe['volume'] > 0), 'enter_short'] = 1 return dataframe # --- Populate Exit Signals (Based on 3-min TF Divergence Only) --- def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: col_b_3m = 'inf_3m_bullish_div' col_s_3m = 'inf_3m_bearish_div' required_cols_for_exit = [col_b_3m, col_s_3m] for col in required_cols_for_exit: if col not in dataframe.columns: logger.warning(f"Exit signal column '{col}' for 3m timeframe not found for pair {metadata['pair']}. Filling with False. Available: {dataframe.columns.tolist()}") dataframe[col] = False if col_s_3m in dataframe.columns and not dataframe[col_s_3m].empty: dataframe.loc[dataframe[col_s_3m], 'exit_long'] = 1 else: if 'exit_long' not in dataframe.columns: dataframe['exit_long'] = 0 if col_b_3m in dataframe.columns and not dataframe[col_b_3m].empty: dataframe.loc[dataframe[col_b_3m], 'exit_short'] = 1 else: if 'exit_short' not in dataframe.columns: dataframe['exit_short'] = 0 return dataframe # --- Custom Exit for 30-candle rule --- def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float, current_profit: float, **kwargs): timeframe_seconds = self.timeframe_to_seconds(self.timeframe) if timeframe_seconds > 0: trade_duration_candles = (current_time - trade.open_date_utc).total_seconds() // timeframe_seconds else: trade_duration_candles = 0 if trade_duration_candles >= 30: logger.info(f"Exiting {pair} (Trade ID: {trade.id}) due to trade duration ({trade_duration_candles} candles) exceeding 30 candles.") return 'time_exit_30_candles' return None