CusTrend_coralTrend_Adx_EMA_Oct_1h

# 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


from freqtrade.persistence import Trade

from freqtrade.strategy import (BooleanParameter, CategoricalParameter, stoploss_from_open, DecimalParameter,
                                IntParameter, IStrategy, informative, merge_informative_pair)

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

# custom indicators
# #############################################################################################################################################################################################


#  Coral Trend Indicator
'''
The "Coral Trend Indicator" is a trend-following indicator that uses a combination of exponential moving averages (EMAs) and the Commodity Channel Index (CCI) to identify trends and potential reversal points

This code imports the necessary libraries (pandas and numpy) and defines a function called coral_trend() that takes a DataFrame of cryptocurrency data (with a "close" column) and the EMA period, CCI period, and CCI threshold as inputs. The function first calculates the fast and slow exponential moving averages (EMAs) of the cryptocurrency and stores them in new columns called "ema_fast" and "ema_slow", respectively. It then calculates the Commodity Channel Index (CCI) of the cryptocurrency and stores it in a new column called "cci".

The function then determines the trend based on the EMAs and CCI, and stores it in a new column called "coral_trend".

The coral_trend() function determines the trend by using the np.where() function to set the value of the "coral_trend" column based on the following conditions:

    If the fast EMA is greater than the slow EMA and the CCI is less than the negative CCI threshold, then the trend is set to 1 (indicating an uptrend).
    If the fast EMA is less than the slow EMA and the CCI is greater than the positive CCI threshold, then the trend is set to -1 (indicating a downtrend).
    Otherwise, the trend is set to 0 (indicating no trend or a neutral market).

The coral_trend() function then returns the modified DataFrame with the "coral_trend" column added.
'''
def coral_trend(df, ema_period=10, cci_period=20, cci_threshold=100):
    # Calculate the exponential moving averages (EMAs)
    df['ema_fast'] = df['close'].ewm(span=ema_period, adjust=False).mean()
    df['ema_slow'] = df['close'].ewm(span=ema_period*2, adjust=False).mean()
    
    # Calculate the Commodity Channel Index (CCI)
    df['cci'] = ((df['close'] - df['close'].rolling(cci_period).mean()) / df['close'].rolling(cci_period).std()) * np.sqrt(cci_period)
    
    # Determine the trend based on the EMAs and CCI
    df['coral_trend'] = np.where((df['ema_fast'] > df['ema_slow']) & (df['cci'] < -cci_threshold), 1, 0)
    df['coral_trend'] = np.where((df['ema_fast'] < df['ema_slow']) & (df['cci'] > cci_threshold), -1, df['coral_trend'])
    
    return df


#double up and down closing based trend

def determine_trend(df):
    df['trend'] = 0
    for i in range(2, len(df)):
        close_prev = df['close'].iloc[i-1]
        close = df['close'].iloc[i]
        high_prev = df['high'].iloc[i-1]
        low_prev = df['low'].iloc[i-1]
        open_prev = df['open'].iloc[i-1]
        high = df['high'].iloc[i]
        low = df['low'].iloc[i]
        open = df['open'].iloc[i] 

        if (
            close_prev < close and
            high_prev < high and
            high_prev < close and
            low_prev < low and  # Close of previous candle is lower than present candle
            close - open > (close_prev - open_prev) and
            # close - open < 3* (close_prev - open_prev) and
            open < close and 
            open_prev < close_prev and
            # (close_prev - open_prev) < (open_prev - low_prev) and
            high_prev - close_prev < close_prev - open_prev and  # High - Close < Close - Open for previous candle
            high - close < close - open  # High - Close < Close - Open for present candle
        ):
            df.at[i, 'trend'] = 1

        elif (
            close_prev > close and
            high_prev > high and
            low_prev > low and   # Close of previous candle is higher than present candle
            low_prev > close and   # Close of previous candle is higher than present candle
            open > close and 
            open_prev > close_prev and
            # (open_prev - close_prev) < (high_prev - open_prev) and
            open - close > (open_prev - close_prev) and
            # open - close < 3* (open_prev - close_prev) and
            close_prev - low_prev < open_prev - close_prev and  # Close - Low < Open - Close for previous candle
            close - low < open - close  # Close - Low < Open - Close for present candle
        ):
            df.at[i, 'trend'] = -1

    return df



