# source: https://raw.githubusercontent.com/xaemiphor/ft/e758f416065c3b69b29138a49a070d1803748bf1/strategies/sweep1_strategy.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these imports --- import numpy as np import pandas as pd from datetime import datetime, timedelta, timezone from pandas import DataFrame from typing import Optional, Union from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, # @informative decorator # Hyperopt Parameters BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, # timeframe helpers timeframe_to_minutes, timeframe_to_next_date, timeframe_to_prev_date, # Strategy helper functions merge_informative_pair, stoploss_from_absolute, stoploss_from_open, ) # -------------------------------- # Add your lib to import here import talib.abstract as ta from technical import qtpylib # This class is a sample. Feel free to customize it. class Github_xaemiphor_ft__sweep1_strategy__20241220_183853(IStrategy): """ This is a sample strategy to inspire you. More information in https://www.freqtrade.io/en/latest/strategy-customization/ You can: :return: a Dataframe with all mandatory indicators for the strategies - Rename the class name (Do not forget to update class_name) - Add any methods you want to build your strategy - Add any lib you need to build your strategy You must keep: - the lib in the section "Do not remove these libs" - the methods: populate_indicators, populate_entry_trend, populate_exit_trend You should keep: - timeframe, minimal_roi, stoploss, trailing_* """ # 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 # Can this strategy go short? can_short: bool = False # Minimal ROI designed for the strategy. # This attribute will be overridden if the config file contains "minimal_roi". minimal_roi = { "60": 0.06, "0": 0.10, } # Optimal stoploss designed for the strategy. # This attribute will be overridden if the config file contains "stoploss". stoploss = -0.25 use_custom_stoploss = True # Example specific variables max_entry_position_adjustment = 10 # This number is explained a bit further down max_dca_multiplier = 5.5 # Trailing stoploss trailing_stop = False # trailing_only_offset_is_reached = False # trailing_stop_positive = 0.01 # trailing_stop_positive_offset = 0.0 # Disabled / not configured # Optimal timeframe for the strategy. timeframe = "5m" # 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 = True exit_profit_offset = 0.01 ignore_roi_if_entry_signal = True # Hyperoptable parameters buy_rsi = IntParameter(low=1, high=50, default=30, space="buy", optimize=True, load=True) sell_rsi = IntParameter(low=50, high=100, default=75, space="sell", optimize=True, load=True) buy_volatility_threshold = DecimalParameter(low=0, high=1, default=0.02, decimals=3, space="buy", optimize=True, load=True) sell_volatility_threshold = DecimalParameter(low=0, high=1, default=0.02, decimals=3, space="sell", optimize=True, load=True) stoploss_volatility_threshold = DecimalParameter(low=0, high=1, default=0.02, decimals=3, space="stoploss", optimize=True, load=True) stoploss_buffer = DecimalParameter(low=0, high=1, default=0.9, decimals=1, space="stoploss", optimize=True, load=True) # Number of candles the strategy requires before producing valid signals startup_candle_count: int = 200 # 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"} plot_config = { "main_plot": { "wbb_upperband": { "color": "#008af4", "type": "line", "fill_to": "wbb_lowerband" }, "wbb_lowerband": { "color": "#008af4", "type": "line" }, "wbb_middleband": { "color": "#008af4", "type": "line" }, "tema": { "color": "#ffffff", "type": "line" } }, "subplots": { "rsi": { "rsi": { "color": "#e01b24", "type": "line" } }, "width": { "wbb_width": { "color": "#7f1f68" } } } } def informative_pairs(self): """ Define additional, informative pair/interval combinations to be cached from the exchange. These pair/interval combinations are non-tradeable, unless they are part of the whitelist as well. For more information, please consult the documentation :return: List of tuples in the format (pair, interval) Sample: return [("ETH/USDT", "5m"), ("BTC/USDT", "15m"), ] """ return [] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Adds several different TA indicators to the given DataFrame Performance Note: For the best performance be frugal on the number of indicators you are using. Let uncomment only the indicator you are using in your strategies or your hyperopt configuration, otherwise you will waste your memory and CPU usage. :param dataframe: Dataframe with data from the exchange :param metadata: Additional information, like the currently traded pair :return: a Dataframe with all mandatory indicators for the strategies """ # Momentum Indicators # ------------------------------------ # ADX dataframe["adx"] = ta.ADX(dataframe) # # Plus Directional Indicator / Movement # dataframe['plus_dm'] = ta.PLUS_DM(dataframe) # dataframe['plus_di'] = ta.PLUS_DI(dataframe) # # Minus Directional Indicator / Movement # dataframe['minus_dm'] = ta.MINUS_DM(dataframe) # dataframe['minus_di'] = ta.MINUS_DI(dataframe) # # Aroon, Aroon Oscillator # aroon = ta.AROON(dataframe) # dataframe['aroonup'] = aroon['aroonup'] # dataframe['aroondown'] = aroon['aroondown'] # dataframe['aroonosc'] = ta.AROONOSC(dataframe) # # Awesome Oscillator # dataframe['ao'] = qtpylib.awesome_oscillator(dataframe) # # Keltner Channel # keltner = qtpylib.keltner_channel(dataframe) # dataframe["kc_upperband"] = keltner["upper"] # dataframe["kc_lowerband"] = keltner["lower"] # dataframe["kc_middleband"] = keltner["mid"] # dataframe["kc_percent"] = ( # (dataframe["close"] - dataframe["kc_lowerband"]) / # (dataframe["kc_upperband"] - dataframe["kc_lowerband"]) # ) # dataframe["kc_width"] = ( # (dataframe["kc_upperband"] - dataframe["kc_lowerband"]) / dataframe["kc_middleband"] # ) # # Ultimate Oscillator # dataframe['uo'] = ta.ULTOSC(dataframe) # # Commodity Channel Index: values [Oversold:-100, Overbought:100] # dataframe['cci'] = ta.CCI(dataframe) # RSI dataframe["rsi"] = ta.RSI(dataframe) # # Inverse Fisher transform on RSI: values [-1.0, 1.0] (https://goo.gl/2JGGoy) # rsi = 0.1 * (dataframe['rsi'] - 50) # dataframe['fisher_rsi'] = (np.exp(2 * rsi) - 1) / (np.exp(2 * rsi) + 1) # # Inverse Fisher transform on RSI normalized: values [0.0, 100.0] (https://goo.gl/2JGGoy) # dataframe['fisher_rsi_norma'] = 50 * (dataframe['fisher_rsi'] + 1) # # Stochastic Slow # stoch = ta.STOCH(dataframe) # dataframe['slowd'] = stoch['slowd'] # dataframe['slowk'] = stoch['slowk'] # Stochastic Fast stoch_fast = ta.STOCHF(dataframe) dataframe["fastd"] = stoch_fast["fastd"] dataframe["fastk"] = stoch_fast["fastk"] # # Stochastic RSI # Please read https://github.com/freqtrade/freqtrade/issues/2961 before using this. # STOCHRSI is NOT aligned with tradingview, which may result in non-expected results. # stoch_rsi = ta.STOCHRSI(dataframe) # dataframe['fastd_rsi'] = stoch_rsi['fastd'] # dataframe['fastk_rsi'] = stoch_rsi['fastk'] # MACD macd = ta.MACD(dataframe) dataframe["macd"] = macd["macd"] dataframe["macdsignal"] = macd["macdsignal"] dataframe["macdhist"] = macd["macdhist"] # MFI dataframe["mfi"] = ta.MFI(dataframe) # # ROC # dataframe['roc'] = ta.ROC(dataframe) # Overlap Studies # ------------------------------------ # Bollinger Bands bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2) dataframe["bb_lowerband"] = bollinger["lower"] dataframe["bb_middleband"] = bollinger["mid"] dataframe["bb_upperband"] = bollinger["upper"] dataframe["bb_percent"] = (dataframe["close"] - dataframe["bb_lowerband"]) / ( dataframe["bb_upperband"] - dataframe["bb_lowerband"] ) dataframe["bb_width"] = (dataframe["bb_upperband"] - dataframe["bb_lowerband"]) / dataframe[ "bb_middleband" ] # Bollinger Bands - Weighted (EMA based instead of SMA) weighted_bollinger = qtpylib.weighted_bollinger_bands( qtpylib.typical_price(dataframe), window=20, stds=2 ) dataframe["wbb_upperband"] = weighted_bollinger["upper"] dataframe["wbb_lowerband"] = weighted_bollinger["lower"] dataframe["wbb_middleband"] = weighted_bollinger["mid"] dataframe["wbb_percent"] = ( (dataframe["close"] - dataframe["wbb_lowerband"]) / (dataframe["wbb_upperband"] - dataframe["wbb_lowerband"]) ) dataframe["wbb_width"] = ( (dataframe["wbb_upperband"] - dataframe["wbb_lowerband"]) / dataframe["wbb_middleband"] ) # # EMA - Exponential Moving Average # dataframe['ema3'] = ta.EMA(dataframe, timeperiod=3) dataframe['ema5'] = ta.EMA(dataframe, timeperiod=5) dataframe['ema10'] = ta.EMA(dataframe, timeperiod=10) # dataframe['ema21'] = ta.EMA(dataframe, timeperiod=21) # dataframe['ema50'] = ta.EMA(dataframe, timeperiod=50) # dataframe['ema100'] = ta.EMA(dataframe, timeperiod=100) # # SMA - Simple Moving Average # dataframe['sma3'] = ta.SMA(dataframe, timeperiod=3) # dataframe['sma5'] = ta.SMA(dataframe, timeperiod=5) # dataframe['sma10'] = ta.SMA(dataframe, timeperiod=10) # dataframe['sma21'] = ta.SMA(dataframe, timeperiod=21) # dataframe['sma50'] = ta.SMA(dataframe, timeperiod=50) # dataframe['sma100'] = ta.SMA(dataframe, timeperiod=100) # Parabolic SAR dataframe["sar"] = ta.SAR(dataframe) # TEMA - Triple Exponential Moving Average dataframe["tema"] = ta.TEMA(dataframe, timeperiod=9) # Cycle Indicator # ------------------------------------ # Hilbert Transform Indicator - SineWave hilbert = ta.HT_SINE(dataframe) dataframe["htsine"] = hilbert["sine"] dataframe["htleadsine"] = hilbert["leadsine"] # Pattern Recognition - Bullish candlestick patterns # ------------------------------------ # # Hammer: values [0, 100] # dataframe['CDLHAMMER'] = ta.CDLHAMMER(dataframe) # # Inverted Hammer: values [0, 100] # dataframe['CDLINVERTEDHAMMER'] = ta.CDLINVERTEDHAMMER(dataframe) # # Dragonfly Doji: values [0, 100] # dataframe['CDLDRAGONFLYDOJI'] = ta.CDLDRAGONFLYDOJI(dataframe) # # Piercing Line: values [0, 100] # dataframe['CDLPIERCING'] = ta.CDLPIERCING(dataframe) # values [0, 100] # # Morningstar: values [0, 100] # dataframe['CDLMORNINGSTAR'] = ta.CDLMORNINGSTAR(dataframe) # values [0, 100] # # Three White Soldiers: values [0, 100] # dataframe['CDL3WHITESOLDIERS'] = ta.CDL3WHITESOLDIERS(dataframe) # values [0, 100] # Pattern Recognition - Bearish candlestick patterns # ------------------------------------ # # Hanging Man: values [0, 100] # dataframe['CDLHANGINGMAN'] = ta.CDLHANGINGMAN(dataframe) # # Shooting Star: values [0, 100] # dataframe['CDLSHOOTINGSTAR'] = ta.CDLSHOOTINGSTAR(dataframe) # # Gravestone Doji: values [0, 100] # dataframe['CDLGRAVESTONEDOJI'] = ta.CDLGRAVESTONEDOJI(dataframe) # # Dark Cloud Cover: values [0, 100] # dataframe['CDLDARKCLOUDCOVER'] = ta.CDLDARKCLOUDCOVER(dataframe) # # Evening Doji Star: values [0, 100] # dataframe['CDLEVENINGDOJISTAR'] = ta.CDLEVENINGDOJISTAR(dataframe) # # Evening Star: values [0, 100] # dataframe['CDLEVENINGSTAR'] = ta.CDLEVENINGSTAR(dataframe) # Pattern Recognition - Bullish/Bearish candlestick patterns # ------------------------------------ # # Three Line Strike: values [0, -100, 100] # dataframe['CDL3LINESTRIKE'] = ta.CDL3LINESTRIKE(dataframe) # # Spinning Top: values [0, -100, 100] # dataframe['CDLSPINNINGTOP'] = ta.CDLSPINNINGTOP(dataframe) # values [0, -100, 100] # # Engulfing: values [0, -100, 100] # dataframe['CDLENGULFING'] = ta.CDLENGULFING(dataframe) # values [0, -100, 100] # # Harami: values [0, -100, 100] # dataframe['CDLHARAMI'] = ta.CDLHARAMI(dataframe) # values [0, -100, 100] # # Three Outside Up/Down: values [0, -100, 100] # dataframe['CDL3OUTSIDE'] = ta.CDL3OUTSIDE(dataframe) # values [0, -100, 100] # # Three Inside Up/Down: values [0, -100, 100] # dataframe['CDL3INSIDE'] = ta.CDL3INSIDE(dataframe) # values [0, -100, 100] # # Chart type # # ------------------------------------ # # Heikin Ashi Strategy # heikinashi = qtpylib.heikinashi(dataframe) # dataframe['ha_open'] = heikinashi['open'] # dataframe['ha_close'] = heikinashi['close'] # dataframe['ha_high'] = heikinashi['high'] # dataframe['ha_low'] = heikinashi['low'] # Retrieve best bid and best ask from the orderbook # ------------------------------------ """ # first check if dataprovider is available if self.dp: if self.dp.runmode.value in ('live', 'dry_run'): ob = self.dp.orderbook(metadata['pair'], 1) dataframe['best_bid'] = ob['bids'][0][0] dataframe['best_ask'] = ob['asks'][0][0] """ return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the entry signal for the given dataframe :param dataframe: DataFrame :param metadata: Additional information, like the currently traded pair :return: DataFrame with entry columns populated """ dataframe.loc[ ( (dataframe["volume"] > 0) # Make sure Volume is not 0 & (dataframe["rsi"] < self.buy_rsi.value) #& (qtpylib.crossed_above(dataframe["rsi"], self.buy_rsi.value)) & (dataframe["high"] < dataframe["wbb_middleband"]) #& (dataframe["tema"] > dataframe["tema"].shift(1)) # Guard: tema is raising #& (dataframe["tema"].shift(2) < dataframe["tema"].shift(1)) # Guard: tema is rebounding ), "enter_long", ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the exit signal for the given dataframe :param dataframe: DataFrame :param metadata: Additional information, like the currently traded pair :return: DataFrame with exit columns populated """ dataframe.loc[ ( (dataframe["volume"] > 0) # Make sure Volume is not 0 & ((dataframe["rsi"] > self.sell_rsi.value) | (dataframe["tema"] > dataframe["wbb_upperband"])) #& (qtpylib.crossed_above(dataframe["rsi"], self.sell_rsi.value)) & (dataframe["low"] > dataframe["wbb_middleband"]) #& (dataframe["tema"] < dataframe["tema"].shift(1)) # Guard: tema is falling #& (dataframe["tema"].shift(2) > dataframe["tema"].shift(1)) # Guard: tema is rebounding ), "exit_long", ] = 1 return dataframe def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs) -> float | None: """ Custom stoploss logic, returning the new distance relative to current_rate (as ratio). e.g. returning -0.05 would create a stoploss 5% below current_rate. The custom stoploss can never be below self.stoploss, which serves as a hard maximum loss. For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/ When not implemented by a strategy, returns the initial stoploss value. Only called when use_custom_stoploss is set to True. :param pair: Pair that's currently analyzed :param trade: trade object. :param current_time: datetime object, containing the current datetime :param current_rate: Rate, calculated based on pricing settings in exit_pricing. :param current_profit: Current profit (as ratio), calculated based on current_rate. :param after_fill: True if the stoploss is called after the order was filled. :param **kwargs: Ensure to keep this here so updates to this won't break your strategy. :return float: New stoploss value, relative to the current_rate """ dataframes, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) dataframe = dataframes.iloc[-1] # evaluate highest to lowest, so that highest possible stop is used if current_profit > 0.40: return stoploss_from_open(0.25, current_profit, is_short=trade.is_short, leverage=trade.leverage) elif current_profit > 0.25: return stoploss_from_open(0.15, current_profit, is_short=trade.is_short, leverage=trade.leverage) elif current_profit > 0.20: return stoploss_from_open(0.07, current_profit, is_short=trade.is_short, leverage=trade.leverage) elif current_profit > 0.10: return stoploss_from_open(0.03, current_profit, is_short=trade.is_short, leverage=trade.leverage) elif (current_profit < 0) & (dataframe["wbb_width"] > self.stoploss_volatility_threshold.value): return stoploss_from_absolute(dataframe["wbb_lowerband"] - (dataframe["wbb_upperband"] - dataframe["wbb_lowerband"]), current_rate, is_short=trade.is_short, leverage=trade.leverage) #if (current_profit > 0) & (current_rate > dataframe["wbb_middleband"]) & (dataframe["wbb_width"] > self.stoploss_volatility_threshold.value): # return stoploss_from_absolute(dataframe["wbb_middleband"], current_rate, is_short=trade.is_short) #elif current_rate > dataframe["wbb_lowerband"]: # #return stoploss_from_absolute(dataframe["wbb_lowerband"], current_rate, is_short=trade.is_short) # #return stoploss_from_absolute(dataframe["wbb_lowerband"] * self.stoploss_buffer.value, current_rate, is_short=trade.is_short) # return stoploss_from_absolute(dataframe["wbb_lowerband"] - (dataframe["wbb_upperband"] - dataframe["wbb_lowerband"]), current_rate, is_short=trade.is_short) return None # This is called when placing the initial order (opening trade) def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float | None, max_stake: float, leverage: float, entry_tag: str | None, side: str, **kwargs) -> float: # We need to leave most of the funds for possible further DCA orders # This also applies to fixed stakes return proposed_stake / self.max_dca_multiplier def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: float | None, max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs ) -> float | None | tuple[float | None, str | None]: """ Custom trade adjustment logic, returning the stake amount that a trade should be increased or decreased. This means extra entry or exit orders with additional fees. Only called when `position_adjustment_enable` is set to True. For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/ When not implemented by a strategy, returns None :param trade: trade object. :param current_time: datetime object, containing the current datetime :param current_rate: Current entry rate (same as current_entry_profit) :param current_profit: Current profit (as ratio), calculated based on current_rate (same as current_entry_profit). :param min_stake: Minimal stake size allowed by exchange (for both entries and exits) :param max_stake: Maximum stake allowed (either through balance, or by exchange limits). :param current_entry_rate: Current rate using entry pricing. :param current_exit_rate: Current rate using exit pricing. :param current_entry_profit: Current profit using entry pricing. :param current_exit_profit: Current profit using exit pricing. :param **kwargs: Ensure to keep this here so updates to this won't break your strategy. :return float: Stake amount to adjust your trade, Positive values to increase position, Negative values to decrease position. Return None for no action. Optionally, return a tuple with a 2nd element with an order reason """ if current_profit > 0.05 and trade.nr_of_successful_exits == 0: # Take half of the profit at +5% return -(trade.stake_amount / 2), "half_profit_5%" if current_profit > -0.05: return None # Obtain pair dataframe (just to show how to access it) dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) # Only buy when not actively falling price. last_candle = dataframe.iloc[-1].squeeze() previous_candle = dataframe.iloc[-2].squeeze() if last_candle["close"] < previous_candle["close"]: return None filled_entries = trade.select_filled_orders(trade.entry_side) count_of_entries = trade.nr_of_successful_entries # Allow up to 3 additional increasingly larger buys (4 in total) # Initial buy is 1x # If that falls to -5% profit, we buy 1.25x more, average profit should increase to roughly -2.2% # If that falls down to -5% again, we buy 1.5x more # If that falls once again down to -5%, we buy 1.75x more # Total stake for this trade would be 1 + 1.25 + 1.5 + 1.75 = 5.5x of the initial allowed stake. # That is why max_dca_multiplier is 5.5 # Hope you have a deep wallet! try: # This returns first order stake size stake_amount = filled_entries[0].stake_amount_filled # This then calculates current safety order size stake_amount = stake_amount * (1 + (count_of_entries * 0.25)) return stake_amount, "1/3rd_increase" except Exception as exception: return None return None