# source: https://raw.githubusercontent.com/remiotore/freqtrade/44beaeb6a420cd8e9f2e4ea93e11d6cfa192ee03/strategies/F_FSupertrendStrategy_5m.py """ Supertrend strategy: * Description: Generate a 3 supertrend indicators for 'buy' strategies & 3 supertrend indicators for 'sell' strategies Buys if the 3 'buy' indicators are 'up' Sells if the 3 'sell' indicators are 'down' * Author: @juankysoriano (Juan Carlos Soriano) * github: https://github.com/juankysoriano/ *** NOTE: This Supertrend strategy is just one of many possible strategies using `Supertrend` as indicator. It should on any case used at your own risk. It comes with at least a couple of caveats: 1. The implementation for the `supertrend` indicator is based on the following discussion: https://github.com/freqtrade/freqtrade-strategies/issues/30 . Concretelly https://github.com/freqtrade/freqtrade-strategies/issues/30#issuecomment-853042401 2. The implementation for `supertrend` on this strategy is not validated; meaning this that is not proven to match the results by the paper where it was originally introduced or any other trusted academic resources """ import logging import numpy as np import talib.abstract as ta from freqtrade.strategy import IntParameter, IStrategy from freqtrade.persistence import Trade from datetime import datetime from numpy.lib import math from pandas import DataFrame import freqtrade.vendor.qtpylib.indicators as qtpylib import pandas as pd from pandas import Series from collections import deque from typing import Tuple class Github_remiotore_freqtrade__F_FSupertrendStrategy_5m__20260111_210550(IStrategy): # Buy params, Sell params, ROI, Stoploss and Trailing Stop are values generated by 'freqtrade hyperopt --strategy Supertrend --hyperopt-loss ShortTradeDurHyperOptLoss --timerange=20210101- --timeframe=1h --spaces all' # It's encourage you find the values that better suites your needs and risk management strategies INTERFACE_VERSION: int = 3 # Can this strategy go short? can_short: bool = True # Buy hyperspace params: buy_params = { "buy_m1": 4, "buy_m2": 7, "buy_m3": 1, "buy_p1": 8, "buy_p2": 9, "buy_p3": 8, } # Sell hyperspace params: sell_params = { "sell_m1": 1, "sell_m2": 3, "sell_m3": 6, "sell_p1": 16, "sell_p2": 18, "sell_p3": 18, } # ROI table: # minimal_roi = {"0": 0.05, "30": 0.125, "45": 0.03, "60": 0.015} minimal_roi = {"0": 0.01} # minimal_roi = {"0": 1} # Stoploss: stoploss = -0.08 # Trailing stop: trailing_stop = True trailing_stop_positive = 0.05 trailing_stop_positive_offset = 0.1 trailing_only_offset_is_reached = False timeframe = "5m" startup_candle_count = 10 buy_m1 = IntParameter(1, 7, default=1) buy_m2 = IntParameter(1, 7, default=3) buy_m3 = IntParameter(1, 7, default=4) buy_p1 = IntParameter(7, 21, default=14) buy_p2 = IntParameter(7, 21, default=10) buy_p3 = IntParameter(7, 21, default=10) sell_m1 = IntParameter(1, 7, default=1) sell_m2 = IntParameter(1, 7, default=3) sell_m3 = IntParameter(1, 7, default=4) sell_p1 = IntParameter(7, 21, default=14) sell_p2 = IntParameter(7, 21, default=10) sell_p3 = IntParameter(7, 21, default=10) custom_info = {} def bot_loop_start(self, current_time: datetime, **kwargs) -> None: for pair in list(self.custom_info): if "unlock_me" in self.custom_info[pair]: message = f"Found reverse position signal - unlocking {pair}" self.dp.send_msg(message) self.unlock_pair(pair) del self.custom_info[pair] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: for multiplier in self.buy_m1.range: for period in self.buy_p1.range: dataframe[f"supertrend_1_buy_{multiplier}_{period}"] = self.supertrend( dataframe, multiplier, period )["STX"] for multiplier in self.buy_m2.range: for period in self.buy_p2.range: dataframe[f"supertrend_2_buy_{multiplier}_{period}"] = self.supertrend( dataframe, multiplier, period )["STX"] for multiplier in self.buy_m3.range: for period in self.buy_p3.range: dataframe[f"supertrend_3_buy_{multiplier}_{period}"] = self.supertrend( dataframe, multiplier, period )["STX"] for multiplier in self.sell_m1.range: for period in self.sell_p1.range: dataframe[f"supertrend_1_sell_{multiplier}_{period}"] = self.supertrend( dataframe, multiplier, period )["STX"] for multiplier in self.sell_m2.range: for period in self.sell_p2.range: dataframe[f"supertrend_2_sell_{multiplier}_{period}"] = self.supertrend( dataframe, multiplier, period )["STX"] for multiplier in self.sell_m3.range: for period in self.sell_p3.range: dataframe[f"supertrend_3_sell_{multiplier}_{period}"] = self.supertrend( dataframe, multiplier, period )["STX"] informative = dataframe # Momentum Indicators # ------------------------------------ # RSI informative['rsi'] = ta.RSI(informative) # Stochastic Slow informative['stoch'] = ta.STOCH(informative)['slowk'] # ROC informative['roc'] = ta.ROC(informative) # Ultimate Oscillator informative['uo'] = ta.ULTOSC(informative) # Awesome Oscillator informative['ao'] = qtpylib.awesome_oscillator(informative) # MACD informative['macd'] = ta.MACD(informative)['macd'] # Commodity Channel Index informative['cci'] = ta.CCI(informative) # CMF informative['cmf'] = chaikin_money_flow(informative, 20) # OBV informative['obv'] = ta.OBV(informative) # MFI informative['mfi'] = ta.MFI(informative) # ADX informative['adx'] = ta.ADX(informative) # ATR informative['atr'] = qtpylib.atr(informative, window=14, exp=False) # Keltner Channel # keltner = qtpylib.keltner_channel(dataframe, window=20, atrs=1) keltner = emaKeltner(informative) informative["kc_upperband"] = keltner["upper"] informative["kc_middleband"] = keltner["mid"] informative["kc_lowerband"] = keltner["lower"] # Bollinger Bands bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(informative), window=20, stds=2) informative['bollinger_upperband'] = bollinger['upper'] informative['bollinger_lowerband'] = bollinger['lower'] # EMA - Exponential Moving Average informative['ema9'] = ta.EMA(informative, timeperiod=9) informative['ema20'] = ta.EMA(informative, timeperiod=20) informative['ema50'] = ta.EMA(informative, timeperiod=50) informative['ema200'] = ta.EMA(informative, timeperiod=200) pivots = pivot_points(informative) informative['pivot_lows'] = pivots['pivot_lows'] informative['pivot_highs'] = pivots['pivot_highs'] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( dataframe[f"supertrend_1_buy_{self.buy_m1.value}_{self.buy_p1.value}"] == "up" ) & ( dataframe[f"supertrend_2_buy_{self.buy_m2.value}_{self.buy_p2.value}"] == "up" ) & ( dataframe[f"supertrend_3_buy_{self.buy_m3.value}_{self.buy_p3.value}"] == "up" ) & ( # The three indicators are 'up' for the current candle dataframe["volume"] > 0 ), "enter_long", ] = 1 dataframe.loc[ ( dataframe[ f"supertrend_1_sell_{self.sell_m1.value}_{self.sell_p1.value}" ] == "down" ) & ( dataframe[ f"supertrend_2_sell_{self.sell_m2.value}_{self.sell_p2.value}" ] == "down" ) & ( dataframe[ f"supertrend_3_sell_{self.sell_m3.value}_{self.sell_p3.value}" ] == "down" ) & ( # The three indicators are 'down' for the current candle dataframe["volume"] > 0 ), "enter_short", ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( dataframe[ f"supertrend_2_sell_{self.sell_m2.value}_{self.sell_p2.value}" ] == "down" ), "exit_long", ] = 1 dataframe.loc[ ( dataframe[f"supertrend_2_buy_{self.buy_m2.value}_{self.buy_p2.value}"] == "up" ), "exit_short", ] = 1 return dataframe def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime, **kwargs) -> bool: dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() if trade.is_short: if last_candle['enter_long'] == 1: if not pair in self.custom_info: self.custom_info[pair] = {} self.custom_info[pair]["unlock_me"] = True else: if last_candle['enter_short'] == 1: if not pair in self.custom_info: self.custom_info[pair] = {} self.custom_info[pair]["unlock_me"] = True return True """ Supertrend Indicator; adapted for freqtrade from: https://github.com/freqtrade/freqtrade-strategies/issues/30 """ def supertrend(self, dataframe: DataFrame, multiplier, period): df = dataframe.copy() df["TR"] = ta.TRANGE(df) df["ATR"] = ta.SMA(df["TR"], period) st = "ST_" + str(period) + "_" + str(multiplier) stx = "STX_" + str(period) + "_" + str(multiplier) # Compute basic upper and lower bands df["basic_ub"] = (df["high"] + df["low"]) / 2 + multiplier * df["ATR"] df["basic_lb"] = (df["high"] + df["low"]) / 2 - multiplier * df["ATR"] # Compute final upper and lower bands df["final_ub"] = 0.00 df["final_lb"] = 0.00 for i in range(period, len(df)): df["final_ub"].iat[i] = ( df["basic_ub"].iat[i] if df["basic_ub"].iat[i] < df["final_ub"].iat[i - 1] or df["close"].iat[i - 1] > df["final_ub"].iat[i - 1] else df["final_ub"].iat[i - 1] ) df["final_lb"].iat[i] = ( df["basic_lb"].iat[i] if df["basic_lb"].iat[i] > df["final_lb"].iat[i - 1] or df["close"].iat[i - 1] < df["final_lb"].iat[i - 1] else df["final_lb"].iat[i - 1] ) # Set the Supertrend value df[st] = 0.00 for i in range(period, len(df)): df[st].iat[i] = ( df["final_ub"].iat[i] if df[st].iat[i - 1] == df["final_ub"].iat[i - 1] and df["close"].iat[i] <= df["final_ub"].iat[i] else df["final_lb"].iat[i] if df[st].iat[i - 1] == df["final_ub"].iat[i - 1] and df["close"].iat[i] > df["final_ub"].iat[i] else df["final_lb"].iat[i] if df[st].iat[i - 1] == df["final_lb"].iat[i - 1] and df["close"].iat[i] >= df["final_lb"].iat[i] else df["final_ub"].iat[i] if df[st].iat[i - 1] == df["final_lb"].iat[i - 1] and df["close"].iat[i] < df["final_lb"].iat[i] else 0.00 ) # Mark the trend direction up/down df[stx] = np.where( (df[st] > 0.00), np.where((df["close"] < df[st]), "down", "up"), np.NaN ) # Remove basic and final bands from the columns df.drop(["basic_ub", "basic_lb", "final_ub", "final_lb"], inplace=True, axis=1) df.fillna(0, inplace=True) return DataFrame(index=df.index, data={"ST": df[st], "STX": df[stx]}) def leverage(self, pair: str, current_time, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag, side: str, **kwargs) -> float: return 5.0 from enum import Enum class PivotSource(Enum): HighLow = 0 Close = 1 def pivot_points(dataframe: DataFrame, window: int = 5, pivot_source: PivotSource = PivotSource.Close) -> DataFrame: high_source = None low_source = None if pivot_source == PivotSource.Close: high_source = 'close' low_source = 'close' elif pivot_source == PivotSource.HighLow: high_source = 'high' low_source = 'low' pivot_points_lows = np.empty(len(dataframe['close'])) * np.nan pivot_points_highs = np.empty(len(dataframe['close'])) * np.nan last_values = deque() # find pivot points for index, row in enumerate(dataframe.itertuples(index=True, name='Pandas')): last_values.append(row) if len(last_values) >= window * 2 + 1: current_value = last_values[window] is_greater = True is_less = True for window_index in range(0, window): left = last_values[window_index] right = last_values[2 * window - window_index] local_is_greater, local_is_less = check_if_pivot_is_greater_or_less(current_value, high_source, low_source, left, right) is_greater &= local_is_greater is_less &= local_is_less if is_greater: pivot_points_highs[index - window] = getattr(current_value, high_source) if is_less: pivot_points_lows[index - window] = getattr(current_value, low_source) last_values.popleft() # find last one if len(last_values) >= window + 2: current_value = last_values[-2] is_greater = True is_less = True for window_index in range(0, window): left = last_values[-2 - window_index - 1] right = last_values[-1] local_is_greater, local_is_less = check_if_pivot_is_greater_or_less(current_value, high_source, low_source, left, right) is_greater &= local_is_greater is_less &= local_is_less if is_greater: pivot_points_highs[index - 1] = getattr(current_value, high_source) if is_less: pivot_points_lows[index - 1] = getattr(current_value, low_source) return pd.DataFrame(index=dataframe.index, data={ 'pivot_lows': pivot_points_lows, 'pivot_highs': pivot_points_highs }) def check_if_pivot_is_greater_or_less(current_value, high_source: str, low_source: str, left, right) -> Tuple[bool, bool]: is_greater = True is_less = True if (getattr(current_value, high_source) < getattr(left, high_source) or getattr(current_value, high_source) < getattr(right, high_source)): is_greater = False if (getattr(current_value, low_source) > getattr(left, low_source) or getattr(current_value, low_source) > getattr(right, low_source)): is_less = False return (is_greater, is_less) def emaKeltner(dataframe): keltner = {} atr = qtpylib.atr(dataframe, window=10) ema20 = ta.EMA(dataframe, timeperiod=20) keltner['upper'] = ema20 + atr keltner['mid'] = ema20 keltner['lower'] = ema20 - atr return keltner def chaikin_money_flow(dataframe, n=20, fillna=False) -> Series: """Chaikin Money Flow (CMF) It measures the amount of Money Flow Volume over a specific period. http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:chaikin_money_flow_cmf Args: dataframe(pandas.Dataframe): dataframe containing ohlcv n(int): n period. fillna(bool): if True, fill nan values. Returns: pandas.Series: New feature generated. """ df = dataframe.copy() mfv = ((df['close'] - df['low']) - (df['high'] - df['close'])) / (df['high'] - df['low']) mfv = mfv.fillna(0.0) # float division by zero mfv *= df['volume'] cmf = (mfv.rolling(n, min_periods=0).sum() / df['volume'].rolling(n, min_periods=0).sum()) if fillna: cmf = cmf.replace([np.inf, -np.inf], np.nan).fillna(0) return Series(cmf, name='cmf')