# ############################################################################################################################################################################################

class CusTrend_coralTrend_Adx_EMA_Oct_1h(IStrategy):

    
    # Strategy interface version - allow new iterations of the strategy interface.
    # Check the documentation or the Sample strategy to get the latest version.
    INTERFACE_VERSION = 3

    # Optimal timeframe for the strategy.
    timeframe = '1h'

    # Can this strategy go short?
    can_short = True



    '''

 




docker-compose run freqtrade backtesting --strategy CusTrend_coralTrend_Adx_EMA_Oct_1h -i 1h --export trades --breakdown month --timerange 20210101-20230525


    '''

    # Minimal ROI designed for the strategy.
    minimal_roi = {'0': 0.101, '373': 0.068, '1088': 0.025, '1336': 0}

    # Optimal stoploss designed for the strategy.
    # This attribute will be overridden if the config file contains "stoploss".
    stoploss = -0.347

    # Trailing stop:
    trailing_stop = True
    trailing_stop_positive = 0.011
    trailing_stop_positive_offset = 0.012
    trailing_only_offset_is_reached = True

    # Run "populate_indicators()" only for new candle.
    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

    # Number of candles the strategy requires before producing valid signals
    startup_candle_count: int = 30

    #leverage here
    leverage_optimize = True
    leverage_num = IntParameter(low=1, high=5, default=5, space='buy', optimize=leverage_optimize)

    # Strategy parameters
    parameters_yes = True
    parameters_no = False

    '''
    adx_long_min = IntParameter(4, 21, default=9, space="buy", optimize = parameters_yes)
    adx_long_max = IntParameter(21, 56, default=38, space="buy", optimize = parameters_yes)
    adx_short_min = IntParameter(4, 21, default=21, space="buy", optimize = parameters_yes)
    adx_short_max = IntParameter(20, 56, default=29, space="buy", optimize = parameters_yes)

    ema_period = IntParameter(22, 200, default=175, space="buy", optimize= parameters_yes)

    volume_check = IntParameter(15, 45, default=36, space="buy", optimize= parameters_yes)
    volume_check_exit = IntParameter(15, 45, default=35, space="sell", optimize= parameters_yes)

    sell_shift = IntParameter(1, 6, default=6, space="sell", optimize= parameters_yes)
    sell_shift_short = IntParameter(1, 6, default=6, space="sell", optimize= parameters_yes) 
    
    '''
    adx_long_min = IntParameter(4, 21, default=17, space="buy", optimize = parameters_yes)
    adx_long_max = IntParameter(21, 56, default=34, space="buy", optimize = parameters_yes)
    adx_short_min = IntParameter(4, 21, default=15, space="buy", optimize = parameters_yes)
    adx_short_max = IntParameter(20, 56, default=42, space="buy", optimize = parameters_yes)

    ema_period = IntParameter(22, 200, default=50, space="buy", optimize= parameters_yes)

    volume_check = IntParameter(15, 45, default=39, space="buy", optimize= parameters_yes)
    volume_check_exit = IntParameter(15, 45, default=41, space="sell", optimize= parameters_yes)

    sell_shift = IntParameter(1, 6, default=6, space="sell", optimize= parameters_yes)
    sell_shift_short = IntParameter(1, 6, default=5, space="sell", optimize= parameters_yes)  # **added this at random-state 11165



    protect_optimize = True
    # cooldown_lookback = IntParameter(1, 40, default=4, space="protection", optimize=protect_optimize)
    max_drawdown_lookback = IntParameter(1, 50, default=2, space="protection", optimize=protect_optimize)
    max_drawdown_trade_limit = IntParameter(1, 3, default=1, space="protection", optimize=protect_optimize)
    max_drawdown_stop_duration = IntParameter(1, 50, default=4, space="protection", optimize=protect_optimize)
    max_allowed_drawdown = DecimalParameter(0.05, 0.30, default=0.10, decimals=2, space="protection",
                                            optimize=protect_optimize)
    stoploss_guard_lookback = IntParameter(1, 50, default=8, space="protection", optimize=protect_optimize)
    stoploss_guard_trade_limit = IntParameter(1, 3, default=1, space="protection", optimize=protect_optimize)
    stoploss_guard_stop_duration = IntParameter(1, 50, default=4, space="protection", optimize=protect_optimize)

    @property
    def protections(self):
        return [
            # {
            #     "method": "CooldownPeriod",
            #     "stop_duration_candles": self.cooldown_lookback.value
            # },
            {
                "method": "MaxDrawdown",
                "lookback_period_candles": self.max_drawdown_lookback.value,
                "trade_limit": self.max_drawdown_trade_limit.value,
                "stop_duration_candles": self.max_drawdown_stop_duration.value,
                "max_allowed_drawdown": self.max_allowed_drawdown.value
            },
            {
                "method": "StoplossGuard",
                "lookback_period_candles": self.stoploss_guard_lookback.value,
                "trade_limit": self.stoploss_guard_trade_limit.value,
                "stop_duration_candles": self.stoploss_guard_stop_duration.value,
                "only_per_pair": False
            }
        ]

 


    # ema_long = IntParameter(50, 250, default=100, space="buy")
    # ema_short = IntParameter(50, 250, default=100, space="buy")

    # sell_rsi = IntParameter(60, 90, default=70, space="sell")

    # Optional order type mapping.
    order_types = {
        'entry': 'limit',
        'exit': 'limit',
        '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'},
            },
            'subplots': {
                # Subplots - each dict defines one additional plot
                "MACD": {
                    'macd': {'color': 'blue'},
                    'macdsignal': {'color': 'orange'},
                },
                "RSI": {
                    'rsi': {'color': 'red'},
                }
            }
        }

    def informative_pairs(self):
        
        # get access to all pairs available in whitelist.
        pairs = self.dp.current_whitelist()
        informative_pairs = [(pair, '4h') for pair in pairs]
        
        return informative_pairs
    

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        if not self.dp:
            # Don't do anything if DataProvider is not available.
            return dataframe
        
        inf_tf = '4h'
        # Get the informative pair
        informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=inf_tf)
        # Get the 14 day rsi
        informative['ema'] = ta.EMA(informative['close'], timeperiod=50)

        # Use the helper function merge_informative_pair to safely merge the pair
        # Automatically renames the columns and merges a shorter timeframe dataframe and a longer timeframe informative pair
        # use ffill to have the 1d value available in every row throughout the day.
        # Without this, comparisons between columns of the original and the informative pair would only work once per day.
        # Full documentation of this method, see below
        dataframe = merge_informative_pair(dataframe, informative, self.timeframe, inf_tf, ffill=True)

        L_determine_trend_strategy = determine_trend(df = dataframe)
        dataframe['trend'] = L_determine_trend_strategy['trend']


        # long_coral_trend = coral_trend(df=dataframe, ema_period = self.ct_ema_period.value, cci_period=self.ct_cci_period.value, cci_threshold = self.ct_cci_threshold.value)

        # dataframe['ct_ema_fast']= long_coral_trend['ema_fast']
        # dataframe['ct_ema_slow'] = long_coral_trend['ema_slow']
        # dataframe['ct_cci'] = long_coral_trend['cci']
        # dataframe['ct_coral_trend'] = long_coral_trend['coral_trend'] 


        # dataframe['psar'] = calculate_psar(df=dataframe, af_start=self.af_start_range.value, af_max=self.af_max_range.value)
        
        # Parabolic SAR
        dataframe['psar'] = ta.SAR(dataframe)

        

        # ADX
        dataframe['adx'] = ta.ADX(dataframe)
        
        # RSI
        dataframe['rsi'] = ta.RSI(dataframe)

        # EMA
        dataframe['ema'] = ta.EMA(dataframe['close'], timeperiod=self.ema_period.value)

        # Volume Weighted
        dataframe['volume_mean'] = dataframe['volume'].rolling(self.volume_check.value).mean().shift(1)
        dataframe['volume_mean_exit'] = dataframe['volume'].rolling(self.volume_check_exit.value).mean().shift(1)


        return dataframe

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

        dataframe.loc[
            (
                        (dataframe['adx'] > self.adx_long_min.value) & # trend strength confirmation
                        (dataframe['adx'] < self.adx_long_max.value) & # trend strength confirmation
                        # (dataframe['adx'] > dataframe['adx'].shift(1)) & 
                        (dataframe['psar'] < dataframe['close']) & 
                        (dataframe['ema'] < dataframe['close']) & 
                        (dataframe['ema_4h'] < dataframe['close']) & 
                        # (dataframe['eth_ema_15m'] < dataframe['close']) & 
                        # (dataframe['btc_ema_30m'] < dataframe['close']) &  
                        # (dataframe['eth_ema_30m'] < dataframe['close']) & 
                        (dataframe['trend'] == 1) &
                        # (dataframe['ct_coral_trend'] > 0) & 
                        (dataframe['rsi'] > 50) &
                        # (dataframe['volume'] > dataframe['volume'].shift(1)) &
                        (dataframe['volume'] > dataframe['volume_mean'])

            ),
            'enter_long'] = 1

        dataframe.loc[
            (
                        (dataframe['adx'] > self.adx_short_min.value) & # trend strength confirmation
                        (dataframe['adx'] < self.adx_short_max.value) & # trend strength confirmation
                        # (dataframe['adx'] > dataframe['adx'].shift(1)) & 
                        (dataframe['psar'] > dataframe['close']) & # trend reversal confirmation
                        (dataframe['ema'] > dataframe['close']) & # trend confirmation
                        (dataframe['ema_4h'] > dataframe['close']) & 
                        # (dataframe['btc_ema_15m'] > dataframe['close']) & 
                        # (dataframe['eth_ema_15m'] > dataframe['close']) & 
                        # (dataframe['btc_ema_30m'] > dataframe['close']) & 
                        # (dataframe['eth_ema_30m'] > dataframe['close']) & 
                        (dataframe['trend'] == -1) & 
                        # (dataframe['ct_coral_trend'] < 0) & 
                        (dataframe['rsi'] < 50) & # momentum indicator
                        # (dataframe['volume'] > dataframe['volume'].shift(1)) &
                        (dataframe['volume'] > dataframe['volume_mean']) # volume weighted indicator
            ),
            'enter_short'] = 1
        
        return dataframe

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

        conditions_long = []
        conditions_short = []
        dataframe.loc[:, 'exit_tag'] = ''

        exit_long = (
                (dataframe['close'] < dataframe['low'].shift(self.sell_shift.value)) &
                (dataframe['volume'] > dataframe['volume_mean_exit'])
        )

        exit_short = (
                (dataframe['close'] > dataframe['high'].shift(self.sell_shift_short.value)) &
                (dataframe['volume'] > dataframe['volume_mean_exit'])
        )


        conditions_short.append(exit_short)
        dataframe.loc[exit_short, 'exit_tag'] += 'exit_short'


        conditions_long.append(exit_long)
        dataframe.loc[exit_long, 'exit_tag'] += 'exit_long'


        if conditions_long:
            dataframe.loc[
                reduce(lambda x, y: x | y, conditions_long),
                'exit_long'] = 1

        if conditions_short:
            dataframe.loc[
                reduce(lambda x, y: x | y, conditions_short),
                'exit_short'] = 1
            
        return dataframe

    def leverage(self, pair: str, current_time: datetime, current_rate: float,
                 proposed_leverage: float, max_leverage: float, side: str,
                 **kwargs) -> float:

        return self.leverage_num.value