# source: https://raw.githubusercontent.com/mkajnar/HighProfitStrategy/ba6f95f17337348906192f80b5be0404c45d5aab/user_data/strategies/SRChartCandleStrat.py import datetime import logging from datetime import datetime from datetime import timedelta, timezone from functools import reduce from typing import Optional import numpy as np import talib.abstract as ta from pandas import DataFrame from talib import CDLDOJI import freqtrade.vendor.qtpylib.indicators as qtpylib from freqtrade.persistence import Trade from freqtrade.strategy import DecimalParameter, IntParameter from freqtrade.strategy.interface import IStrategy class Elliot: base_nb_candles_sell = 22 base_nb_candles_buy = 12 ewo_low = 10.289 ewo_high = 3.001 fast_ewo = 50 slow_ewo = 200 low_offset = 0.987 rsi_buy = 58 high_offset = 1.014 buy_ema_cofi = 0.97 buy_fastk = 20 buy_fastd = 20 buy_adx = 30 buy_ewo_high = 3.55 def use_hyperopts(self, base_nb_candles_sell, base_nb_candles_buy, low_offset, ewo_high, ewo_low, rsi_buy, high_offset, buy_ema_cofi, buy_fastk, buy_fastd, buy_adx, buy_ewo_high): self.base_nb_candles_sell = base_nb_candles_sell self.base_nb_candles_buy = base_nb_candles_buy self.low_offset = low_offset self.ewo_high = ewo_high self.ewo_low = ewo_low self.rsi_buy = rsi_buy self.high_offset = high_offset self.buy_ema_cofi = buy_ema_cofi self.buy_fastk = buy_fastk self.buy_fastd = buy_fastd self.buy_adx = buy_adx self.buy_ewo_high = buy_ewo_high def populate_indicators(self, dataframe: DataFrame): dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['rsi_fast'] = ta.RSI(dataframe, timeperiod=4) dataframe['EWO'] = self.__EWO(dataframe, self.fast_ewo, self.slow_ewo) for val in self.base_nb_candles_buy.range: dataframe[f'ma_buy_{val}'] = ta.EMA(dataframe, timeperiod=val) for val in self.base_nb_candles_sell.range: dataframe[f'ma_sell_{val}'] = ta.EMA(dataframe, timeperiod=val) return dataframe def populate_entry_trend_v1(self, dataframe: DataFrame, conditions: list): buy1ewo = ( (dataframe['rsi_fast'] < 35) & (dataframe['close'] < (dataframe[f'ma_buy_{self.base_nb_candles_buy.value}'] * self.low_offset.value)) & (dataframe['EWO'] > self.ewo_high.value) & (dataframe['rsi'] < self.rsi_buy.value) & (dataframe['volume'] > 0) & (dataframe['close'] < ( dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) ) dataframe.loc[buy1ewo, 'enter_tag'] += 'buy1eworsi_' conditions.append(buy1ewo) return (dataframe, conditions) def populate_entry_trend_v2(self, dataframe: DataFrame, conditions: list): buy2ewo = ( (dataframe['rsi_fast'] < 35) & (dataframe['close'] < (dataframe[f'ma_buy_{self.base_nb_candles_buy.value}'] * self.low_offset.value)) & (dataframe['EWO'] < self.ewo_low.value) & (dataframe['volume'] > 0) & (dataframe['close'] < ( dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) ) dataframe.loc[buy2ewo, 'enter_tag'] += 'buy2ewo_' conditions.append(buy2ewo) return (dataframe, conditions) def populate_entry_trend_cofi(self, dataframe: DataFrame, conditions: list): is_cofi = ( (dataframe['open'] < dataframe['ema_8'] * self.buy_ema_cofi.value) & (qtpylib.crossed_above(dataframe['fastk'], dataframe['fastd'])) & (dataframe['fastk'] < self.buy_fastk.value) & (dataframe['fastd'] < self.buy_fastd.value) & (dataframe['adx'] > self.buy_adx.value) & (dataframe['EWO'] > self.buy_ewo_high.value) ) dataframe.loc[is_cofi, 'enter_tag'] += 'cofi_' conditions.append(is_cofi) return (dataframe, conditions) def __EWO(self, dataframe: DataFrame, ema_length=5, ema2_length=3): df = dataframe.copy() ema1 = ta.EMA(df, timeperiod=ema_length) ema2 = ta.EMA(df, timeperiod=ema2_length) return (ema1 - ema2) / df['close'] * 100 class Github_mkajnar_HighProfitStrategy__SRChartCandleStrat__20240110_145548(IStrategy): INTERFACE_VERSION = 3 max_safety_orders = 3 lowest_prices = {} highest_prices = {} price_drop_percentage = {} pairs_close_to_high = [] support_dict = {} resistance_dict = {} out_open_trades_limit = 10 stoploss = -0.9 trailing_stop = True trailing_stop_positive = 0.003 trailing_stop_positive_offset = 0.008 trailing_only_offset_is_reached = True use_exit_signal = True exit_profit_only = True ignore_roi_if_entry_signal = True position_adjustment_enable = True order_time_in_force = { 'entry': 'gtc', 'exit': 'gtc' } minimal_roi = { "0": 0.5, "120": 0.3, "240": 0.1, "360": 0.07, "480": 0.05, "720": 0.03, "960": 0.01, "1440": 0.005, "2880": 0.003, "4320": 0.001, "5760": 0.000 } order_types = { 'entry': 'market', 'exit': 'market', 'stoploss': 'market', 'stoploss_on_exchange': False } timeframed_drops = { '1m': -0.01, '5m': -0.05, '15m': -0.05, '30m': -0.075, '1h': -0.1 } timeframes_in_minutes = { '1m': 1, '5m': 5, '15m': 15, '30m': 30, '1h': 60, '4h': 240, '1d': 1440 } elliot = Elliot() base_nb_candles_sell = IntParameter(8, 20, default=elliot.base_nb_candles_sell, space='sell', optimize=False) base_nb_candles_buy = IntParameter(8, 20, default=elliot.base_nb_candles_buy, space='buy', optimize=False) low_offset = DecimalParameter(0.975, 0.995, default=elliot.low_offset, space='buy', optimize=True) ewo_high = DecimalParameter(3.0, 5, default=elliot.ewo_high, space='buy', optimize=True) ewo_low = DecimalParameter(-20.0, -7.0, default=-elliot.ewo_low, space='buy', optimize=True) rsi_buy = IntParameter(30, 70, default=elliot.rsi_buy, space='buy', optimize=False) high_offset = DecimalParameter(1.000, 1.010, default=elliot.high_offset, space='sell', optimize=True) buy_ema_cofi = DecimalParameter(0.96, 0.98, default=elliot.buy_ema_cofi, optimize=True) buy_fastk = IntParameter(20, 30, default=elliot.buy_fastk, optimize=True) buy_fastd = IntParameter(20, 30, default=elliot.buy_fastd, optimize=True) buy_adx = IntParameter(20, 30, default=elliot.buy_adx, optimize=True) buy_ewo_high = DecimalParameter(2, 12, default=elliot.buy_ewo_high, optimize=True) elliot.use_hyperopts(base_nb_candles_sell, base_nb_candles_buy, low_offset, ewo_high, ewo_low, rsi_buy, high_offset, buy_ema_cofi, buy_fastk, buy_fastd, buy_adx, buy_ewo_high) @property def protections(self): return [ { "method": "CooldownPeriod", "stop_duration_candles": 1 }, { "method": "MaxDrawdown", "lookback_period_candles": 48, "trade_limit": 20, "stop_duration_candles": 4, "max_allowed_drawdown": 0.2 }, { "method": "StoplossGuard", "lookback_period_candles": 24, "trade_limit": 4, "stop_duration_candles": 2, "only_per_pair": False }, { "method": "LowProfitPairs", "lookback_period_candles": 6, "trade_limit": 2, "stop_duration_candles": 60, "required_profit": 0.02 }, { "method": "LowProfitPairs", "lookback_period_candles": 24, "trade_limit": 4, "stop_duration_candles": 2, "required_profit": 0.01 } ] def version(self) -> str: return "Github_mkajnar_HighProfitStrategy__SRChartCandleStrat__20240110_145548 v1.0" def dynamic_stop_loss_take_profit(self, dataframe: DataFrame) -> DataFrame: atr = ta.ATR(dataframe, timeperiod=14) dataframe['stop_loss'] = dataframe['low'].shift(1) - atr.shift(1) * 0.8 dataframe['take_profit'] = dataframe['high'].shift(1) + atr.shift(1) * 2.5 return dataframe def calculate_dca_price(self, base_value, decline, target_percent): return (((base_value / 100) * abs(decline)) / target_percent) * 100 def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: float, max_stake: float, **kwargs): if current_rate is None: return None # Update the current time to the current UTC time pct_threshold = self.timeframed_drops[self.timeframe] # logging.info(f"Timeframe: {self.timeframe}, Pct Threshold: {pct_threshold}") current_time = datetime.utcnow() try: # Get an analyzed dataframe for the specified trading pair and timeframe dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) df = dataframe.copy() except Exception as e: # Log an error if there's an issue retrieving the dataframe and exit the method logging.error(f"Error getting analyzed dataframe: {e}") return None if self.is_pair_locked(pair=trade.pair): return None last_candle = df.iloc[-1] # Initialize the lowest price for the trade pair if it's not already set if trade.pair not in self.lowest_prices: self.lowest_prices[trade.pair] = trade.open_rate if trade.pair not in self.price_drop_percentage: self.price_drop_percentage[trade.pair] = {"last_drop_time": current_time, "last_drop_rate": current_rate} # Update the lowest price if the current rate is lower if current_rate < self.lowest_prices[trade.pair]: self.lowest_prices[trade.pair] = current_rate # Calculate the percentage drop from the opening rate price_drop = (self.lowest_prices[trade.pair] - trade.open_rate) / trade.open_rate # Record the time if the price drop exceeds 5% if ((price_drop <= pct_threshold) and (self.price_drop_percentage[trade.pair].get("last_drop_time", current_time) != current_time)): if "last_drop_rate" in self.price_drop_percentage[trade.pair].keys(): last = self.price_drop_percentage[trade.pair].get("last_drop_rate") if last is not None: if current_rate < last: self.price_drop_percentage[trade.pair]["last_drop_time"] = current_time self.price_drop_percentage[trade.pair]["last_drop_rate"] = current_rate # If a 5% drop has been recorded and it's been more than 3 hours since then if self.price_drop_percentage[trade.pair].get("last_drop_time") > current_time: time_since_last_drop = current_time - self.price_drop_percentage[trade.pair]["last_drop_time"] if time_since_last_drop.total_seconds() / 3600 >= 3: logging.info(f"Locking {trade.pair}") self.lock_pair(trade.pair, until=datetime.now(timezone.utc) + timedelta( minutes=8 * 60), reason='STILLDROP_LOCK') # self.locked.append(trade.pair) return None # Avoid further DCA last_buy_order = None count_of_buys = sum(order.ft_order_side == 'buy' and order.status == 'closed' for order in trade.orders) for order in reversed(trade.orders): if order.ft_order_side == 'buy' and order.status == 'closed': last_buy_order = order break # Record the time when a 5% drop was first noted if trade.pair not in self.price_drop_percentage: self.price_drop_percentage[trade.pair] = {"last_drop_time": None, "last_drop_rate": None} if self.max_safety_orders >= count_of_buys: # Calculate the percentage difference between the last buy order and the current rate pct_diff = self.calculate_percentage_difference(original_price=last_buy_order.price, current_price=current_rate) # Check if the percentage difference is less than the threshold value if pct_diff < pct_threshold: if last_buy_order and current_rate < last_buy_order.price: # Check RSI conditions for DCA rsi_value = last_candle['rsi'] # Assuming RSI is part of the dataframe w_rsi = last_candle['weighted_rsi'] # Assuming weighted RSI is part of the dataframe if rsi_value <= w_rsi: # Log information about the store # logging.info(f'AP1 {trade.pair}, Profit: {current_profit}, Stake {trade.stake_amount}') # Get the total stake amount in the wallet total_stake_amount = self.wallets.get_total_stake_amount() # Calculate the amount for the next bet using DCA (Dollar Cost Averaging) calculated_dca_stake = self.calculate_dca_price(base_value=trade.stake_amount, decline=current_profit * 100, target_percent=1) # Adjust the bet size if it is higher than the available balance while calculated_dca_stake >= total_stake_amount: calculated_dca_stake = calculated_dca_stake / 4 # Log information about the adjusted bet # logging.info(f'AP2 {trade.pair}, DCA: {calculated_dca_stake}') # Return the adjusted bet size return calculated_dca_stake # Return None if no conditions are met to adjust the trade position return None def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: self.prepare(dataframe=dataframe) self.dynamic_stop_loss_take_profit(dataframe=dataframe) self.prepare_adx(dataframe=dataframe) self.prepare_rsi(dataframe=dataframe) self.prepare_stochastic(dataframe=dataframe) self.prepare_ema_diff_buy_signal(dataframe=dataframe) self.prepare_sma(dataframe=dataframe) self.prepare_ewo(dataframe=dataframe) self.prepare_doji(dataframe=dataframe) self.prepare_fibs(dataframe=dataframe) self.calculate_support_resistance_dicts(metadata['pair'], dataframe) dataframe = self.elliot.populate_indicators(dataframe=dataframe) return dataframe pass def populate_entry_trend_sr(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Checking the distance to support and resistance if metadata['pair'] in self.support_dict and metadata['pair'] in self.resistance_dict: supports = self.support_dict[metadata['pair']] resistances = self.resistance_dict[metadata['pair']] if supports and resistances: # Calculating the nearest support and resistance level for each candle dataframe['nearest_support'] = dataframe['close'].apply( lambda x: min([support for support in supports if support <= x], default=x, key=lambda support: abs(x - support)) ) dataframe['nearest_resistance'] = dataframe['close'].apply( lambda x: min([resistance for resistance in resistances if resistance >= x], default=x, key=lambda resistance: abs(x - resistance)) ) # Calculation of the percentage difference between the price and the nearest support/resistance dataframe['distance_to_support_pct'] = (dataframe['nearest_support'] - dataframe['close']) / dataframe[ 'close'] * 100 dataframe['distance_to_resistance_pct'] = (dataframe['nearest_resistance'] - dataframe['close']) / \ dataframe['close'] * 100 # Generating buy signals based on support and resistance buy_threshold = 0.1 # 0.1 % dataframe.loc[ (dataframe['distance_to_support_pct'] >= 0) & (dataframe['distance_to_support_pct'] <= buy_threshold) & (dataframe['distance_to_resistance_pct'] >= buy_threshold), 'buy_signal' ] = 1 dataframe.loc[ (dataframe['distance_to_support_pct'] >= 0) & (dataframe['distance_to_support_pct'] <= buy_threshold) & (dataframe['distance_to_resistance_pct'] >= buy_threshold), 'enter_tag' ] += 'sr_buy_mid' # Remove helper columns dataframe.drop( ['nearest_support', 'nearest_resistance', 'distance_to_support_pct', 'distance_to_resistance_pct'], axis=1, inplace=True) # Adding conditions for EMA and volume dataframe.loc[(dataframe['volume'] > 0) & (dataframe['ema_diff_buy_signal'].astype(int) > 0), 'buy_ema'] = 1 dataframe.loc[ (dataframe['volume'] > 0) & (dataframe['ema_diff_buy_signal'].astype(int) > 0), 'enter_tag'] += 'ema_dbs_' # Generating buy signals only if both conditions are met dataframe.loc[(dataframe['buy_signal'] == 1) & (dataframe['buy_ema'] == 1) & ( dataframe['rsi'] <= dataframe['weighted_rsi']), 'enter_long'] = 1 # Remove helper columns if 'buy_support' in dataframe.columns: dataframe.drop(['buy_support'], axis=1, inplace=True) if 'buy_ema' in dataframe.columns: dataframe.drop(['buy_ema'], axis=1, inplace=True) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [] if dataframe is not None: last_candle = dataframe.iloc[-1] if last_candle['doji_candle'] == 1: logging.info(f"Doji detected on {metadata['pair']}") if not self.is_pair_locked(pair=metadata['pair']): self.lock_pair(pair=metadata['pair'], until=datetime.now(timezone.utc) + timedelta( minutes=self.timeframe_to_minutes(self.timeframe) * 14), reason='DOJI_LOCK') return dataframe # Elliot Waves (dataframe, conditions) = self.elliot.populate_entry_trend_v1(dataframe, conditions) (dataframe, conditions) = self.elliot.populate_entry_trend_v2(dataframe, conditions) (dataframe, conditions) = self.elliot.populate_entry_trend_cofi(dataframe, conditions) # Checking the distance to the nearest resistance dataframe = self.populate_entry_trend_sr(dataframe=dataframe, metadata=metadata) # Checking the distance to the support if metadata['pair'] in self.support_dict: s = self.support_dict[metadata['pair']] if s: # Calculating the nearest support level for each candle that is below the current price dataframe['nearest_support'] = dataframe['close'].apply( lambda x: min([support for support in s if support <= x], default=x, key=lambda support: abs(x - support)) ) if 'nearest_support' in dataframe.columns: # Calculation of the percentage difference between the price and the nearest support dataframe['distance_to_support_pct'] = (dataframe['nearest_support'] - dataframe['close']) / \ dataframe['close'] * 100 # Generating buy signals based on support buy_threshold = 0.1 # 0.1 % dataframe.loc[ (dataframe['distance_to_support_pct'] >= 0) & (dataframe['distance_to_support_pct'] <= buy_threshold), 'buy_support' ] = 1 dataframe.loc[ (dataframe['distance_to_support_pct'] >= 0) & (dataframe['distance_to_support_pct'] <= buy_threshold), 'enter_tag' ] += 'sr_buy' # Remove helper columns dataframe.drop(['nearest_support', 'distance_to_support_pct'], axis=1, inplace=True) # Adding conditions for EMA and volume dataframe.loc[(dataframe['volume'] > 0) & (dataframe['ema_diff_buy_signal'].astype(int) > 0), 'buy_ema'] = 1 dataframe.loc[ (dataframe['volume'] > 0) & ( dataframe['ema_diff_buy_signal'].astype(int) > 0), 'enter_tag'] += 'ema_dbs_' # Generating buy signals only if both conditions are met dataframe.loc[(dataframe['buy_support'] == 1) & (dataframe['buy_ema'] == 1) & ( dataframe['rsi'] <= dataframe['weighted_rsi']), 'enter_long'] = 1 # Remove helper columns if 'buy_support' in dataframe.columns: dataframe.drop(['buy_support'], axis=1, inplace=True) if 'buy_ema' in dataframe.columns: dataframe.drop(['buy_ema'], axis=1, inplace=True) dont_buy_conditions = [ (dataframe['enter_long'].shift(1) == 1 & (dataframe['sma_2'].shift(1) < dataframe['sma_2'])) ] if conditions: final_condition = reduce(lambda x, y: x | y, conditions) dataframe.loc[final_condition, 'enter_long'] = 1 if dont_buy_conditions: for condition in dont_buy_conditions: dataframe.loc[condition, 'enter_long'] = 0 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( (dataframe['close'] > dataframe['fib_618']) & (dataframe['sma_50'].shift(1) > dataframe['sma_50']) ), 'exit_tag'] = 'fib_618_sma_50' dataframe.loc[ ( (dataframe['close'] > dataframe['fib_618']) & (dataframe['sma_50'].shift(1) > dataframe['sma_50']) ), 'exit_long'] = 1 dataframe.loc[ ( (dataframe['close'] < dataframe['stop_loss'].shift(1)) | (dataframe['close'] > dataframe['take_profit'].shift(1)) ), 'exit_tag'] = 'psl' dataframe.loc[ ( (dataframe['close'] < dataframe['stop_loss'].shift(1)) | (dataframe['close'] > dataframe['take_profit'].shift(1)) ), 'exit_long'] = 1 dataframe.loc[:, 'exit_short'] = 0 return dataframe def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: # try: # dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) # df = dataframe.copy() # except Exception as e: # logging.error(f"Error getting analyzed dataframe: {e}") # return None result = Trade.get_open_trade_count() < self.out_open_trades_limit return result pass def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float, current_profit: float, **kwargs): if current_profit < -0.15 and (current_time - trade.open_date_utc).days >= 7: return 'unclog' def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime, **kwargs) -> bool: exit_reason = f"{exit_reason}_{trade.enter_tag}" if 'unclog' in exit_reason or 'force' in exit_reason: # logging.info(f"CTE - FORCE or UNCLOG, EXIT") return True current_profit = trade.calc_profit_ratio(rate) dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) if current_profit >= 0.005 and 'psl' in exit_reason: logging.info(f"CTE - PSL EXIT: {pair}, {current_profit}, {rate}, {exit_reason}, {amount}") return True # Checking if the current high is higher than the open of the last candle last_candle = dataframe.iloc[-1] if last_candle['high'] > last_candle['open']: # logging.info(f"CTE - Cena stále roste (high > open), HOLD") return False if current_profit <= 0.005: return False # Current EMA values ema_8_current = dataframe['ema_8'].iat[-1] ema_14_current = dataframe['ema_14'].iat[-1] # EMA values e previous candle ema_8_previous = dataframe['ema_8'].iat[-2] ema_14_previous = dataframe['ema_14'].iat[-2] # EMA difference calculation between current and previous candle diff_current = abs(ema_8_current - ema_14_current) diff_previous = abs(ema_8_previous - ema_14_previous) # Calculation of percentage change between diff_current and diff_previous diff_change_pct = (diff_previous - diff_current) / diff_previous if current_profit >= 0.0025: if ema_8_current <= ema_14_current and diff_change_pct >= 0.025: # logging.info(f"CTE - EMA 8 {ema_8_current} <= EMA 14 {ema_14_current} with decrease in difference >= 3%, EXIT") return True elif ema_8_current > ema_14_current and diff_current > diff_previous: # logging.info(f"CTE - EMA 8 {ema_8_current} > EMA 14 {ema_14_current} with increasing difference, HOLD") return False else: # logging.info(f"CTE - Conditions not met, EXIT") return True else: return False pass def prepare(self, dataframe: DataFrame): if 'enter_tag' not in dataframe.columns: dataframe.loc[:, 'enter_tag'] = '' if 'exit_tag' not in dataframe.columns: dataframe.loc[:, 'exit_tag'] = '' if 'enter_long' not in dataframe.columns: dataframe.loc[:, 'enter_long'] = 0 if 'exit_long' not in dataframe.columns: dataframe.loc[:, 'exit_long'] = 0 pass def prepare_rsi(self, dataframe): dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['rsi_fast'] = ta.RSI(dataframe, timeperiod=4) dataframe['rsi_slow'] = ta.RSI(dataframe, timeperiod=20) # Calculation of weighted average RSI for the last 300 candles weights = np.linspace(1, 0, 300) # Weights from 1 (newest) to 0 (oldest) weights /= weights.sum() # Normalizing the weights so that their sum is 1 dataframe['weighted_rsi'] = dataframe['rsi'].rolling(window=300).apply( lambda x: np.sum(weights * x[-300:]), raw=False ) pass def prepare_stochastic(self, dataframe): stoch_fast = ta.STOCHF(dataframe, 5, 3, 0, 3, 0) dataframe['fastd'] = stoch_fast['fastd'] dataframe['fastk'] = stoch_fast['fastk'] pass def prepare_ema_diff_buy_signal(self, dataframe): ema_8 = ta.EMA(dataframe, timeperiod=8) ema_14 = ta.EMA(dataframe, timeperiod=14) # dataframe['ema_21'] = ta.EMA(dataframe, timeperiod=21) # dataframe['ema_50'] = ta.EMA(dataframe, timeperiod=50) condition = ema_8 > ema_14 percentage_difference = 100 * (ema_8 - ema_14).abs() / ema_14 ema_pct_diff = percentage_difference.where(condition, -percentage_difference) prev_ema_pct_diff = ema_pct_diff.shift(1) crossover_up = (ema_8.shift(1) < ema_14.shift(1)) & (ema_8 > ema_14) close_to_crossover_up = (ema_8 < ema_14) & (ema_8.shift(1) < ema_14.shift(1)) & (ema_8 > ema_8.shift(1)) ema_buy_signal = ((ema_pct_diff < 0) & (prev_ema_pct_diff < 0) & (ema_pct_diff.abs() < prev_ema_pct_diff.abs())) dataframe['ema_diff_buy_signal'] = ( (ema_buy_signal | crossover_up | close_to_crossover_up) & (dataframe['rsi'] <= 55) & ( dataframe['volume'] > 0)) dataframe['ema_8'] = ema_8 dataframe['ema_14'] = ema_14 pass def prepare_sma(self, dataframe): dataframe['sma_2'] = ta.SMA(dataframe, timeperiod=2) dataframe['sma_9'] = ta.SMA(dataframe, timeperiod=9) dataframe['sma_50'] = ta.SMA(dataframe, timeperiod=50) pass def calculate_percentage_difference(self, original_price, current_price): percentage_diff = ((current_price - original_price) / original_price) return percentage_diff def timeframe_to_minutes(self, timeframe): """Converts the timeframe to minutes.""" if timeframe.endswith('m'): return int(timeframe[:-1]) elif timeframe.endswith('h'): return int(timeframe[:-1]) * 60 elif timeframe.endswith('d'): return int(timeframe[:-1]) * 1440 else: raise ValueError("Unknown timeframe: {}".format(timeframe)) def prepare_ewo(self, dataframe): dataframe['EWO'] = self.__EWO(dataframe, Elliot.fast_ewo, Elliot.slow_ewo) pass def __EWO(self, dataframe: DataFrame, ema_length=5, ema2_length=3): df = dataframe.copy() ema1 = ta.EMA(df, timeperiod=ema_length) ema2 = ta.EMA(df, timeperiod=ema2_length) return (ema1 - ema2) / df['close'] * 100 def prepare_doji(self, dataframe): dataframe['doji_candle'] = ( CDLDOJI(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close']) > 0).astype(int) pass def calculate_support_resistance_dicts(self, pair: str, df: DataFrame): try: df = self.calculate_support_resistance(df) self.support_dict[pair] = self.calculate_dynamic_clusters(df['support'].dropna().tolist(), 4) self.resistance_dict[pair] = self.calculate_dynamic_clusters(df['resistance'].dropna().tolist(), 4) except Exception as ex: logging.error(str(ex)) def pivot_points(self, high, low, period=10): pivot_high = high.rolling(window=2 * period + 1, center=True).max() pivot_low = low.rolling(window=2 * period + 1, center=True).min() return high == pivot_high, low == pivot_low def calculate_support_resistance(self, df, period=10, loopback=290): high_pivot, low_pivot = self.pivot_points(df['high'], df['low'], period) df['resistance'] = df['high'][high_pivot] df['support'] = df['low'][low_pivot] return df def calculate_dynamic_clusters(self, values, max_clusters): """ Dynamically calculates the averaged clusters from the given list of values. Args: values (list): List of values to cluster. max_clusters (int): Maximum number of clusters to create. Returns: list: List of average values for each cluster created. """ def cluster_values(threshold): sorted_values = sorted(values) clusters = [] current_cluster = [sorted_values[0]] for value in sorted_values[1:]: if value - current_cluster[-1] <= threshold: current_cluster.append(value) else: clusters.append(current_cluster) current_cluster = [value] clusters.append(current_cluster) return clusters threshold = 0.3 # Initial threshold value while True: clusters = cluster_values(threshold) if len(clusters) <= max_clusters: break threshold += 0.3 # Calculation of means for each cluster cluster_averages = [round(sum(cluster) / len(cluster), 2) for cluster in clusters] return cluster_averages def prepare_fibs(self, dataframe): high_max = dataframe['high'].rolling(window=30).max() low_min = dataframe['low'].rolling(window=30).min() diff = high_max - low_min dataframe['fib_236'] = high_max - 0.236 * diff dataframe['fib_382'] = high_max - 0.382 * diff dataframe['fib_500'] = high_max - 0.500 * diff dataframe['fib_618'] = high_max - 0.618 * diff dataframe['fib_786'] = high_max - 0.786 * diff pass def prepare_adx(self, dataframe): dataframe['adx'] = ta.ADX(dataframe) pass