ScalpingStrategy

from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, merge_informative_pair
from pandas import DataFrame
import talib.abstract as ta


class ScalpingStrategy(IStrategy):

    INTERFACE_VERSION = 3
    timeframe = "3m"
    can_short = True

    minimal_roi = {
        "0":   0.015,   # 1.5%
        "10":  0.01,    # 1% after 10 min
        "20":  0.007,   # 0.7% after 20 min
        "40":  0.004    # 0.4% after 40 min
    }

    stoploss = -0.008
    trailing_stop = True
    trailing_stop_positive = 0.005
    trailing_stop_positive_offset = 0.008
    trailing_only_offset_is_reached = True

    # Buy parameters
    ema_fast = IntParameter(3, 15, default=8, space="buy")
    ema_slow = IntParameter(10, 30, default=21, space="buy")
    rsi_long_min = IntParameter(20, 45, default=30, space="buy")
    rsi_long_max = IntParameter(50, 75, default=65, space="buy")

    # Sell parameters
    ema_fast_short = IntParameter(3, 15, default=8, space="sell")
    ema_slow_short = IntParameter(10, 30, default=21, space="sell")
    rsi_short_min = IntParameter(45, 65, default=50, space="sell")
    rsi_short_max = IntParameter(65, 90, default=80, space="sell")

    startup_candle_count: int = 50

    def informative_pairs(self):
        pairs = self.dp.current_whitelist()
        return [(pair, "15m") for pair in pairs]

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

        # EMA
        dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=self.ema_fast.value)
        dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=self.ema_slow.value)
        dataframe["ema_fast_short"] = ta.EMA(dataframe, timeperiod=self.ema_fast_short.value)
        dataframe["ema_slow_short"] = ta.EMA(dataframe, timeperiod=self.ema_slow_short.value)

        # RSI
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)

        # Bollinger Bands
        bollinger = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
        dataframe["bb_upper"] = bollinger["upperband"]
        dataframe["bb_lower"] = bollinger["lowerband"]
        dataframe["bb_mid"]   = bollinger["middleband"]

        # Stochastic
        stoch = ta.STOCH(dataframe, fastk_period=14, slowk_period=3, slowd_period=3)
        dataframe["stoch_k"] = stoch["slowk"]
        dataframe["stoch_d"] = stoch["slowd"]

        # Volume MA
        dataframe["volume_ma"] = ta.SMA(dataframe["volume"], timeperiod=20)

        # 15m trend filter
        informative_15m = self.dp.get_pair_dataframe(
            pair=metadata["pair"], timeframe="15m"
        )
        informative_15m["ema_15m_fast"] = ta.EMA(informative_15m, timeperiod=9)
        informative_15m["ema_15m_slow"] = ta.EMA(informative_15m, timeperiod=21)
        informative_15m["trend_up"] = (
            informative_15m["ema_15m_fast"] > informative_15m["ema_15m_slow"]
        ).astype(int)

        dataframe = merge_informative_pair(
            dataframe, informative_15m, self.timeframe, "15m", ffill=True
        )

        return dataframe

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

        # LONG - relaxed conditions
        dataframe.loc[
            (
                (dataframe["trend_up_15m"] == 1) &
                (dataframe["ema_fast"] > dataframe["ema_slow"]) &
                (dataframe["rsi"] >= self.rsi_long_min.value) &
                (dataframe["rsi"] <= self.rsi_long_max.value) &
                (dataframe["stoch_k"] < 40) &
                (dataframe["close"] < dataframe["bb_mid"]) &
                (dataframe["volume"] > 0)
            ),
            ["enter_long", "enter_tag"]
        ] = 1, "scalp_long"

        # SHORT - relaxed conditions
        dataframe.loc[
            (
                (dataframe["trend_up_15m"] == 0) &
                (dataframe["ema_fast_short"] < dataframe["ema_slow_short"]) &
                (dataframe["rsi"] >= self.rsi_short_min.value) &
                (dataframe["rsi"] <= self.rsi_short_max.value) &
                (dataframe["stoch_k"] > 60) &
                (dataframe["close"] > dataframe["bb_mid"]) &
                (dataframe["volume"] > 0)
            ),
            ["enter_short", "enter_tag"]
        ] = 1, "scalp_short"

        return dataframe

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

        # Exit long at BB middle
        dataframe.loc[
            (dataframe["close"] >= dataframe["bb_mid"]) &
            (dataframe["rsi"] > 55),
            ["exit_long", "exit_tag"]
        ] = 1, "bb_mid_exit_long"

        # Exit short at BB middle
        dataframe.loc[
            (dataframe["close"] <= dataframe["bb_mid"]) &
            (dataframe["rsi"] < 45),
            ["exit_short", "exit_tag"]
        ] = 1, "bb_mid_exit_short"

        return dataframe