# source: https://raw.githubusercontent.com/remiotore/freqtrade/44beaeb6a420cd8e9f2e4ea93e11d6cfa192ee03/strategies/EI3v2_kalidem.py from freqtrade.strategy.interface import IStrategy from typing import Dict, List from functools import reduce from pandas import DataFrame import talib.abstract as ta import numpy as np import freqtrade.vendor.qtpylib.indicators as qtpylib import datetime from technical.util import resample_to_interval, resampled_merge from datetime import datetime, timedelta from freqtrade.persistence import Trade from freqtrade.strategy import ( stoploss_from_open, merge_informative_pair, DecimalParameter, IntParameter, CategoricalParameter, ) import technical.indicators as ftt import math import logging logger = logging.getLogger(__name__) def EWO(dataframe, ema_length=5, ema2_length=3): df = dataframe.copy() ema1 = ta.EMA(df, timeperiod=ema_length) ema2 = ta.EMA(df, timeperiod=ema2_length) emadif = (ema1 - ema2) / df["close"] * 100 return emadif class Github_remiotore_freqtrade__EI3v2_kalidem__20260111_210550(IStrategy): INTERFACE_VERSION = 3 """ minimal_roi = { "0": 0.08, "20": 0.04, "40": 0.032, "87": 0.016, "201": 0, "202": -1 } """ buy_params = { "base_nb_candles_buy": 12, "rsi_buy": 58, "ewo_high": 3.001, "ewo_low": -10.289, "low_offset": 0.987, "lambo2_ema_14_factor": 0.981, "lambo2_enabled": True, "lambo2_rsi_14_limit": 39, "lambo2_rsi_4_limit": 44, "buy_adx": 20, "buy_fastd": 20, "buy_fastk": 22, "buy_ema_cofi": 0.98, "buy_ewo_high": 4.179, } sell_params = { "base_nb_candles_sell": 22, "high_offset": 1.014, "high_offset_2": 1.01, } @property def protections(self): return [ {"method": "CooldownPeriod", "stop_duration_candles": 5}, { "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, }, ] minimal_roi = {"0": 0.99} stoploss = -0.99 base_nb_candles_buy = IntParameter( 8, 20, default=buy_params["base_nb_candles_buy"], space="buy", optimize=False ) base_nb_candles_sell = IntParameter( 8, 20, default=sell_params["base_nb_candles_sell"], space="sell", optimize=False ) low_offset = DecimalParameter( 0.985, 0.995, default=buy_params["low_offset"], space="buy", optimize=True ) high_offset = DecimalParameter( 1.005, 1.015, default=sell_params["high_offset"], space="sell", optimize=True ) high_offset_2 = DecimalParameter( 1.01, 1.02, default=sell_params["high_offset_2"], space="sell", optimize=True ) lambo2_ema_14_factor = DecimalParameter( 0.8, 1.2, decimals=3, default=buy_params["lambo2_ema_14_factor"], space="buy", optimize=True, ) lambo2_rsi_4_limit = IntParameter( 5, 60, default=buy_params["lambo2_rsi_4_limit"], space="buy", optimize=True ) lambo2_rsi_14_limit = IntParameter( 5, 60, default=buy_params["lambo2_rsi_14_limit"], space="buy", optimize=True ) fast_ewo = 50 slow_ewo = 200 ewo_low = DecimalParameter( -20.0, -8.0, default=buy_params["ewo_low"], space="buy", optimize=True ) ewo_high = DecimalParameter( 3.0, 3.4, default=buy_params["ewo_high"], space="buy", optimize=True ) rsi_buy = IntParameter( 30, 70, default=buy_params["rsi_buy"], space="buy", optimize=False ) trailing_stop = True trailing_stop_positive = 0.001 trailing_stop_positive_offset = 0.012 trailing_only_offset_is_reached = True is_optimize_cofi = False buy_ema_cofi = DecimalParameter(0.96, 0.98, default=0.97, optimize=is_optimize_cofi) buy_fastk = IntParameter(20, 30, default=20, optimize=is_optimize_cofi) buy_fastd = IntParameter(20, 30, default=20, optimize=is_optimize_cofi) buy_adx = IntParameter(20, 30, default=30, optimize=is_optimize_cofi) buy_ewo_high = DecimalParameter(2, 12, default=3.553, optimize=is_optimize_cofi) use_exit_signal = True exit_profit_only = True exit_profit_offset = 0.01 ignore_roi_if_entry_signal = False order_time_in_force = {"entry": "gtc", "exit": "gtc"} timeframe = "5m" inf_1h = "1h" process_only_new_candles = True startup_candle_count = 200 plot_config = { "main_plot": {"ma_buy": {"color": "orange"}, "ma_sell": {"color": "orange"}} } def custom_exit( self, pair: str, trade: "Trade", current_time: "datetime", current_rate: float, current_profit: float, **kwargs, ): if current_profit < -0.04 and (current_time - trade.open_date_utc).days >= 4: return "unclog" def informative_pairs(self): pairs = self.dp.current_whitelist() informative_pairs = [(pair, "1h") for pair in pairs] if self.config["stake_currency"] in [ "USDT", "BUSD", "USDC", "DAI", "TUSD", "PAX", "USD", "EUR", "GBP", ]: btc_info_pair = f"BTC/{self.config['stake_currency']}" else: btc_info_pair = "BTC/USDT" informative_pairs.append((btc_info_pair, self.timeframe)) informative_pairs.append((btc_info_pair, self.inf_1h)) return informative_pairs def pump_dump_protection(self, dataframe: DataFrame, metadata: dict) -> DataFrame: df36h = dataframe.copy().shift(432) # TODO FIXME: This assumes 5m timeframe df24h = dataframe.copy().shift(288) # TODO FIXME: This assumes 5m timeframe dataframe["volume_mean_short"] = dataframe["volume"].rolling(4).mean() dataframe["volume_mean_long"] = df24h["volume"].rolling(48).mean() dataframe["volume_mean_base"] = df36h["volume"].rolling(288).mean() dataframe["volume_change_percentage"] = ( dataframe["volume_mean_long"] / dataframe["volume_mean_base"] ) dataframe["rsi_mean"] = dataframe["rsi"].rolling(48).mean() dataframe["pnd_volume_warn"] = np.where( dataframe["volume_mean_short"] / dataframe["volume_mean_long"] > 5.0, -1, 0 ) return dataframe def base_tf_btc_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["price_trend_long"] = ( dataframe["close"].rolling(8).mean() / dataframe["close"].shift(8).rolling(144).mean() ) ignore_columns = ["date", "open", "high", "low", "close", "volume"] dataframe.rename( columns=lambda s: f"btc_{s}" if s not in ignore_columns else s, inplace=True ) return dataframe def info_tf_btc_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["rsi_8"] = ta.RSI(dataframe, timeperiod=8) ignore_columns = ["date", "open", "high", "low", "close", "volume"] dataframe.rename( columns=lambda s: f"btc_{s}" if s not in ignore_columns else s, inplace=True ) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: if self.config["stake_currency"] in ["USDT", "BUSD"]: btc_info_pair = f"BTC/{self.config['stake_currency']}" else: btc_info_pair = "BTC/USDT" btc_info_tf = self.dp.get_pair_dataframe(btc_info_pair, self.inf_1h) btc_info_tf = self.info_tf_btc_indicators(btc_info_tf, metadata) dataframe = merge_informative_pair( dataframe, btc_info_tf, self.timeframe, self.inf_1h, ffill=True ) drop_columns = [ f"{s}_{self.inf_1h}" for s in ["date", "open", "high", "low", "close", "volume"] ] dataframe.drop( columns=dataframe.columns.intersection(drop_columns), inplace=True ) btc_base_tf = self.dp.get_pair_dataframe(btc_info_pair, self.timeframe) btc_base_tf = self.base_tf_btc_indicators(btc_base_tf, metadata) dataframe = merge_informative_pair( dataframe, btc_base_tf, self.timeframe, self.timeframe, ffill=True ) drop_columns = [ f"{s}_{self.timeframe}" for s in ["date", "open", "high", "low", "close", "volume"] ] dataframe.drop( columns=dataframe.columns.intersection(drop_columns), inplace=True ) 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) dataframe["hma_50"] = qtpylib.hull_moving_average(dataframe["close"], window=50) dataframe["sma_9"] = ta.SMA(dataframe, timeperiod=9) dataframe["EWO"] = EWO(dataframe, self.fast_ewo, self.slow_ewo) dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["rsi_fast"] = ta.RSI(dataframe, timeperiod=4) dataframe["rsi_slow"] = ta.RSI(dataframe, timeperiod=20) dataframe["ema_14"] = ta.EMA(dataframe, timeperiod=14) dataframe["rsi_4"] = ta.RSI(dataframe, timeperiod=4) dataframe["rsi_14"] = ta.RSI(dataframe, timeperiod=14) dataframe["dema_30"] = ftt.dema(dataframe, period=30) dataframe["dema_200"] = ftt.dema(dataframe, period=200) dataframe["pump_strength"] = ( dataframe["dema_30"] - dataframe["dema_200"] ) / dataframe["dema_30"] stoch_fast = ta.STOCHF(dataframe, 5, 3, 0, 3, 0) dataframe["fastd"] = stoch_fast["fastd"] dataframe["fastk"] = stoch_fast["fastk"] dataframe["adx"] = ta.ADX(dataframe) dataframe["ema_8"] = ta.EMA(dataframe, timeperiod=8) dataframe = self.pump_dump_protection(dataframe, metadata) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [] dataframe.loc[:, "enter_tag"] = "" lambo2 = ( (dataframe["close"] < dataframe["ema_14"] * self.lambo2_ema_14_factor.value) & (dataframe["rsi_4"] < int(self.lambo2_rsi_4_limit.value)) & (dataframe["rsi_14"] < int(self.lambo2_rsi_14_limit.value)) ) dataframe.loc[lambo2, "enter_tag"] += "lambo2_" conditions.append(lambo2) 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) 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) 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) if conditions: dataframe.loc[reduce(lambda x, y: x | y, conditions), "enter_long"] = 1 dont_buy_conditions = [] dont_buy_conditions.append(dataframe["pnd_volume_warn"] < 0.0) dont_buy_conditions.append(dataframe["btc_rsi_8_1h"] < 35.0) 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: conditions = [] conditions.append( (dataframe["close"] > dataframe["hma_50"]) & ( dataframe["close"] > dataframe[f"ma_sell_{self.base_nb_candles_sell.value}"] * self.high_offset_2.value ) & (dataframe["rsi"] > 50) & (dataframe["volume"] > 0) & (dataframe["rsi_fast"] > dataframe["rsi_slow"]) | (dataframe["close"] < dataframe["hma_50"]) & ( dataframe["close"] > dataframe[f"ma_sell_{self.base_nb_candles_sell.value}"] * self.high_offset.value ) & (dataframe["volume"] > 0) & (dataframe["rsi_fast"] > dataframe["rsi_slow"]) ) if conditions: dataframe.loc[reduce(lambda x, y: x | y, conditions), "exit_long"] = 1 return dataframe 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: trade.exit_reason = exit_reason + "_" + trade.enter_tag return True def pct_change(a, b): return (b - a) / a class EI3v2_tag_cofi_dca_green(Github_remiotore_freqtrade__EI3v2_kalidem__20260111_210550): initial_safety_order_trigger = -0.018 max_safety_orders = 8 safety_order_step_scale = 1.2 safety_order_volume_scale = 1.4 buy_params = {"dca_min_rsi": 35} buy_params.update(Github_remiotore_freqtrade__EI3v2_kalidem__20260111_210550.buy_params) dca_min_rsi = IntParameter( 35, 75, default=buy_params["dca_min_rsi"], space="buy", optimize=True ) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe = super().populate_indicators(dataframe, metadata) dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) return dataframe 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_profit > self.initial_safety_order_trigger: return None dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() previous_candle = dataframe.iloc[-2].squeeze() if last_candle["close"] < previous_candle["close"]: return None count_of_buys = 0 for order in trade.orders: if order.ft_is_open or order.ft_order_side != "entry": continue if order.status == "closed": count_of_buys += 1 if 1 <= count_of_buys <= self.max_safety_orders: safety_order_trigger = abs(self.initial_safety_order_trigger) + abs( self.initial_safety_order_trigger ) * self.safety_order_step_scale * ( math.pow(self.safety_order_step_scale, count_of_buys - 1) - 1 ) / ( self.safety_order_step_scale - 1 ) if current_profit <= -1 * abs(safety_order_trigger): try: stake_amount = self.wallets.get_trade_stake_amount(trade.pair, None) stake_amount = stake_amount * math.pow( self.safety_order_volume_scale, count_of_buys - 1 ) amount = stake_amount / current_rate logger.info( f"Initiating safety order buy #{count_of_buys} for {trade.pair} with stake amount of {stake_amount} which equals {amount}" ) return stake_amount except Exception as exception: logger.info( f"Error occured while trying to get stake amount for {trade.pair}: {str(exception)}" ) return None return None