# source: https://raw.githubusercontent.com/kainyang/freqtrade_strategies/6953037cadbbac7820669bfbfa3b1613ecd0edc0/divergence.py # --- freqtrade strategy: Github_kainyang_freqtrade_strategies__divergence__20250521_095753 (Updated to use DecimalParameter) --- # Required imports from functools import reduce 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, CategoricalParameter, merge_informative_pair) logger = logging.getLogger(__name__) class Github_kainyang_freqtrade_strategies__divergence__20250521_095753(IStrategy): INTERFACE_VERSION = 3 # Strategy timeframe timeframe = '1m' # 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: current_rsi_buffer = self.rsi_buffer.value for tf_info_str in ['3m', '5m']: 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. Defining empty signal columns.") dataframe[f'inf_{tf_info_str}_bullish_div'] = False dataframe[f'inf_{tf_info_str}_bearish_div'] = False 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] if existing_cols_in_inf_df: dataframe = merge_informative_pair(dataframe, inf_df[existing_cols_in_inf_df], self.timeframe, tf_info_str, ffill=True, append_prefix=True) else: for col_base_name in columns_to_merge: dataframe[f'inf_{tf_info_str}_{col_base_name}'] = False return dataframe # --- Populate indicators for base timeframe (1m) --- def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: 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) 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