BatsContest

"""
78/100:    245 trades. Avg profit   1.40%. Total profit  0.03034187 BTC ( 342.11Σ%). Avg duration 301.9 min. Objective: -154.45381
"""
import logging
from datetime import datetime, timedelta
from functools import reduce
from typing import Optional, Union

import talib as ta
from diskcache import Index
from freqtrade.persistence import Trade
from freqtrade.strategy import (
    CategoricalParameter,
    DecimalParameter,
    IntParameter,
    stoploss_from_open,
)
from freqtrade.strategy.interface import IStrategy
from lazyft.space_handler import SpaceHandler
from pandas import DataFrame
from technical import qtpylib

# from lft_rest.rest_strategy import BaseRestStrategy

logger = logging.getLogger(__name__)


class BatsContest(IStrategy):
    sh = SpaceHandler(__file__, disable=__name__ != __qualname__)

    # Stoploss:
    stoploss = -1

    # ROI table:
    # minimal_roi = {"160": 0.1}
    minimal_roi = {"0": 1000}

    timeframe = '4h'
    use_custom_stoploss = True

    custom_cooldowns: dict[str, datetime] = {}

    # region Opt Configuration
    load = True
    optimize_rsi = sh.get_space('rsi')
    optimize_atr_roi = sh.get_space('atr_roi')
    optimize_atr_stoploss = sh.get_space('atr_stoploss')
    optimize_high = sh.get_space('high')
    optimize_custom_stoploss = sh.get_space('custom_stoploss')
    optimize_cooldown = sh.get_space('cooldown')
    optimize_risk_reward_ratio = sh.get_space('risk_reward_ratio')
    # endregion

    # region Opt Parameters
    # atr
    atr_roi_multiplier = DecimalParameter(
        1.0, 4.0, default=2, space='buy', optimize=optimize_atr_roi, load=load
    )
    atr_roi_timeperiod = CategoricalParameter(
        list(range(2, 30 + 1, 2)), default=2, space='buy', optimize=optimize_atr_roi, load=load
    )

    atr_stoploss_multiplier = DecimalParameter(
        1.0, 4.0, default=2, space='sell', optimize=optimize_atr_stoploss, load=load
    )
    atr_stoploss_timeperiod = CategoricalParameter(
        list(range(2, 30 + 1, 2)),
        default=2,
        space='sell',
        optimize=optimize_atr_stoploss,
        load=load,
    )
    rsi_value = IntParameter(30, 70, default=50, space='buy', optimize=optimize_rsi, load=load)
    rsi_timeperiod = CategoricalParameter(
        list(range(2, 30 + 1, 2)), default=30, space='buy', optimize=optimize_rsi, load=load
    )

    close_high_periods = CategoricalParameter(
        list(range(30, 150 + 1, 3)), default=48, space='buy', optimize=optimize_high, load=load
    )

    use_custom_stoploss_ = CategoricalParameter(
        [True, False],
        default=sh.get_setting('use_custom_stoploss', False),
        space='sell',
        optimize=optimize_custom_stoploss,
        load=load,
    )
    # close_high_periods_inf = CategoricalParameter(
    #     list(range(30, 80 + 1, 3)), default=48, space='buy', optimize=optimize_high, load=load
    # )
    pHSL = DecimalParameter(
        -0.99,
        -0.040,
        default=-0.99,
        decimals=3,
        space='sell',
        optimize=False,
        load=False,
    )
    # profit threshold 1, trigger point, SL_1 is used
    pPF_1 = DecimalParameter(
        0.01,
        0.9,
        default=0.176,
        decimals=3,
        space='sell',
        optimize=optimize_custom_stoploss,
        load=True,
    )
    pSL_1 = DecimalParameter(
        0.01,
        0.9,
        default=0.176,
        decimals=3,
        space='sell',
        optimize=optimize_custom_stoploss,
        load=True,
    )

    # profit threshold 2, SL_2 is used
    pPF_2 = DecimalParameter(
        0.01,
        0.9,
        default=0.627,
        decimals=3,
        space='sell',
        optimize=optimize_custom_stoploss,
        load=True,
    )
    pSL_2 = DecimalParameter(
        0.01,
        0.9,
        default=0.389,
        decimals=3,
        space='sell',
        optimize=optimize_custom_stoploss,
        load=True,
    )

    loss_cooldown = IntParameter(
        1,
        40,
        default=5,
        space='sell',
        optimize=optimize_cooldown,
        load=False,
    )
    profit_cooldown = IntParameter(
        1,
        40,
        default=20,
        space='sell',
        optimize=optimize_cooldown,
        load=False,
    )

    risk_reward_ratio = DecimalParameter(
        1.0,
        6.0,
        default=3.5,
        decimals=1,
        space='sell',
        optimize=optimize_risk_reward_ratio,
        load=True,
    )
    set_to_break_even_at_profit = DecimalParameter(
        0.5,
        3.0,
        default=1.0,
        decimals=1,
        space='sell',
        optimize=optimize_risk_reward_ratio,
        load=True,
    )

    stoploss_type = CategoricalParameter(
        ['a', 'b'],
        default='a',
        space='sell',
        optimize=optimize_custom_stoploss,
        load=load,
    )

    custom_sell_type = CategoricalParameter(
        ['a', 'b'],
        default='a',
        space='buy',
        optimize=optimize_atr_roi and optimize_atr_stoploss,
        load=load,
    )
    # endregion
    # buy_params = {
    #     "atr_roi_multiplier": 3.487,
    #     "atr_roi_timeperiod": 26,
    #     "close_high_periods": 48,  # value loaded from strategy
    #     "rsi_timeperiod": 30,  # value loaded from strategy
    #     "rsi_value": 50,  # value loaded from strategy
    # }
    #
    # # Sell hyperspace params:
    # sell_params = {
    #     "atr_stoploss_multiplier": 2.837,
    #     "atr_stoploss_timeperiod": 26,
    #     "pHSL": -0.99,  # value loaded from strategy
    #     "pPF_1": 0.176,  # value loaded from strategy
    #     "pPF_2": 0.627,  # value loaded from strategy
    #     "pSL_1": 0.176,  # value loaded from strategy
    #     "pSL_2": 0.389,  # value loaded from strategy
    # }

    # Buy hyperspace params:
    # buy_params = {
    #     "atr_roi_multiplier": 3.557,
    #     "atr_roi_timeperiod": 8,
    #     "close_high_periods": 68,  # value loaded from strategy
    #     "rsi_value": 50,  # value loaded from strategy
    # }

    # Sell hyperspace params:
    # sell_params = {
    #     "pPF_1": 0.176,
    #     "pPF_2": 0.627,
    #     "pSL_1": 0.176,
    #     "pSL_2": 0.389,
    #     "atr_stoploss_multiplier": 1.411,  # value loaded from strategy
    #     "atr_stoploss_timeperiod": 10,  # value loaded from strategy
    # }

    # buy_params = {
    #     "atr_roi_multiplier": 2.065,
    #     "atr_roi_timeperiod": 12,
    #     "close_high_periods": 68,  # value loaded from strategy
    #     "close_high_periods_inf": 48,  # value loaded from strategy
    #     "rsi_timeperiod": 30,  # value loaded from strategy
    #     "rsi_value": 50,  # value loaded from strategy
    # }
    #
    # # Sell hyperspace params:
    # sell_params = {
    #     "atr_stoploss_multiplier": 3.928,
    #     "atr_stoploss_timeperiod": 2,
    # }

    def __init__(self, config: dict) -> None:
        super().__init__(config)
        if self.is_live_or_dry:
            self.persistence = Index(f'{self.__class__.__name__}_persistence')
        else:
            self.persistence = {}

    # def custom_stoploss(
    #     self,
    #     pair: str,
    #     trade: 'Trade',
    #     current_time: datetime,
    #     current_rate: float,
    #     current_profit: float,
    #     **kwargs,
    # ) -> float:
    #     """
    #     custom_stoploss using a risk/reward ratio
    #     """
    #
    #     result = break_even_sl = takeprofit_sl = -1
    #     custom_info_pair = self.persistence.get(pair)
    #
    #     if custom_info_pair is None:
    #         return self.stoploss
    #
    #     # get the atr stoploss at the time of purchase
    #     initial_sl_abs = custom_info_pair['stoploss']
    #
    #     # calculate initial stoploss at open_date
    #     initial_sl = initial_sl_abs / current_rate - 1
    #
    #     # calculate take profit threshold
    #     # by using the initial risk and multiplying it
    #     risk_distance = trade.open_rate - initial_sl_abs
    #
    #     # reward_distance = risk_distance*self.custom_info['risk_reward_ratio']
    #     # replace with hyperoptable parameter
    #     reward_distance = risk_distance * self.risk_reward_ratio.value
    #
    #     # take_profit tries to lock in profit once price gets over
    #     # risk/reward ratio treshold
    #     take_profit_price_abs = trade.open_rate + reward_distance
    #
    #     # take_profit gets triggerd at this profit
    #     take_profit_pct = take_profit_price_abs / trade.open_rate - 1
    #
    #     # break_even tries to set sl at open_rate+fees (0 loss)
    #     # break_even_profit_distance = risk_distance*self.custom_info['set_to_break_even_at_profit']
    #     # replace with hyperoptable parameter
    #     break_even_profit_distance = risk_distance * self.set_to_break_even_at_profit.value
    #
    #     # break_even gets triggerd at this profit
    #     break_even_profit_pct = (break_even_profit_distance + current_rate) / current_rate - 1
    #
    #     result = initial_sl
    #     if current_profit >= break_even_profit_pct:
    #         break_even_sl = (
    #             trade.open_rate * (1 + trade.fee_open + trade.fee_close) / current_rate
    #         ) - 1
    #         result = break_even_sl
    #
    #     if current_profit >= take_profit_pct:
    #         takeprofit_sl = take_profit_price_abs / current_rate - 1
    #         result = takeprofit_sl
    #
    #     return result

    # # Stoploss:
    # stoploss = -0.203
    #
    # Trailing stop:
    # trailing_stop = True
    # trailing_stop_positive = 0.015
    # trailing_stop_positive_offset = 0.115
    # trailing_only_offset_is_reached = True

    # trailing_stop = True
    # trailing_stop_positive = 0.01
    # trailing_stop_positive_offset = 0.066
    # trailing_only_offset_is_reached = True

    # buy_params = {
    #     "atr_roi_multiplier": 3.557,
    #     "atr_roi_timeperiod": 8,
    #     "close_high_periods": 48,  # value loaded from strategy
    #     "rsi_value": 50,  # value loaded from strategy
    # }

    # Sell hyperspace params:
    # sell_params = {
    #     "atr_stoploss_multiplier": 1.411,
    #     "atr_stoploss_timeperiod": 10,
    # }
    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> float:
        if self.stoploss_type.value == 'a':
            return self.custom_stoploss_a(
                pair, trade, current_time, current_rate, current_profit, **kwargs
            )
        elif self.stoploss_type.value == 'b':
            return self.custom_stoploss_b(
                pair, trade, current_time, current_rate, current_profit, **kwargs
            )
        return self.stoploss

    def custom_stoploss_a(
        self,
        pair: str,
        trade: 'Trade',
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> float:
        if not self.use_custom_stoploss_.value:
            return self.stoploss
        # hard stoploss profit
        hard_stop_loss = self.pHSL.value
        profit_level_1 = self.pPF_1.value
        stoploss_level_1 = self.pSL_1.value
        profit_level_2 = self.pPF_2.value
        stoploss_level_2 = self.pSL_2.value

        # For profits between PF_1 and PF_2 the stoploss (sl_profit) used is linearly interpolated
        # between the values of SL_1 and SL_2. For all profits above PL_2 the sl_profit value
        # rises linearly with current profit, for profits below PF_1 the hard stoploss profit is used.

        if current_profit > profit_level_2:
            sl_profit = stoploss_level_2 + (current_profit - profit_level_2)
        elif current_profit > profit_level_1:
            sl_profit = stoploss_level_1 + (
                (current_profit - profit_level_1)
                * (stoploss_level_2 - stoploss_level_1)
                / (profit_level_2 - profit_level_1)
            )
        else:
            sl_profit = hard_stop_loss

        # Only for hyperopt invalid return
        if sl_profit >= current_profit:
            return self.stoploss

        return stoploss_from_open(sl_profit, current_profit)

    def custom_stoploss_b(
        self,
        pair: str,
        trade: 'Trade',
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> float:
        """
        custom_stoploss using a risk/reward ratio
        """
        if not self.use_custom_stoploss_.value:
            return self.stoploss
        result = break_even_sl = takeprofit_sl = -1
        custom_info_pair = self.persistence.get(pair)

        if custom_info_pair is None:
            return self.stoploss

        # get the atr stoploss at the time of purchase
        initial_sl_abs = custom_info_pair['stoploss']

        # calculate initial stoploss at open_date
        initial_sl = initial_sl_abs / current_rate - 1

        # calculate take profit threshold
        # by using the initial risk and multiplying it
        risk_distance = trade.open_rate - initial_sl_abs

        # reward_distance = risk_distance*self.custom_info['risk_reward_ratio']
        # replace with hyperoptable parameter
        reward_distance = risk_distance * self.risk_reward_ratio.value

        # take_profit tries to lock in profit once price gets over
        # risk/reward ratio treshold
        take_profit_price_abs = trade.open_rate + reward_distance

        # take_profit gets triggerd at this profit
        take_profit_pct = take_profit_price_abs / trade.open_rate - 1

        # break_even tries to set sl at open_rate+fees (0 loss)
        # break_even_profit_distance = risk_distance*self.custom_info['set_to_break_even_at_profit']
        # replace with hyperoptable parameter
        break_even_profit_distance = risk_distance * self.set_to_break_even_at_profit.value

        # break_even gets triggerd at this profit
        break_even_profit_pct = (break_even_profit_distance + current_rate) / current_rate - 1

        result = initial_sl
        if current_profit >= break_even_profit_pct:
            break_even_sl = (
                trade.open_rate * (1 + trade.fee_open + trade.fee_close) / current_rate
            ) - 1
            result = break_even_sl

        if current_profit >= take_profit_pct:
            takeprofit_sl = take_profit_price_abs / current_rate - 1
            result = takeprofit_sl

        return result

    @property
    def is_live_or_dry(self):
        return self.config["runmode"].value in ("live", "dry_run")

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # atr
        for i in self.atr_roi_timeperiod.range:
            dataframe[f'roi_atr_{i}'] = ta.ATR(
                dataframe['high'],
                dataframe['low'],
                dataframe['close'],
                timeperiod=i,
            )
        for i in self.atr_stoploss_timeperiod.range:
            dataframe[f'stoploss_atr_{i}'] = ta.ATR(
                dataframe['high'],
                dataframe['low'],
                dataframe['close'],
                timeperiod=i,
            )

        # rsi
        for i in self.rsi_timeperiod.range:
            dataframe[f'rsi_{i}'] = ta.RSI(dataframe['close'], timeperiod=i)
        # self.close_high_periods.value-period close high
        for val in self.close_high_periods.range:
            dataframe[f'close_high_{val}'] = dataframe['close'].rolling(val).max()
        return dataframe

    def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        conditions = []
        # if close > N bar high

        conditions.append(
            dataframe['close'] == dataframe[f'close_high_{self.close_high_periods.value}']
        )
        # if self.enable_inf:
        #     conditions.append(
        #         dataframe['close']
        #         == dataframe[f'close_high_{self.close_high_periods_inf.value}_1d']
        #     )

        # rsi > self.rsi_value.value
        conditions.append(dataframe[f'rsi_{self.rsi_timeperiod.value}'] >= self.rsi_value.value)

        if conditions:
            dataframe.loc[reduce(lambda x, y: x & y, conditions), 'buy'] = 1
        return dataframe

    def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe['sell'] = 0
        return dataframe

    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time: datetime,
        **kwargs,
    ) -> bool:
        # make sure cooldown period has passed
        if pair in self.custom_cooldowns:
            if current_time < self.custom_cooldowns[pair]:
                logger.debug(f'{pair} cooldown period not over yet')
                return False
            else:
                try:
                    logger.debug(f'{pair} cooldown period over')
                    del self.custom_cooldowns[pair]
                except KeyError:
                    pass

        # get atr of last candle
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = dataframe.iloc[-1].squeeze()
        stoploss_atr = rate - (
            self.atr_stoploss_multiplier.value
            * last_candle[f'stoploss_atr_{self.atr_stoploss_timeperiod.value}']
        )
        roi_atr = rate + (
            self.atr_roi_multiplier.value * last_candle[f'roi_atr_{self.atr_roi_timeperiod.value}']
        )
        # roi_atr = last_candle[f'roi_atr_{self.atr_roi_timeperiod.value}']
        # print(f'{pair} stoploss_atr: {stoploss_atr} roi_atr: {roi_atr}')
        self.persistence[pair] = {'stoploss': stoploss_atr, 'roi': roi_atr}
        return True

    def custom_sell_a(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> Optional[Union[str, bool]]:
        """
        Calculate stop based on:
            Fixed dollar value ($1,000)
            Y * average true range from entry
            Z * average true range from entry (profit target)
        :param pair:
        :param trade:
        :param current_time:
        :param current_rate:
        :param current_profit:
        :param kwargs:
        :return:
        """
        # get the candle at the open date
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = dataframe.iloc[-1].squeeze()
        stoploss_atr = last_candle[f'stoploss_atr_{self.atr_stoploss_timeperiod.value}']
        roi_atr = last_candle[f'roi_atr_{self.atr_roi_timeperiod.value}']

        # stoploss_atr = self.persistence[pair]['stoploss_atr']
        # roi_atr = self.persistence[pair]['roi_atr']

        # if price is below atr stoploss from entry (`Trade.open_rate`)
        if current_rate <= (trade.open_rate - (stoploss_atr * self.atr_stoploss_multiplier.value)):
            return 'atr_stoploss'
        # if price is above atr roi from entry
        if current_rate >= (trade.open_rate + (roi_atr * self.atr_roi_multiplier.value)):
            return 'atr_roi'
        return None

    def custom_sell_b(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> Optional[Union[str, bool]]:
        """
        Calculate stop based on:
            Fixed dollar value ($1,000)
            Y * average true range from entry
            Z * average true range from entry (profit target)
        :param pair:
        :param trade:
        :param current_time:
        :param current_rate:
        :param current_profit:
        :param kwargs:
        :return:
        """
        # get the candle at the open date
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = dataframe.iloc[-1].squeeze()
        # stoploss_atr = last_candle[f'stoploss_atr_{self.atr_stoploss_timeperiod.value}']
        roi_atr = last_candle[f'roi_atr_{self.atr_roi_timeperiod.value}']

        stoploss_atr = self.persistence[pair]['stoploss']
        roi = self.persistence[pair]['roi']

        # atr trailing stoploss
        peak_profit = self.persistence[pair].get('peak_profit', 0)
        if current_profit > peak_profit:
            logger.info(f'{pair} new peak profit: {current_profit}')
            new_stoploss = current_rate - (
                self.atr_stoploss_multiplier.value
                * last_candle[f'stoploss_atr_{self.atr_stoploss_timeperiod.value}']
            )
            self.persistence[pair]['stoploss'] = new_stoploss
            self.persistence[pair]['peak_profit'] = current_profit
        elif not peak_profit and current_profit > 0:
            logger.info(f'{pair} peak profit set: {current_profit}')
            self.persistence[pair]['peak_profit'] = current_profit

        logger.info(
            f'{pair} stoploss_atr: {stoploss_atr} roi_atr: {roi} peak_profit: '
            f'{peak_profit} current_rate: {current_rate} current_profit: {current_profit}'
        )
        if current_rate <= stoploss_atr:
            logger.info(f'{pair} stoploss triggered')
            return 'atr_stoploss'

        # roi_atr = self.persistence[pair]['roi_atr']

        # if price is below atr stoploss from entry (`Trade.open_rate`)
        # if current_rate <= (trade.open_rate - (stoploss_atr * self.atr_stoploss_multiplier.value)):
        #     return 'atr_stoploss'
        # if price is above atr roi from entry
        if current_rate >= roi:
            logger.info(f'{pair} roi triggered')
            return 'atr_roi'

        return None

    def custom_sell(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> Optional[Union[str, bool]]:
        if self.custom_sell_type.value == 'a':
            return self.custom_sell_a(
                pair, trade, current_time, current_rate, current_profit, **kwargs
            )
        elif self.custom_sell_type.value == 'b':
            return self.custom_sell_b(
                pair, trade, current_time, current_rate, current_profit, **kwargs
            )

    def confirm_trade_exit(
        self,
        pair: str,
        trade: Trade,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        sell_reason: str,
        current_time: datetime,
        **kwargs,
    ) -> bool:
        try:
            del self.persistence[pair]
        except KeyError:
            pass
        return super().confirm_trade_exit(
            pair,
            trade,
            order_type,
            amount,
            rate,
            time_in_force,
            sell_reason,
            current_time,
            **kwargs,
        )

    def custom_exit_price(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        proposed_rate: float,
        current_profit: float,
        **kwargs,
    ) -> float:
        # If last trade was a loser, wait 5 bars before entering next trade (minimizes
        # whipsaws).
        # If last trade was a winner, wait 20 bars before entering next trade (be patient after
        # wins)
        if current_profit < 0:
            until = current_time + timedelta(
                minutes=self.timeframe_to_minutes * int(self.loss_cooldown.value)
            )
            if not self.is_live_or_dry:
                self.custom_cooldowns[pair] = until
            else:
                self.lock_pair(pair, until=until, reason='Losing trade')
        else:
            until = current_time + timedelta(
                minutes=self.timeframe_to_minutes * int(self.profit_cooldown.value)
            )
            if not self.is_live_or_dry:
                self.custom_cooldowns[pair] = until
            else:
                self.lock_pair(pair, until=until, reason='Winning trade')
        return proposed_rate

    @property
    def timeframe_to_minutes(self) -> int:
        """
        Get the timeframe in minutes.
        1d = 1440
        4h = 240
        :return: int
        """
        import re

        timeframe = self.timeframe
        # timeframe can be: 1m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 8h, 12h, 1d
        # hours
        if re.search(r'h', timeframe):
            return int(re.sub(r'h', '', timeframe)) * 60
        # days
        if re.search(r'd', timeframe):
            return int(re.sub(r'd', '', timeframe)) * 1440
        # minute
        return int(re.sub(r'm', '', timeframe))


# class BatsContestRest(BaseRestStrategy, BatsContest):
#     backtest_days = 180
#     hyperopt_days = 180
#     hyperopt_epochs = 40
#     min_avg_profit = 0.01
#     min_backtest_trades = 2
#     min_hyperopt_trades = 2
#     min_win_ratio = 0.40
#     request_hyperopt = False
#     timeframe_detail = '5m'