Heikin_Ashi1D174_spot

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these libs ---
from functools import reduce

import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime
from typing import Optional, Union, Dict

from freqtrade.persistence import Trade
from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter,
                                IntParameter, IStrategy, merge_informative_pair, stoploss_from_absolute)

# --------------------------------
# Add your lib to import here
import talib.abstract as ta
from technical import qtpylib
from technical.candles import heikinashi

Debug = False


class Heikin_Ashi1D174_spot(IStrategy):
    INTERFACE_VERSION = 3

    # class HyperOpt:
    #     def stoploss_space():
    #         return [SKDecimal(-0.9, -0.5, decimals=2, name='stoploss')]
    #
    #     # Define custom ROI space
    #     def roi_space():
    #         return [
    #             Integer(0, 1, name='roi_t'),
    #             Integer(100, 101, name='roi_p'),
    #         ]
    #
    #     def generate_roi_table(params: Dict) -> Dict[int, float]:
    #         roi_table = {}
    #         roi_table[0] = params['roi_p']
    #
    #         return roi_table

    timeframe = '1d'

    # Can this strategy go short?
    can_short: bool = False

    minimal_roi = {
        "0": 0.249,
        "9457": 0.133,
        "23285": 0.068,
        "40133": 0
    }

    stoploss = -0.342

    trailing_stop = True

    # trailing_only_offset_is_reached = False
    # trailing_stop_positive = 0.01
    # trailing_stop_positive_offset = 0.0  # Disabled / not configured

    process_only_new_candles = True

    # These values can be overridden in the config.
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    startup_candle_count: int = 300

    # Optional order type mapping.
    order_types = {
        'entry': 'market',
        'exit': 'market',
        'stoploss': 'market',
        'stoploss_on_exchange': False
    }

    # Optional order time in force.
    order_time_in_force = {
        'entry': 'GTC',
        'exit': 'GTC'
    }

    @property
    def plot_config(self):
        return {
            # Main plot indicators (Moving averages, ...)
            'main_plot': {
                'tema': {},
                'sar': {'color': 'white'},
                f'sl_l{self.buy_sl_mult.value}': {'color': 'red'},
                f'sl_h{self.buy_sl_mult.value}': {'color': 'green'}
            },
            'subplots': {
                # Subplots - each dict defines one additional plot
                "MACD": {
                    'macd': {'color': 'blue'},
                    'macdsignal': {'color': 'orange'},
                },
                "RSI": {
                    'rsi': {'color': 'red'},
                },
                "Willy": {
                    f'willy{self.buy_willy.value}': {'color': 'red'},
                    f'ema_will{self.buy_ema.value}': {'color': 'blue'},
                },
                "sell_willy": {
                    f'sell_willy{self.sell_willy.value}': {'color': 'blue'},
                    f'sell_ema_will{self.sell_ema.value}': {'color': 'blue'}
                },
                "Doji_in_row": {
                    f"doji_in_row{self.sell_doji_in_row.value}": {'color': 'blue', 'type': 'bar'}
                },
            }
        }

    buy_RSI_long = IntParameter(30, 60, default=58, space="buy", optimize=True)
    buy_RSI_short = IntParameter(30, 60, default=55, space="buy", optimize=True)
    buy_bear_row = IntParameter(1, 3, default=1, space='buy', optimize=True)
    buy_bull_row = IntParameter(1, 3, default=1, space='buy', optimize=True)
    buy_ema = IntParameter(7, 22, default=18, space='buy', optimize=True)
    buy_high_line = IntParameter(-30, 0, default=-2, space="buy", optimize=True)
    buy_low_line = IntParameter(-100, -70, default=-89, space="buy", optimize=True)
    buy_on_willy = BooleanParameter(default=False, space='buy', optimize=True)
    buy_sl_mult = DecimalParameter(0.5, 5, default=3.4, decimals=1, space='buy', optimize=True)
    buy_willy = IntParameter(15, 30, default=23, space='buy', optimize=True)
    sell_doji_in_row = IntParameter(1, 6, default=2, space="sell", optimize=True)
    sell_ema = IntParameter(7, 22, default=12, space='sell', optimize=True)
    sell_high_line = IntParameter(-25, 0, default=-10, space="sell", optimize=True)
    sell_low_line = IntParameter(-100, -75, default=-92, space="sell", optimize=True)
    sell_willy = IntParameter(15, 30, default=22, space='sell', optimize=True)
    sell_willy_exit = BooleanParameter(default=True, space='sell', optimize=True)



    def informative_pairs(self):
        pairs = self.dp.current_whitelist()
        informative_pairs = [(pair, '1w') for pair in pairs]
        informative_pairs += [("ETH/USDT", "1w"), ("BTC/USDT", "1w"), ("DOGE/USDT", "1w"), ("BNB/USDT", "1w"),
                              ("XRP/USDT", "1w"),
                              ("ADA/USDT", "1w"), ("MATIC/USDT", "1w"), ("DOT/USDT", "1w"),
                              ("TRX/USDT", "1w"), ("LTC/USDT", "1w"),
                              ("SOL/USDT", "1w"), ("UNI/USDT", "1w"), ("AVAX/USDT", "1w"), ("LPT/USDT", "1w"),
                              ("LINK/USDT", "1w"), ("XMR/USDT", "1w"), ("ATOM/USDT", "1w"),
                              ("ETC/USDT", "1w"), ("BCH/USDT", "1w"), ("XLM/USDT", "1w"), ("APE/USDT", "1w"),
                              ("QNT/USDT", "1w"),
                              ("ALGO/USDT", "1w"), ("VET/USDT", "1w"), ("NEAR/USDT", "1w"),
                              ("HBAR/USDT", "1w"), ("ICP/USDT", "1w"), ("FIL/USDT", "1w"), ("EOS/USDT", "1w"),
                              ("EGLD/USDT", "1w"),
                              ("FLOW/USDT", "1w"), ("THETA/USDT", "1w"), ("AAVE/USDT", "1w"), ("XTZ/USDT", "1w"),
                              ("AXS/USDT", "1w"), ("CHZ/USDT", "1w"), ("SAND/USDT", "1w"), ("ZEC/USDT", "1w"),
                              ("MANA/USDT", "1w"),
                              ("FTM/USDT", "1w"),
                              ("MKR/USDT", "1w"),
                              ("GRT/USDT", "1w"),
                              ("KLAY/USDT", "1w"), ("DASH/USDT", "1w"), ("APT/USDT", "1w"),
                              ("IOTA/USDT", "1w"),
                              ("RUNE/USDT", "1w"), ("NEO/USDT", "1w"), ("SNX/USDT", "1w"),
                              ("IMX/USDT", "1w"),
                              ("FTT/USDT", "1w"), ("1INCH/USDT", "1w"),
                              ("LDO/USDT", "1w"), ("ZIL/USDT", "1w"), ("BAT/USDT", "1w"),
                              ("CRV/USDT", "1w"), ("LRC/USDT", "1w"),
                              ("XEM/USDT", "1w"), ("ENJ/USDT", "1w"),
                              ("HNT/USDT", "1w"), ("CVX/USDT", "1w"), ("HOT/USDT", "1w"), ("BAL/USDT", "1w"),
                              ("KAVA/USDT", "1w"), ("RVN/USDT", "1w"),
                              ("COMP/USDT", "1w"), ("CELO/USDT", "1w"), ("ENS/USDT", "1w"),
                              ("OP/USDT", "1w"),
                              ("KSM/USDT", "1w"), ("AR/USDT", "1w"), ("SUSHI/USDT", "1w"),
                              ("QTUM/USDT", "1w"),
                              ("YFI/USDT", "1w"), ("ROSE/USDT", "1w"), ("ONE/USDT", "1w"),
                              ("BNX/USDT", "1w"),
                              ("IOTX/USDT", "1w"), ("WAVES/USDT", "1w"),
                              ("GMT/USDT", "1w"), ("ANKR/USDT", "1w"),
                              ("MASK/USDT", "1w")
                              ]
        return informative_pairs

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

        inf_tf = '1w'
        informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=inf_tf)
        informative['bear_cand'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, 1)['Bear_cand']
        informative['bull_cand'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, 1)['Bull_cand']
        dataframe = merge_informative_pair(dataframe, informative, self.timeframe, inf_tf, ffill=True)

        dataframe['rsi'] = ta.RSI(dataframe)
        dataframe['plot_rsi'] = self.plot_RSI(dataframe)['Plot_rsi']
        dataframe['ha_open'] = heikinashi(dataframe)['open']
        dataframe['ha_high'] = heikinashi(dataframe)['high']
        dataframe['ha_low'] = heikinashi(dataframe)['low']
        dataframe['ha_close'] = heikinashi(dataframe)['close']
        dataframe['doji_cand'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, 1)['Doji_cand']
        dataframe['bear_cand'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, 1)['Bear_cand']
        dataframe['bull_cand'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, 1)['Bull_cand']
        dataframe['red_or_green'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, 1)['Red_or_green']
        dataframe['plot_bull'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, 1)['Plot_bull']
        dataframe['plot_bear'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, 1)['Plot_bear']
        dataframe['plot_doji'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, 1)['Plot_doji']
        for val in self.buy_willy.range:
            dataframe[f'willy{val}'] = self.willy_ema(dataframe, low_line=self.buy_low_line.value,
                                                      up_line=self.buy_high_line.value, willyLen=val, emaLen=13)[
                'Willy']
        for val in self.buy_ema.range:
            dataframe[f'ema_will{val}'] = self.willy_ema(dataframe, low_line=-80, up_line=-20, willyLen=21, emaLen=val)[
                'Ema_will']
        for val in self.buy_sl_mult.range:
            dataframe[f'sl_l{val}'] = self.cust_stoploss(dataframe, length=20, mult=val)['Sl_low']
        for val in self.buy_sl_mult.range:
            dataframe[f'sl_h{val}'] = self.cust_stoploss(dataframe, length=20, mult=val)['Sl_high']

        for val in self.sell_willy.range:
            dataframe[f'sell_willy{val}'] = self.willy_ema(dataframe, low_line=self.buy_low_line.value,
                                                           up_line=self.buy_high_line.value, willyLen=val, emaLen=13)[
                'Willy']
        for val in self.sell_ema.range:
            dataframe[f'sell_ema_will{val}'] = \
                self.willy_ema(dataframe, low_line=-80, up_line=-20, willyLen=21, emaLen=val)[
                    'Ema_will']
        for val in self.sell_doji_in_row.range:
            dataframe[f'doji_in_row{val}'] = self.heikin_asi_trigger(dataframe, metadata, val, 1, 1)['Doji_in_row']
        for val in self.buy_bull_row.range:
            dataframe[f'buy_bull_row{val}'] = self.heikin_asi_trigger(dataframe, metadata, 1, val, 1)['Bull_in_row']
        for val in self.buy_bear_row.range:
            dataframe[f'buy_bear_row{val}'] = self.heikin_asi_trigger(dataframe, metadata, 1, 1, val)['Bear_in_row']

        return dataframe

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

        conditions = [(dataframe['volume'] > 0)]

        conditions.append((dataframe['rsi'] > self.buy_RSI_long.value) &  # Signal: RSI
                          (dataframe[f'buy_bull_row{self.buy_bull_row.value}'] == self.buy_bull_row.value))

        if self.buy_on_willy.value:
            conditions.append((dataframe[f'willy{self.buy_willy.value}'] < self.buy_low_line.value) &
                              (dataframe[f'ema_will{self.buy_ema.value}'] < self.buy_low_line.value))

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

        conditions = [(dataframe['volume'] > 0)]
        conditions.append((dataframe['rsi'] < self.buy_RSI_short.value) &  # Signal: RSI
                          (dataframe[f'buy_bear_row{self.buy_bear_row.value}'] == self.buy_bear_row.value))

        if self.buy_on_willy.value:
            conditions.append((dataframe[f'willy{self.buy_willy.value}'] > self.buy_high_line.value) &
                              (dataframe[f'ema_will{self.buy_ema.value}'] > self.buy_high_line.value))

        if Debug:
            print(conditions)
        sell_conditions = reduce(lambda x, y: x & y, conditions)
        dataframe.loc[sell_conditions, 'enter_short'] = 1

        if Debug:
            print(dataframe.tail(20))

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        if not self.sell_willy_exit.value:
            conditions = [(dataframe['bear_cand'] > 0) |
                          (dataframe[f'doji_in_row{self.sell_doji_in_row.value}'] == self.sell_doji_in_row.value)
                          ]
        else:
            conditions = [(dataframe['bear_cand'] > 0) |
                          (dataframe[f'doji_in_row{self.sell_doji_in_row.value}'] == self.sell_doji_in_row.value) |
                          ((dataframe[f'sell_willy{self.sell_willy.value}'] > self.sell_high_line.value) &
                           (dataframe[f'sell_ema_will{self.sell_ema.value}'] > self.sell_high_line.value))
                          ]

        sell_conditions = reduce(lambda x, y: x & y, conditions)
        dataframe.loc[sell_conditions, 'exit_long'] = 1

        if not self.sell_willy_exit.value:
            conditions = [(dataframe['bull_cand'] > 0) |
                          (dataframe[f'doji_in_row{self.sell_doji_in_row.value}'] == self.sell_doji_in_row.value)]
        else:
            conditions = [(dataframe['bull_cand'] > 0) |
                          (dataframe[f'doji_in_row{self.sell_doji_in_row.value}'] == self.sell_doji_in_row.value) |
                          ((dataframe[f'sell_willy{self.sell_willy.value}'] < self.sell_low_line.value) &
                          (dataframe[f'sell_ema_will{self.sell_ema.value}'] < self.sell_low_line.value))
                          ]
        sell_conditions = reduce(lambda x, y: x & y, conditions)
        dataframe.loc[sell_conditions, 'exit_short'] = 1

        if Debug:
            print(dataframe.tail(30))
        return dataframe

    def plot_RSI(self, dataframe: DataFrame):
        df = dataframe.copy()
        df['drop_rsi'] = ta.RSI(df)
        df['plot_rsi'] = np.where((df["drop_rsi"] > self.buy_RSI_long.value), df['high'] + (df['high'] * 0.01),
                                  np.where((df["drop_rsi"] < self.buy_RSI_short.value),
                                           df['low'] - (df['low'] * 0.01), np.nan))

        df.drop(['drop_rsi'], inplace=True, axis=1)
        return DataFrame(index=df.index, data={
            'Plot_rsi': df['plot_rsi']
        })

    def heikin_asi_trigger(self, dataframe: DataFrame, metadata: dict, doji_sum, bull_row, bear_row):
        def _flat_top(close, low, open, high):

            if high == open and low < close:
                return 1
            else:
                return 0

        def _wick_length(close, low, open, high):

            if close > open:
                top_wick = high - close
                bottom_wick = open - low

            else:
                top_wick = high - open
                bottom_wick = close - low

            if top_wick > 0 and bottom_wick > 0:
                return 1
            else:
                return 0

        def _is_red_green(open, close):
            if open < close:
                return 1
            else:
                return 0

        def _flat_bottom(close, low, open, high):

            if open == low and high > close:
                return 1
            else:
                return 0

        df = dataframe.copy()
        df['ha_open'] = heikinashi(df)['open']
        df['ha_high'] = heikinashi(df)['high']
        df['ha_low'] = heikinashi(df)['low']
        df['ha_close'] = heikinashi(df)['close']
        df['ha_red_or_green'] = np.vectorize(_is_red_green)(df['ha_open'], df['ha_close'])
        df['bull_cand'] = np.vectorize(_flat_bottom)(df['ha_close'], df['ha_low'], df['ha_open'], df['ha_high'])
        df['plot_bull'] = np.where(df['bull_cand'] > 0, df['high'], np.NaN)
        df['bear_cand'] = np.vectorize(_flat_top)(df['ha_close'], df['ha_low'], df['ha_open'], df['ha_high'])
        df['plot_bear'] = np.where(df['bear_cand'] > 0, df['low'], np.NaN)
        df['doji_cand'] = np.vectorize(_wick_length)(df['ha_close'], df['ha_low'], df['ha_open'], df['ha_high'])
        df['plot_doji'] = np.where(df['doji_cand'] > 0, df['close'], np.NaN)
        df['doji_in_row'] = df['doji_cand'].rolling(window=doji_sum).sum()
        df['bull_in_row'] = df['bull_cand'].rolling(window=bull_row).sum()
        df['bear_in_row'] = df['bear_cand'].rolling(window=bear_row).sum()

        df.drop(['ha_open', 'ha_high', 'ha_low', 'ha_close'], inplace=True, axis=1)

        return DataFrame(index=df.index, data={
            'Doji_cand': df['doji_cand'],
            'Bear_cand': df['bear_cand'],
            'Bull_cand': df['bull_cand'],
            'Red_or_green': df['ha_red_or_green'],
            'Plot_bull': df['plot_bull'],
            'Plot_bear': df['plot_bear'],
            'Plot_doji': df['plot_doji'],
            'Doji_in_row': df['doji_in_row'],
            'Bull_in_row': df['bull_in_row'],
            'Bear_in_row': df['bear_in_row']

        })

    def willy_ema(self, dataframe: DataFrame, low_line=-80.0, up_line=-20.0, willyLen=21, emaLen=13):

        def max_help(open, high, low, close):
            return max(open, high, low, close)

        def min_help(open, high, low, close):
            return min(open, high, low, close)

        df = dataframe.copy()
        df['highest'] = np.vectorize(max_help)(df['open'], df['high'], df['low'], df['close'])
        df['lowest'] = np.vectorize(min_help)(df['open'], df['high'], df['low'], df['close'])
        df['upper'] = df['highest'].rolling(willyLen).max()
        df['lower'] = df['lowest'].rolling(willyLen).min()
        df['willy'] = 100 * (df['close'] - df['upper']) / (df['upper'] - df['lower'])
        df['ema_will'] = ta.EMA(df['willy'], emaLen)
        # df['ema_willy'] =
        df.drop(['highest', 'lowest', 'upper', 'lower'], inplace=True, axis=1)
        df['low_line'] = low_line
        df['up_line'] = up_line
        return DataFrame(index=df.index, data={
            'Willy': df['willy'],
            'Ema_will': df['ema_will'],
            'Low_line': df['low_line'],
            'Up_line': df['up_line']
        })

    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:

        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        candle = dataframe.iloc[-1].squeeze()
        stoploss_pricel = candle[f'sl_l{self.buy_sl_mult.value}']
        stoploss_prices = candle[f'sl_h{self.buy_sl_mult.value}']
        middle_of_cand = (candle['high'] + candle['low'] + candle['close'] + candle['open']) / 4
        if candle['ha_open'] < candle['ha_close']:
            stoploss_ = (middle_of_cand - stoploss_pricel) / candle['close'] * 100

            amount_ = self.wallets.get_total_stake_amount() / 100
            if Debug:
                print('stoploss from amount = >', stoploss_)
                print('полній кошилек =>>>', self.wallets.get_total_stake_amount())
                print('amount', amount_)
                print('Amount final =>>', amount_ / stoploss_ * 100)
            return (amount_ / stoploss_) * 100

        if candle['ha_open'] > candle['ha_close']:
            stoploss_ = (middle_of_cand - stoploss_prices) / candle['close'] * -1 * 100
            amount_ = self.wallets.get_total_stake_amount() / 100
            if Debug:
                print('stoploos from amount =>', stoploss_)
                print('полній кошилек =>>>', self.wallets.get_total_stake_amount())
                print('amount', amount_)
                print('Amount =>>', (amount_ / stoploss_) * 100)
            return (amount_ / stoploss_) * 100

        return 2000

    use_custom_stoploss = True

    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: 'datetime',
                        current_rate: float, current_profit: float, **kwargs) -> float:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        candle = dataframe.iloc[-1].squeeze()

        stoploss_pricel = candle[f'sl_l{self.buy_sl_mult.value}']
        stoploss_prices = candle[f'sl_h{self.buy_sl_mult.value}']
        middle_of_cand = (candle['high'] + candle['low'] + candle['close'] + candle['open']) / 4
        if Debug:
            print('trade ID =>', trade.id, 'Trade is short? =>', trade.is_short)
            print('sl hihg =>', candle[f'sl_h{self.buy_sl_mult.value}'], 'sl low =>', candle[f'sl_l{self.buy_sl_mult.value}'])
            print('candle middle =>', middle_of_cand)
        if not trade.is_short:
            if Debug:
                print('   ')
                print('Trade is long', 'Stoploss ================================>',
                      (middle_of_cand - stoploss_pricel) / candle['close'] * -1)
                print('   ')
            return (middle_of_cand - stoploss_pricel) / candle['close'] * -1

        if trade.is_short:
            if Debug:
                print('   ')
                print('Trade is short', 'Stoploss ==============================>',
                      (middle_of_cand - stoploss_prices) / candle['close'])
                print('   ')
            return (middle_of_cand - stoploss_prices) / candle['close']

        return 100

    def cust_stoploss(self, dataframe: DataFrame, mult, length):
        df = dataframe.copy()
        df['sma'] = qtpylib.sma(df['close'], length)
        df['tr'] = qtpylib.atr(df, length)

        df['sl_high'] = qtpylib.sma(df['tr'], length)

        def help1(x, mul, high):
            return x * mul + high

        df['sl_high'] = np.vectorize(help1)(df['sl_high'], mult, df['high'])

        df['sl_low'] = qtpylib.sma(df['tr'], length)

        def help1(x, mul, low):
            return low - x * mul

        df['sl_low'] = np.vectorize(help1)(df['sl_low'], mult, df['low'])
        df.drop(['sma', 'tr'], axis=1, inplace=True)
        return DataFrame(index=df.index, data={
            'Sl_high': df['sl_high'],
            'Sl_low': df['sl_low'],
        })