# source: https://raw.githubusercontent.com/ak4noir/freqtrade-strategies/2f02b6510af8dddbdfeabb09bdc8c5609de8ff53/user_data/strategies/mom.py # run on 15 minute # - close trade off the tail end of a vector candle # - don't sell if above 50 ema on 1h+ timeframe # - 100x for momentum plays (e.g. first green vector candle above the 50 ema) # - take profits on first candle close below 5 ema # - on 5 minute chart use 200 ema and 800 ema to determine if 5 minute timeframe will see a deep retracement or expansion import logging from datetime import datetime from enum import IntFlag from functools import reduce from typing import Optional import freqtrade.vendor.qtpylib.indicators as qtpylib import numpy as np import talib.abstract as ta from freqtrade.persistence import Trade from freqtrade.strategy import ( DecimalParameter, IStrategy, informative, stoploss_from_open, ) from pandas import DataFrame logger = logging.getLogger(__name__) class VectorColor(IntFlag): GREEN = 2 BLUE = 1 BULLISH = GREEN | BLUE RED = -2 VIOLET = -1 BEARISH = RED | VIOLET WHITE = 0 GRAY = WHITE VECTOR = GREEN | BLUE | RED | VIOLET NON_VECTOR = WHITE | GRAY def detect_pullback(df: DataFrame, periods=30, method="pct_outlier"): """ Pullback & Outlier Detection Know when a sudden move and possible reversal is coming Method 1: StDev Outlier (z-score) Method 2: Percent-Change Outlier (z-score) Method 3: Candle Open-Close %-Change outlier_threshold - Recommended: 2.0 - 3.0 df['pullback_flag']: 1 (Outlier Up) / -1 (Outlier Down) """ if method == "stdev_outlier": outlier_threshold = 2.0 df["dif"] = df["close"] - df["close"].shift(1) df["dif_squared_sum"] = (df["dif"] ** 2).rolling(window=periods + 1).sum() df["std"] = np.sqrt( (df["dif_squared_sum"] - df["dif"].shift(0) ** 2) / (periods - 1) ) df["z"] = df["dif"] / df["std"] df["pullback_flag"] = np.where(df["z"] >= outlier_threshold, 1, 0) df["pullback_flag"] = np.where( df["z"] <= -outlier_threshold, -1, df["pullback_flag"] ) if method == "pct_outlier": outlier_threshold = 2.0 df["pb_pct_change"] = df["close"].pct_change() df["pb_zscore"] = qtpylib.zscore(df, window=periods, col="pb_pct_change") df["pullback_flag"] = np.where(df["pb_zscore"] >= outlier_threshold, 1, 0) df["pullback_flag"] = np.where( df["pb_zscore"] <= -outlier_threshold, -1, df["pullback_flag"] ) if method == "candle_body": pullback_pct = 1.0 df["change"] = df["close"] - df["open"] df["pullback"] = (df["change"] / df["open"]) * 100 df["pullback_flag"] = np.where(df["pullback"] >= pullback_pct, 1, 0) df["pullback_flag"] = np.where( df["pullback"] <= -pullback_pct, -1, df["pullback_flag"] ) return df def detect_vector_candle(dataframe: DataFrame) -> DataFrame: """ determine if the current candle is a vector candle a vector candle represents a period of time of abnormal volume and can be an indicator of a breakout - if candle vol > 1.5 * mean volume then the candle is blue (bullish) or violet (bearish) - if candle vol > 2 * mean volume then the candle is green (bullish) or red (bearish) - otherwise it's white (bullish) or gray (bearish) - https://www.tradingview.com/script/8zgJTM9u-Traders-Reality-Lib/ - https://www.forexfactory.com/thread/540706-pvsra-price-volume-sr-analysis - https://www.forexfactory.com/attachment/file/1205218 """ df = dataframe.copy() avg_vol = df["volume"].rolling(10).mean() vol_high = df["volume"].rolling(10).max() vol_low = df["volume"].rolling(10).min() vol_spread = df.iloc[-10:]["volume"] * (vol_high - vol_low) vol_spread_high = vol_spread.max() # initialize as NON_VECTOR for when volume is 0 volume df.loc[:, "vector_color"] = VectorColor.NON_VECTOR.value # white & gray are not actually vectors df.loc[(df["close"] > df["open"]), "vector_color"] = VectorColor.WHITE.value df.loc[(df["close"] < df["open"]), "vector_color"] = VectorColor.GRAY.value df.loc[ (df["volume"] > avg_vol * 1.5) & (df["close"] > df["open"]), "vector_color", ] = VectorColor.BLUE.value df.loc[ (df["volume"] > avg_vol * 1.5) & (df["close"] < df["open"]), "vector_color", ] = VectorColor.VIOLET.value df.loc[ ((df["volume"] > avg_vol * 2) | (vol_spread > vol_spread_high)) & (df["close"] > df["open"]), "vector_color", ] = VectorColor.GREEN.value df.loc[ (df["volume"] > avg_vol * 2) & (df["close"] < df["open"]), "vector_color", ] = VectorColor.RED.value df["vector_count"] = df.rolling(10)["vector_color"].sum() return df def smi_trend( dataframe: DataFrame, k_length=9, d_length=3, smoothing_type="EMA", smoothing=10 ): """ Stochastic Momentum Index (SMI) Trend Indicator SMI > 0 and SMI > MA: (2) Bull SMI < 0 and SMI > MA: (1) Possible Bullish Reversal SMI > 0 and SMI < MA: (-1) Possible Bearish Reversal SMI < 0 and SMI < MA: (-2) Bear Returns: pandas.Series: New feature generated """ df = dataframe.copy() ll = df["low"].rolling(window=k_length).min() hh = df["high"].rolling(window=k_length).max() diff = hh - ll rdiff = df["close"] - (hh + ll) / 2 avgrel = rdiff.ewm(span=d_length).mean().ewm(span=d_length).mean() avgdiff = diff.ewm(span=d_length).mean().ewm(span=d_length).mean() smi = np.where(avgdiff != 0, (avgrel / (avgdiff / 2) * 100), 0) if smoothing_type == "SMA": smi_ma = ta.SMA(smi, timeperiod=smoothing) elif smoothing_type == "EMA": smi_ma = ta.EMA(smi, timeperiod=smoothing) elif smoothing_type == "WMA": smi_ma = ta.WMA(smi, timeperiod=smoothing) elif smoothing_type == "DEMA": smi_ma = ta.DEMA(smi, timeperiod=smoothing) elif smoothing_type == "TEMA": smi_ma = ta.TEMA(smi, timeperiod=smoothing) else: raise ValueError("Choose an MA Type: 'SMA', 'EMA', 'WMA', 'DEMA', 'TEMA'") conditions = [ (np.greater(smi, 0) & np.greater(smi, smi_ma)), # (2) Bull (np.less(smi, 0) & np.greater(smi, smi_ma)), # (1) Possible Bullish Reversal (np.greater(smi, 0) & np.less(smi, smi_ma)), # (-1) Possible Bearish Reversal (np.less(smi, 0) & np.less(smi, smi_ma)), # (-2) Bear ] smi_trend = np.select(conditions, [2, 1, -1, -2]) return smi, smi_ma, smi_trend def custom_stoploss_from_open( HSL: float, PF_1: float, SL_1: float, PF_2: float, SL_2: float, current_profit: float, **kwargs, ) -> float: # hard stoploss profit # HSL = self.pHSL.value # PF_1 = self.pPF_1.value # SL_1 = self.pSL_1.value # PF_2 = self.pPF_2.value # SL_2 = self.pSL_2.value # for profits between PF_1 and PF_2 the stoploss (sl_profit) used is linearly interpolated # between the values of SL_1 and SL_2. For all profits above PL_2 the sl_profit value # rises linearly with current profit, for profits below PF_1 the hard stoploss profit is used. if current_profit > PF_2: sl_profit = SL_2 + (current_profit - PF_2) elif current_profit > PF_1: sl_profit = SL_1 + ((current_profit - PF_1) * (SL_2 - SL_1) / (PF_2 - PF_1)) else: sl_profit = HSL # only for hyperopt invalid return if sl_profit >= current_profit: return -0.99 return stoploss_from_open(sl_profit, current_profit) class Github_ak4noir_freqtrade_strategies__mom__20240407_180126(IStrategy): # open trade on first green vector candle close above ema50 on 4h timeframe # exit long on first red vector candle close below ema50 on 4h timeframe # if profit > 0, take profit immediately after vector candle closes in direction of position (green for long, red for short) # if profit > 0, take profit when crossing ema5 on 4h timeframe # if profit < 0, exit when candle close below ema5 and above ema50 on 4h timeframe INTERFACE_VERSION = 3 """ PASTE OUTPUT FROM HYPEROPT HERE """ # Buy hyperspace params: buy_params: dict = {} # Sell hyperspace params: sell_params: dict = {} # ROI table: minimal_roi: dict = {} # Stoploss: stoploss = -0.10 # Trailing stop: trailing_stop = False trailing_stop_positive = 0.17251 trailing_stop_positive_offset = 0.2516 trailing_only_offset_is_reached = False """ END HYPEROPT """ # make sure these match or are not overridden in config use_exit_signal: bool = True exit_profit_only: bool = False exit_profit_offset: float = 0.01 ignore_roi_if_entry_signal: bool = True # use the custom_stoploss function use_custom_stoploss: bool = True timeframe: str = "5m" # run "populate_indicators()" only for new candle process_only_new_candles: bool = True # enable adjust_trade_position function to handle trade inbetween entry and exit position_adjustment_enable: bool = True # number of candles the strategy requires before producing valid signals (default: 72) startup_candle_count: int = 200 plot_config: dict = { "main_plot": { "ema5_1h": {"color": "#e5a50a", "type": "line"}, "ema50_1h": {"color": "#1a5fb4", "type": "line"}, "ema13_1h": {"color": "#26a269", "type": "line"}, }, "subplots": { "smi": { "smi_trend": {"color": "#e5a50a", "type": "line"}, "smi_trend_1h": {"color": "#1a5fb4", "type": "line"}, }, "pullback": { "pullback_flag": {"color": "#e5a50a", "type": "line"}, "pullback_flag_1h": {"color": "#1a5fb4", "type": "line"}, }, "vectors": { "vector_color": {"color": "#613583", "type": "bar"}, "vector_count": {"color": "#e364a5"}, }, }, } ## stoploss parameters # hard stoploss threshold pHSL = DecimalParameter(-0.200, 0.040, default=-0.10, decimals=3, space="sell") # profit threshold 1, trigger SL_1 is used pPF_1 = DecimalParameter(0.008, 0.020, default=0.016, decimals=3, space="sell") pSL_1 = DecimalParameter(0.008, 0.020, default=0.011, decimals=3, space="sell") # profit threshold 2, trigger SL_2 is used pPF_2 = DecimalParameter(0.040, 0.100, default=0.070, decimals=3, space="sell") pSL_2 = DecimalParameter(0.020, 0.070, default=0.030, decimals=3, space="sell") def adjust_trade_position( self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: float, max_stake: float, **kwargs, ) -> Optional[float]: df, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) last_candle = df.iloc[-1].squeeze() stake_amount: float = trade.stake_amount logger.debug(f"{trade.pair}: current_profit is {current_profit*100}%") logger.debug( f"{trade.pair}: using {stake_amount} of {max_stake} available {trade.stake_currency}" ) if current_profit > 0.005 and stake_amount > (2.5 * min_stake): count_of_exits = trade.nr_of_successful_exits take_profit = None if count_of_exits > 0: seconds_since_last_exit = ( datetime.now() - trade.select_filled_orders("sell")[-1].order_filled_date ).total_seconds() else: seconds_since_last_exit = -1 logger.debug( f"{trade.pair}: {seconds_since_last_exit} seconds since last exit" ) # take half when current_profit > 5% if current_profit > 0.05: take_profit = -(stake_amount / 2) # take some off the table if a potential bearish reversal elif last_candle["pullback_flag_1h"] == 1: take_profit = -(stake_amount / 4) # otherwise, take profit after a bullish vector candle close elif last_candle["vector_color"] == VectorColor.BULLISH: take_profit = -(stake_amount / 4) if take_profit: logger.info(f"{trade.pair}: adjusting position by {take_profit}") return take_profit else: # count_of_entries = trade.nr_of_successful_entries # add to position if green vector candle above 50ema_1h but below the 5ema_1h pass return None def custom_stake_amount( self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float, max_stake: float, **kwargs, ) -> float: if self.config["stake_amount"] == "unlimited": return proposed_stake return proposed_stake def custom_stoploss( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> float: # df, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) # last_candle = df.iloc[-1].squeeze() # if last_candle["ema5_1h"] < last_candle["ema50_1h"]: # return stoploss_from_absolute(last_candle["ema5_1h"], current_rate) # else: # return custom_stoploss_from_open( # self.pHSL.value, # self.pPF_1.value, # self.pSL_1.value, # self.pPF_2.value, # self.pSL_2.value, # current_profit, # ) return custom_stoploss_from_open( self.pHSL.value, self.pPF_1.value, self.pSL_1.value, self.pPF_2.value, self.pSL_2.value, current_profit, ) @informative("1h") def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # exponential moving averages dataframe["ema5"] = ta.EMA(dataframe, timeperiod=5) dataframe["ema13"] = ta.EMA(dataframe, timeperiod=13) dataframe["ema50"] = ta.EMA(dataframe, timeperiod=50) # pullback dataframe = detect_pullback(dataframe) # smi dataframe["smi"], dataframe["smi_ma"], dataframe["smi_trend"] = smi_trend( dataframe ) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # vector candle dataframe = detect_vector_candle(dataframe) # pullback dataframe = detect_pullback(dataframe) # smi dataframe["smi"], dataframe["smi_ma"], dataframe["smi_trend"] = smi_trend( dataframe ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [] dataframe.loc[:, "enter_tag"] = "" # enter long position on first or second green vector candle confirmation in the count with a close above ema50_1h # and no signs of reversal green_vector_breakout = ( (dataframe["vector_color"] == VectorColor.GREEN) & (dataframe["open"] > dataframe["ema50_1h"]) & (dataframe["close"] > dataframe["ema50_1h"]) & (dataframe["vector_count"] > 0) & (dataframe["vector_count"] <= 2) & (dataframe["pullback_flag_1h"] < 1) & (dataframe["smi_trend_1h"] == 1) & (dataframe["volume"] > 0) ) dataframe.loc[ green_vector_breakout, "enter_tag" ] += "bullish_green_vector_confirmation" conditions.append(green_vector_breakout) # enter long position on cross above ema50_1h if smi_trend_1h is bullish # and no signs of reversal smi_bullish_breakout = ( (dataframe["smi_trend_1h"] == 2) & (dataframe["ema13_1h"] < dataframe["ema50_1h"]) & (qtpylib.crossed_above(dataframe["close"], dataframe["ema50_1h"])) & (dataframe["pullback_flag_1h"] < 1) & (dataframe["volume"] > 0) ) dataframe.loc[smi_bullish_breakout, "enter_tag"] += "+bullish_smi" conditions.append(smi_bullish_breakout) # enter long position on cross above ema50_1h if smi_trend_1h is bullish # and no signs of reversal smi_bullish_reversal = ( (dataframe["smi_trend_1h"] == 1) & (dataframe["ema13_1h"] < dataframe["ema50_1h"]) & (qtpylib.crossed_above(dataframe["close"], dataframe["ema50_1h"])) & (dataframe["pullback_flag_1h"] < 1) & (dataframe["volume"] > 0) ) dataframe.loc[smi_bullish_reversal, "enter_tag"] += "+bullish_smi_reversal" if conditions: dataframe.loc[reduce(lambda x, y: x | y, conditions), "enter_long"] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [] dataframe.loc[:, "exit_tag"] = "" # exit long if red vector candle confirmation below ema50_1h # when smi is bearish red_vector_breakdown = ( (dataframe["vector_color"] == VectorColor.RED) & (dataframe["vector_count"] <= -4) & (dataframe["volume"] > 0) & (dataframe["smi_trend_1h"] < 0) & (dataframe["close"] < dataframe["ema50_1h"]) & (dataframe["open"] < dataframe["ema50_1h"]) ) dataframe.loc[ red_vector_breakdown, "exit_tag" ] += "bearish_red_vector_confirmation" conditions.append(red_vector_breakdown) if conditions: dataframe.loc[reduce(lambda x, y: x | y, conditions), "exit_long"] = 1 return dataframe @property def protections(self): return [ {"method": "CooldownPeriod", "stop_duration_candles": 24}, { # if a stoploss exit is triggered that results in a loss in a 1 hour period, # stop trading that pair for 4 hours "method": "StoplossGuard", "lookback_period_candles": 24, # 1h "trade_limit": 1, "stop_duration_candles": 24 * 4, # 4h "required_profit": 0.0, "only_per_pair": True, "only_per_side": False, }, ]