ERMD_Strategy.py

# --- Do not remove these libs ---
from datetime import datetime
from functools import reduce
from typing import Optional, Union
from freqtrade.persistence import Trade
from freqtrade.strategy import BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, IStrategy, stoploss_from_open
import pandas as pd
from pandas import DataFrame
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
import warnings

warnings.simplefilter(action='ignore', category=pd.errors.PerformanceWarning)

DEBUG = False

class ERMD_Strategy(IStrategy):
    INTERFACE_VERSION: int = 3

    can_short = True  # enable short direction
    timeframe = "1d"
    startup_candle_count = 100  # how many candles we need to skip to start receiving robust signals
    risk = 0.01
    min_pos_pct = 0.01      # minimum position size, percent of current portfolio value
    max_pos_pct = DecimalParameter(0.01, 0.5, default=0.48, decimals=2, optimize=True, space="buy")         # maximum position size, percent of current portfolio value

    # stop loss
    stoploss = -0.10                # maximum stop loss distance
    tsl_break_even = DecimalParameter(0.0, 0.1, default=0.04, decimals=2, optimize=True, space="buy")       # place break-even stop loss when desired profit is reached
    tsl_start_offset = DecimalParameter(0.0, 0.1, default=0.071, decimals=3, optimize=True, space="buy")    # start trailing stop loss when this profit level is reached
    tsl_trailing_distance = DecimalParameter(0.01, 0.2, default=0.09, decimals=2, optimize=True, space="buy")      # distance to price

    # take profit
    tp1_dist = 0.2
    tp1_size = 0.3
    tp2_dist = 0.4
    tp2_size = 0.3
    tp3_dist = 0.6
    tp3_size = 0.4
    minimal_roi = {"0": 0.8380000000000001}       # optimized tp

    # Parameters: signal
    buy_ema_enabled = BooleanParameter(default=False, optimize=True)
    buy_ema1_length = IntParameter(3, 100, default=5, optimize=True)
    buy_ema2_length = IntParameter(3, 100, default=10, optimize=True)
    buy_ema3_length = IntParameter(3, 100, default=20, optimize=True)
    buy_rsi_enabled = BooleanParameter(default=False, optimize=True)
    buy_rsi_length = IntParameter(3, 100, default=14, optimize=True)
    buy_rsi_threshold = IntParameter(50, 100, default=60, optimize=True)
    buy_macd_enabled = BooleanParameter(default=True, optimize=True)
    buy_macd_fast = IntParameter(3, 50, default=12, optimize=True)
    buy_macd_slow = IntParameter(20, 200, default=26, optimize=True)
    buy_macd_smooth = IntParameter(1, 30, default=9, optimize=True)
    buy_candlestick_enabled = BooleanParameter(default=False, optimize=True)

    sell_ema_enabled = BooleanParameter(default=True, optimize=True)
    sell_ema1_length = IntParameter(3, 100, default=5, optimize=True)
    sell_ema2_length = IntParameter(3, 100, default=10, optimize=True)
    sell_ema3_length = IntParameter(3, 100, default=20, optimize=True)
    sell_rsi_enabled = BooleanParameter(default=True, optimize=True)
    sell_rsi_length = IntParameter(3, 100, default=14, optimize=True)
    sell_rsi_threshold = IntParameter(0, 50, default=40, optimize=True)
    sell_macd_enabled = BooleanParameter(default=True, optimize=True)
    sell_macd_fast = IntParameter(3, 50, default=12, optimize=True)
    sell_macd_slow = IntParameter(20, 200, default=26, optimize=True)
    sell_macd_smooth = IntParameter(1, 30, default=9, optimize=True)
    sell_candlestick_enabled = BooleanParameter(default=False, optimize=True)

    # signals
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """ adds indicators to dataframe """

        # BUY
        # EMAs
        for val in self.buy_ema1_length.range:
            dataframe[f"buy_ema1_{val}"] = ta.EMA(dataframe, val)
        for val in self.buy_ema2_length.range:
            dataframe[f"buy_ema2_{val}"] = ta.EMA(dataframe, val)
        for val in self.buy_ema3_length.range:
            dataframe[f"buy_ema3_{val}"] = ta.EMA(dataframe, val)

        # RSI
        for val in self.buy_rsi_length.range:
            dataframe[f"buy_rsi_{val}"] = ta.RSI(dataframe, val)

        # MACD
        for fast_val in self.buy_macd_fast.range:
            for slow_val in self.buy_macd_slow.range:
                for smooth_val in self.buy_macd_smooth.range:
                    macd = ta.MACD(dataframe, fast_val, slow_val, smooth_val)
                    dataframe[f'buy_macd_{fast_val}_{slow_val}_{smooth_val}'] = macd['macd']
                    dataframe[f'buy_macds_{fast_val}_{slow_val}_{smooth_val}'] = macd['macdsignal']

        # SELL
        # EMAs
        for val in self.sell_ema1_length.range:
            dataframe[f"sell_ema1_{val}"] = ta.EMA(dataframe, val)
        for val in self.sell_ema2_length.range:
            dataframe[f"sell_ema2_{val}"] = ta.EMA(dataframe, val)
        for val in self.sell_ema3_length.range:
            dataframe[f"sell_ema3_{val}"] = ta.EMA(dataframe, val)

        # RSI
        for val in self.sell_rsi_length.range:
            dataframe[f"sell_rsi_{val}"] = ta.RSI(dataframe, val)

        # MACD
        for fast_val in self.sell_macd_fast.range:
            for slow_val in self.sell_macd_slow.range:
                for smooth_val in self.sell_macd_smooth.range:
                    macd = ta.MACD(dataframe, fast_val, slow_val, smooth_val)
                    dataframe[f'sell_macd_{fast_val}_{slow_val}_{smooth_val}'] = macd['macd']
                    dataframe[f'sell_macds_{fast_val}_{slow_val}_{smooth_val}'] = macd['macdsignal']

        # DOJI candlestick pattern
        doji = ta.CDLDOJI(dataframe)
        doji_dir = (dataframe["high"] - dataframe["close"]) / (dataframe["close"] - dataframe["low"])
        dataframe.loc[((doji == 100) & (doji_dir > 1), "doji")] = dataframe["high"]
        dataframe.loc[((doji == 100) & (doji_dir < 1), "doji")] = dataframe["low"]

        # MARUBOZU candlestick pattern
        body = (dataframe["close"] - dataframe["open"]).abs()
        wick_top = dataframe["high"] - dataframe[["open", "close"]].max(axis=1)
        wick_bottom = dataframe[["open", "close"]].min(axis=1) - dataframe["low"]
        marubozu = (wick_top/body < 0.25) & (wick_bottom/body < 0.25)
        dataframe.loc[(marubozu & (body > 0), 'marubozu')] = dataframe["high"]
        dataframe.loc[(marubozu & (body < 0), 'marubozu')] = dataframe["low"]

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """ generate signals """

        # BUY signal
        # only trade if there is volume
        conditions = [(dataframe['volume'] > 0)]

        # Close > EMA1 > EMA2 > EMA3
        if self.buy_ema_enabled.value:
            conditions.append((dataframe["close"] > dataframe[f"buy_ema1_{self.buy_ema1_length.value}"]) &
                              (dataframe[f"buy_ema1_{self.buy_ema1_length.value}"] > dataframe[f"buy_ema2_{self.buy_ema2_length.value}"]) &
                              (dataframe[f"buy_ema2_{self.buy_ema2_length.value}"] > dataframe[f"buy_ema3_{self.buy_ema3_length.value}"]))

        # RSI > RSI_Threshold_Long
        if self.buy_rsi_enabled:
            conditions.append(dataframe[f"buy_rsi_{self.buy_rsi_length.value}"] > self.buy_rsi_threshold.value)

        # MACD cross MACDs above or MACD > 0
        if self.buy_macd_enabled.value:
            macd = dataframe[f'buy_macd_{self.buy_macd_fast.value}_{self.buy_macd_slow.value}_{self.buy_macd_smooth.value}']
            macds = dataframe[f'buy_macds_{self.buy_macd_fast.value}_{self.buy_macd_slow.value}_{self.buy_macd_smooth.value}']
            conditions.append(qtpylib.crossed_above(macd, macds) | (macd > 0))

        # Candlestick patterns: any DOJI and green MARUBOZU
        if self.buy_candlestick_enabled.value:
            doji = dataframe['doji'] == dataframe["high"]
            marubozu = dataframe['marubozu'] == dataframe["high"]
            conditions.append(doji | marubozu)

        # combine conditions and return
        buy_conditions = reduce(lambda x, y: x & y, conditions)
        dataframe.loc[buy_conditions, 'enter_long'] = 1

        # SELL signal
        # only trade if there is volume
        conditions = [(dataframe['volume'] > 0)]

        # Close < EMA1 < EMA2 < EMA3
        if self.sell_ema_enabled.value:
            conditions.append((dataframe["close"] < dataframe[f"sell_ema1_{self.sell_ema1_length.value}"]) &
                              (dataframe[f"sell_ema1_{self.sell_ema1_length.value}"] < dataframe[f"sell_ema2_{self.sell_ema2_length.value}"]) &
                              (dataframe[f"sell_ema2_{self.sell_ema2_length.value}"] < dataframe[f"sell_ema3_{self.sell_ema3_length.value}"]))

        # RSI < RSI_Threshold_Long
        if self.sell_rsi_enabled:
            conditions.append(dataframe[f"sell_rsi_{self.sell_rsi_length.value}"] < self.sell_rsi_threshold.value)

        # MACD cross MACDs below or MACD < 0
        if self.sell_macd_enabled.value:
            macd = dataframe[f'sell_macd_{self.sell_macd_fast.value}_{self.sell_macd_slow.value}_{self.sell_macd_smooth.value}']
            macds = dataframe[f'sell_macds_{self.sell_macd_fast.value}_{self.sell_macd_slow.value}_{self.sell_macd_smooth.value}']
            conditions.append(qtpylib.crossed_below(macd, macds) | (macd < 0))

        # Candlestick patterns, any DOJI and red MARUBOZU
        if self.sell_candlestick_enabled.value:
            doji = dataframe['doji'] == dataframe['low']
            marubozu = dataframe['marubozu'] == dataframe['low']
            conditions.append(doji | marubozu)

        # combine conditions and return
        sell_conditions = reduce(lambda x, y: x & y, conditions)
        dataframe.loc[sell_conditions, 'enter_short'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        return dataframe

    stop_info = {}

    def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float,
                    current_profit: float, **kwargs) -> Optional[Union[str, bool]]:
        """ exit on EMA3 cross, or exit at entry candle low (long) or high (short) """

        dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe)
        last_candle = dataframe.iloc[-1].squeeze()
        close = current_rate

        # exit on EMA20 cross
        if trade.is_short:
            ema3 = last_candle[f"sell_ema3_{self.sell_ema3_length.value}"]
            if close > ema3:
                if DEBUG:
                    print(f"{pair} EMA20 exit short at {current_time}, price={current_rate}, level={ema3}, profit={current_profit}")
                return "ema20_cross_exit"
        else:
            ema3 = last_candle[f"buy_ema3_{self.buy_ema3_length.value}"]
            if close < ema3:
                if DEBUG:
                    print(f"{pair} EMA20 exit long at {current_time}, price={current_rate}, level={ema3}, profit={current_profit}")
                return "ema20_cross_exit"

        # exit on entry candle low/high
        if trade.pair in self.stop_info:
            if trade.is_short:
                if close > self.stop_info[trade.pair]["level"]:
                    if DEBUG:
                        print(f"{pair} entry candle high break exit short at {current_time}, price={current_rate}, "
                              f"level={self.stop_info[trade.pair]['level']}, profit={current_profit}")
                    return "candle-based stop_loss"
            else:
                if close < self.stop_info[trade.pair]['level']:
                    if DEBUG:
                        print(f"{pair} entry candle low break exit long at {current_time}, price={current_rate}, "
                              f"level={self.stop_info[trade.pair]['level']}, profit={current_profit}")
                    return "candle-based stop_loss"

        # exit on 3rd take profit, if it sums to 100%
        if self.tp1_size + self.tp2_size + self.tp3_size == 1:
            if trade.is_short:
                ema1 = last_candle[f"sell_ema1_{self.sell_ema1_length.value}"]
                if close < ema1 * (1-self.tp3_dist):
                    if DEBUG:
                        print(f"{pair} 3rd take profit filled => exit short at {current_time}, price={current_rate}, "
                              f"level={ema1 * (1-self.tp3_dist)}, profit={current_profit}")
                    return "take profit 3"
            else:
                ema1 = last_candle[f"buy_ema1_{self.buy_ema1_length.value}"]
                if close > ema1 * (1+self.tp3_dist):
                    if DEBUG:
                        print(f"{pair} 3rd take profit filled => exit long at {current_time}, price={current_rate}, "
                              f"level={ema1 * (1+self.tp3_dist)}, profit={current_profit}")
                    return "take profit 3"

        return False

    # Stop Loss
    use_custom_stoploss = True

    def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
                        current_profit: float, **kwargs) -> float:
        """ custom trailing stop loss """

        if self.tsl_break_even.value <= current_profit < self.tsl_start_offset.value:
            return stoploss_from_open(0, current_profit, is_short=trade.is_short)
        elif current_profit >= self.tsl_start_offset.value:
            return self.tsl_trailing_distance.value
        else:
            return -1

    # Position Size
    def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
                            proposed_stake: float, min_stake: Optional[float], max_stake: float,
                            leverage: float, entry_tag: Optional[str], side: str,
                            **kwargs) -> float:
        """ custom position size """

        if self.wallets:
            total = self.wallets.get_total("USDT")
        else:
            if DEBUG:
                print("Wallets are not available, using max_stake instead of portfolio value to calculate position size!")
            total = max_stake

        dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe)
        last_candle = dataframe.iloc[-1].squeeze()

        if side == "long":
            level = last_candle["low"]
            stop_distance = current_rate / level - 1
            self.stop_info[pair] = {'level': level}
        else:
            level = last_candle["high"]
            stop_distance = level / current_rate - 1
            self.stop_info[pair] = {'level': level}

        if stop_distance > abs(self.stoploss):
            if DEBUG:
                print(f"{pair} at {current_time}: stop distance {round(stop_distance*100)}% is too large, "
                      f"reducing stop loss to {abs(round(self.stoploss*100))}!")
            stop_distance = abs(self.stoploss)

        result = total * self.risk / stop_distance

        if DEBUG:
            print(f"{pair} at {current_time} position calculation: level={level}; current price={current_rate}; "
                  f"distance={stop_distance}; result={result}; min_stake={min_stake}; max_stake={max_stake}")

        # apply maximum/minimum position size
        min_pos = self.min_pos_pct * total
        max_pos = self.max_pos_pct.value * total
        if result < min_pos:
            if DEBUG:
                print(f"Calculated position ({round(result, 2)}) is less than minimum position size ({round(min_pos)})! Skipping...")
            return 0
        if result > max_pos:
            if DEBUG:
                print(f"Calculated position ({round(result, 2)}) is greater than maximum position size ({round(min_pos)})! Reduced!")
            result = max_pos

        # check min/max stake
        if result < min_stake:
            if DEBUG:
                print(f"Calculated stake ({round(result,2)}) is less than minimum stake ({min_stake})! Skipping trade...")
            return 0
        elif result > max_stake:
            if DEBUG:
                print(f"Calculated stake ({round(result,2)}) is greater than maximum stake ({max_stake})! Reducing to maximum!")
            result = max_stake

        return result

    # Take Profit
    position_adjustment_enable = True
    max_entry_position_adjustment = -1

    def adjust_trade_position(self, trade: Trade, current_time: datetime,
                              current_rate: float, current_profit: float,
                              min_stake: Optional[float], max_stake: float,
                              current_entry_rate: float, current_exit_rate: float,
                              current_entry_profit: float, current_exit_profit: float,
                              **kwargs) -> Optional[float]:
        """ custom take profits """

        dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)

        if trade.is_short:
            if trade.nr_of_successful_exits == 0:
                if current_rate < (1-self.tp1_dist) * dataframe[f"sell_ema1_{self.sell_ema1_length.value}"].iloc[-1].squeeze():
                    return -trade.amount_requested * self.tp1_size * current_rate
            if trade.nr_of_successful_exits == 1:
                if current_rate < (1-self.tp2_dist) * dataframe[f"sell_ema1_{self.sell_ema1_length.value}"].iloc[-1].squeeze():
                    return -trade.amount_requested * self.tp2_size * current_rate
            if trade.nr_of_successful_exits == 2:
                if current_rate < (1-self.tp3_dist) * dataframe[f"sell_ema1_{self.sell_ema1_length.value}"].iloc[-1].squeeze():
                    if self.tp1_size + self.tp2_size + self.tp3_size != 1:
                        return -trade.amount_requested * self.tp3_size * current_rate
        else:
            if trade.nr_of_successful_exits == 0:
                if current_rate > (1 + self.tp1_dist) * dataframe[f"buy_ema1_{self.buy_ema1_length.value}"].iloc[-1].squeeze():
                    return -trade.amount_requested * self.tp1_size * current_rate
            if trade.nr_of_successful_exits == 1:
                if current_rate > (1 + self.tp2_dist) * dataframe[f"buy_ema1_{self.buy_ema1_length.value}"].iloc[-1].squeeze():
                    return -trade.amount_requested * self.tp2_size * current_rate
            if trade.nr_of_successful_exits == 2:
                if current_rate > (1 + self.tp3_dist) * dataframe[f"buy_ema1_{self.buy_ema1_length.value}"].iloc[-1].squeeze():
                    if self.tp1_size + self.tp2_size + self.tp3_size != 1:
                        return -trade.amount_requested * self.tp3_size * current_rate

        return None

    # set up plotting
    @property
    def plot_config(self):
        plot_config = {
            'main_plot': {
                f"buy_ema1_{self.buy_ema1_length.value}": {},
                f"buy_ema2_{self.buy_ema2_length.value}": {},
                f"buy_ema3_{self.buy_ema3_length.value}": {},
                f"doji": {"type": "scatter"},
                f"marubozu": {"type": "scatter"}
            },
            'subplots': {
                "MACD": {
                    f'buy_macd_{self.buy_macd_fast.value}_{self.buy_macd_slow.value}_{self.buy_macd_smooth.value}': {},
                    f'buy_macds_{self.buy_macd_fast.value}_{self.buy_macd_slow.value}_{self.buy_macd_smooth.value}': {},
                },
                "RSI": {
                    f'buy_rsi_{self.buy_rsi_length.value}': {}
                }
            }}

        return plot_config