FSD – Custom indicators part 2

DISCLAIMER
The codes in this post is just for educational purpose, and not to be used on live trading

If we use existing functions to create our custom indicators on previous post, in this post we will create a new function to calculate our custom indicator.

One of popular indicator that hasn’t been implemented in the existing technical libraries is Elliot Wave Oscillator. Below is the code to calculate it (credit to whoever wrote it first. I’m just using it). Put it just above your strategy class definition.

def EWO(source, sma_length=5, sma2_length=35):
    sma1 = ta.SMA(source, timeperiod=sma_length)
    sma2 = ta.SMA(source, timeperiod=sma2_length)
    smadif = (sma1 - sma2) / source * 100
    return smadif

class strat_template (IStrategy):

And to use it, just call it from populate_indicators. The full code will look like this

from freqtrade.strategy import IStrategy, informative
from pandas import DataFrame
import freqtrade.vendor.qtpylib.indicators as qtpylib
import talib.abstract as ta
from freqtrade.persistence import Trade
from datetime import datetime, timedelta
from typing import Optional, Union

def EWO(source, sma_length=5, sma2_length=35):
    sma1 = ta.SMA(source, timeperiod=sma_length)
    sma2 = ta.SMA(source, timeperiod=sma2_length)
    smadif = (sma1 - sma2) / source * 100
    return smadif

class strat_template (IStrategy):

    def version(self) -> str:
        return "template-v1"

    INTERFACE_VERSION = 3

    minimal_roi = {
        "0": 0.05
    }

    stoploss = -0.05

    timeframe = '15m'

    process_only_new_candles = True
    startup_candle_count = 999

    use_custom_stoploss = True

    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float:
        
        sl_new = 1

        if (current_time - timedelta(minutes=15) >= trade.open_date_utc):

            dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
            current_candle = dataframe.iloc[-1].squeeze()
            current_profit = trade.calc_profit_ratio(current_candle['close'])

            if (current_profit >= 0.03):
                sl_new = 0.01

        return sl_new

    def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]:

        if ((current_time - timedelta(minutes=15)) >= trade.open_date_utc):

            dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
            current_candle = dataframe.iloc[-1].squeeze()
            current_profit = trade.calc_profit_ratio(current_candle['close'])
            
            if (current_profit >= 0):
                if (current_candle['rsi'] >= 70):
                    return "rsi_overbought"

    @informative('30m')
    def populate_indicators_inf1(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        
        dataframe['rsi'] = ta.RSI(dataframe['close'], 14)

        return dataframe

    @informative('1h')
    def populate_indicators_inf2(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        
        dataframe['rsi'] = ta.RSI(dataframe['close'], 14)

        return dataframe

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

        dataframe['ema_9'] = ta.EMA(dataframe['close'], 9)
        dataframe['ema_20'] = ta.EMA(dataframe['close'], 20)
        dataframe['rsi'] = ta.RSI(dataframe['close'], 14)
        dataframe['ema_9_rsi'] = ta.EMA(dataframe['rsi'], 9)
        dataframe['ewo'] = EWO(dataframe['close'], 50, 200)

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

        dataframe.loc[
            qtpylib.crossed_above(dataframe['ema_9'], dataframe['ema_20'])
            &
            (dataframe['rsi_30m'] < 50)
            &
            (dataframe['rsi_1h'] < 30)
            &
            (dataframe['ema_9_rsi'] < 70)
            &
            (dataframe['ewo'] > 3)
            &
            (dataframe['volume'] > 0)
            , ['enter_long', 'enter_tag']
        ] = (1, 'golden cross')

        return dataframe

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

        dataframe.loc[
            qtpylib.crossed_below(dataframe['ema_9'], dataframe['ema_20'])
            &
            (dataframe['volume'] > 0)
            , ['exit_long', 'exit_tag']
        ] = (1, 'death cross')

        return dataframe