AdaptiveRenkoStrategy

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these libs ---
import logging
import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime
from typing import Optional, Union
from freqtrade.strategy import BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter, informative, merge_informative_pair
# --------------------------------
# Add your lib to import here
from freqtrade.exchange import timeframe_to_minutes
# https://github.com/quantroom-pro/pyrenko
import pyrenko 
import pandas_ta as pta
from technical import qtpylib
import scipy.optimize as opt

logger = logging.getLogger(__name__)

def evaluate_renko(brick, history, column_name):
    renko_obj = pyrenko.renko()
    renko_obj.set_brick_size(brick_size = brick, auto = False)
    renko_obj.build_history(prices = history)
    return renko_obj.evaluate()[column_name]

class AdaptiveRenkoStrategy(IStrategy):
    
    """
    This strategy is based on research work by Sergey Malchevskiy and uses Renko bricks to identify trends.
    The brick size is optimized using ATR to find out boundaries from an informative period.
    For more information on the brick size optimization, see the following article: 
    https://towardsdatascience.com/renko-brick-size-optimization-34d64400f60e

    This strategy is designed to work with FreqTrade and can be used in conjunction with FreqAI.
    To match with research, brick size optimization should change from ATR to IQR from source data
    For more information on using Bayesian optimization on this strategy as a training model, see the following article: 
    https://towardsdatascience.com/bayesian-optimization-in-trading-77202ffed530

    Author: Mr Robot (@heymrrobot)
    """    
    
    INTERFACE_VERSION = 3

    timeframe = '3m'
    timeframe_minutes = timeframe_to_minutes(timeframe)
    can_short: bool = True
    trailing_stop = False
    startup_candle_count: int = 7*2
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False
    process_only_new_candles = True
    
    # Initialize renko_obj_obs as a dictionary
    custom_renkodict = {}
                       
    def informative_pairs(self):
        pairs = self.dp.current_whitelist()
        informative_pairs = [(pair, '15m') for pair in pairs]
        return informative_pairs
        
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        inf_tf = '15m'
        informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=inf_tf)
        # Calculate the ATR (Average True Range) for the informative DataFrame
        atr = pta.atr(informative['high'], informative['low'], informative['close'], timeperiod=10)
        # Add the ATR values as a new column to the informative DataFrame
        informative['atr'] = [0 if np.isnan(x) else x for x in atr]
        dataframe = merge_informative_pair(dataframe, informative, self.timeframe, inf_tf, ffill=True)
        
        # If the pair is not in the custom_renkodict dictionary, add it with a new pyrenko.renko() object
        if metadata['pair'] not in self.custom_renkodict:
            self.custom_renkodict[metadata['pair']] = pyrenko.renko()
            # Optimize the Renko brick size using fminbound for the new pyrenko.renko() object using the fminbound optimization function
            opt_bs = opt.fminbound(lambda x: -evaluate_renko(brick=x, history=dataframe.close, column_name='score'), np.min(dataframe['atr_15m']), np.max(dataframe['atr_15m']), disp=0)
            # Create a new pyrenko.renko() object for the pair with the optimized brick size and add it to the custom_renkodict dictionary
            self.custom_renkodict[metadata['pair']].set_brick_size(brick_size=opt_bs, auto=False)
            self.custom_renkodict[metadata['pair']].build_history(prices=dataframe.close)
            logger.info(f'Created a new pyrenko.renko() object for pair {metadata["pair"]}')
        
        # Get the renko object for the given pair from custom_renkodict    
        renko_obj = self.custom_renkodict[metadata['pair']]
        # Get the list of renko prices from the renko object
        prices = renko_obj.get_renko_prices()
        logger.info(f'Got the list of renko prices for pair {metadata["pair"]}')
        # Get the list of renko directions from the renko object
        directions = renko_obj.get_renko_directions()
        self.prev_brick_direction = directions[-2] if len(directions) >= 2 else None
        self.last_brick_direction = directions[-1] if len(directions) >= 1 else None
    
        # Check if there are any prices in the prices list
        if len(prices) > 0:
            # Create the next Renko bars using the last close price from the dataframe and get the number of created bars
            num_created_bars = renko_obj.do_next(dataframe.close.iloc[-1])
            logger.info(f'Created {num_created_bars} new Renko bars for pair {metadata["pair"]}')
            # If new bars were created
            if num_created_bars != 0:
                # If the previous brick direction is not the same as the last brick direction
                if self.prev_brick_direction != self.last_brick_direction:
                    opt_bs = opt.fminbound(lambda x: -evaluate_renko(brick=x, history=dataframe.close, column_name='score'), np.min(dataframe['atr_15m']), np.max(dataframe['atr_15m']), disp=0)
                    if opt_bs != renko_obj.brick_size:
                        renko_obj.set_brick_size(brick_size=opt_bs, auto=False)
                        self.opt_bs = opt_bs

        # Evaluate the Renko bars and update the score and directions dictionaries
        renko_eval = renko_obj.evaluate()
        logger.info(f'Evaluated the Renko bars for pair {metadata["pair"]}, values: {renko_eval}, brick size: {renko_obj.brick_size}')
        self.score = {metadata['pair']: renko_eval['score']}
        self.directions = {metadata['pair']: directions}
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Add two new columns to the DataFrame for entering a long and short position, and initialize them to zero.
        dataframe['enter_long'] = 0
        dataframe['enter_short'] = 0

        # Check if the current pair is not in the custom_renkodict, or if the score for the pair is less than 4, or if the length of the directions list for the pair is less than 2.
        # If any of these conditions is true, return the original DataFrame without making any changes.
        if metadata['pair'] not in self.custom_renkodict or self.score.get(metadata['pair'], -1) < 4 or len(self.directions[metadata['pair']]) < 2:
            return dataframe

        # Get the direction of the last brick and the direction of the second-to-last brick for the current pair from the directions list.
        last_brick_direction = self.directions[metadata['pair']][-1]
        prev_brick_direction = self.directions[metadata['pair']][-2]
        
        # If the direction of the last brick is up (1) and the direction of the previous brick is down (-1), 
        # set the value of enter_long to 1 and enter_tag to 'buy_direction' for all rows in the enter_long and enter_tag columns vice versa.
        if last_brick_direction == 1 and prev_brick_direction == -1:
            dataframe.loc[:, ['enter_long', 'enter_tag']] = (1, 'buy_direction')
            logger.info(f'Set the value of enter_long to 1 and enter_tag to "buy_direction" for pair {metadata["pair"]}')
        elif last_brick_direction == -1 and prev_brick_direction == 1:
            dataframe.loc[:, ['enter_short', 'enter_tag']] = (1, 'short_direction')
            logger.info(f'Set the value of enter_short to 1 and enter_tag to "short_direction" for pair {metadata["pair"]}')
            
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        if metadata['pair'] not in self.custom_renkodict:
            return dataframe
        
        last_brick_direction = self.directions[metadata['pair']][-1]
        prev_brick_direction = self.directions[metadata['pair']][-2]

        if last_brick_direction == -1 and prev_brick_direction == 1:
            dataframe.loc[:, ['exit_long', 'enter_tag']] = (1, 'direction_long_exit')
            logger.info(f'Set the value of exit_long to 1 and enter_tag to "direction_long_exit" for pair {metadata["pair"]}')
            # Flush the custom_renkodict for the corresponding pair
            # del self.custom_renkodict[metadata['pair']]

        elif last_brick_direction == 1 and prev_brick_direction == -1:
            dataframe.loc[:, ['exit_short', 'enter_tag']] = (1, 'direction_short_exit')
            logger.info(f'Set the value of exit_short to 1 and enter_tag to "direction_short_exit" for pair {metadata["pair"]}')
            # Flush the custom_renkodict for the corresponding pair
            # del self.custom_renkodict[metadata['pair']]
        return dataframe