# source: https://raw.githubusercontent.com/1amcord/freqtrade_settings/b078a9d66f5e88f6ad18d35066241aaf91c31032/strategies/RsiHedge.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # --- Do not remove these libs --- import numpy as np # noqa import pandas as pd # noqa from pandas import DataFrame from freqtrade.persistence import Trade from functools import reduce from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter, DecimalParameter, merge_informative_pair) # -------------------------------- # Add your lib to import here import talib.abstract as ta import pandas_ta as pta import freqtrade.vendor.qtpylib.indicators as qtpylib import logging logger = logging.getLogger(__name__) class github_1amcord_freqtrade_settings__RsiHedge__20220814_101522(IStrategy): """ This is a strategy template to get you started. 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_buy_trend, populate_sell_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 = 2 # Optimal timeframe for the strategy. timeframe = '5m' minimal_roi = { "2880": 0.0, # Sell after 1 day if the profit is not negative } stoploss = -10 trailing_stop = False # trailing_stop_positive = 0.001 # trailing_stop_positive_offset = 0.05 # trailing_only_offset_is_reached = True use_custom_stoploss = True # Adjust position amount position_adjustment_enable = True max_entry_position_adjustment = 2 # Run "populate_indicators()" only for new candle. process_only_new_candles = False # These values can be overridden in the "ask_strategy" section in the config. use_sell_signal = True sell_profit_only = False ignore_roi_if_buy_signal = False # Number of candles the strategy requires before producing valid signals startup_candle_count: int = 30 # Strategy parameters buy_rsi = IntParameter(15, 45, default=40, space="buy") min_stoploss_profit = DecimalParameter( 0.01, 0.1, decimals=2, default=0.02, space="sell") # Optional order type mapping. order_types = { 'buy': 'market', 'sell': 'market', 'stoploss': 'market', 'stoploss_on_exchange': True, 'stoploss_on_exchange_interval': 30, 'stoploss_on_exchange_limit_ratio': 0.99 } # Optional order time in force. order_time_in_force = { 'buy': 'gtc', 'sell': 'gtc' } @property def plot_config(self): return { # Main plot indicators (Moving averages, ...) 'main_plot': { 'sma21_1d': {'color': 'white'}, 'ema9': {'color': 'yellow'}, 'ema6_1d': {'color': 'green'}, 'sar_1d': {'color': 'blue'}, }, 'subplots': { # Subplots - each dict defines one additional plot "RSI": { 'rsi': {'color': 'red'}, } } } 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"), ] """ # get access to all pairs available in whitelist. pairs = self.dp.current_whitelist() # Assign tf to each pair so they can be downloaded and cached for strategy. informative_pairs = [(pair, '1d') for pair in pairs] return informative_pairs 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 """ if not self.dp: # Don't do anything if DataProvider is not available. return dataframe inf_tf = '1d' # Get the informative pair informative = self.dp.get_pair_dataframe( pair=metadata['pair'], timeframe=inf_tf) # Get the 21 day SMA for the 1d timeframe informative['sma21'] = ta.SMA(informative, timeperiod=21) informative['avg_price_last_day'] = ( (informative['high'] + informative['low'])/2) # Parabolic SAR informative['sar'] = ta.SAR(informative) # EMA6 informative['ema6'] = ta.EMA(informative, timeperiod=6) # Use the helper function merge_informative_pair to safely merge the pair # Automatically renames the columns and merges a shorter timeframe dataframe and a longer timeframe informative pair # use ffill to have the 1d value available in every row throughout the day. # Without this, comparisons between columns of the original and the informative pair would only work once per day. # Full documentation of this method, see below dataframe = merge_informative_pair( dataframe, informative, self.timeframe, inf_tf, ffill=True) # RSI dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) # ema9 dataframe['ema9'] = ta.EMA(dataframe, timeperiod=9) # Average price dataframe['avg_price'] = ((dataframe['high'] + dataframe['low'])/2) dataframe['min_stoploss_profit'] = self.min_stoploss_profit.value # 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_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the buy signal for the given dataframe :param dataframe: DataFrame populated with indicators :param metadata: Additional information, like the currently traded pair :return: DataFrame with buy column """ # Only buy when not actively falling price. last_candle = dataframe.iloc[-1].squeeze() conditions = [] # Stay away from sma21_1d to avoid sells when crossing below sma21_1d conditions.append((((dataframe['sma21_1d']*0.97) > dataframe['close']) | ((dataframe['sma21_1d']*1.03) < dataframe['close']))) # No falling prices conditions.append((dataframe['ema6_1d'] <= last_candle['avg_price'])) conditions.append((dataframe['rsi'] < self.buy_rsi.value)) conditions.append((dataframe['volume'] > 0)) if conditions: dataframe.loc[ reduce(lambda x, y: x & y, conditions), 'buy'] = 1 # dataframe.loc[ # ( # # No falling prices # (dataframe['ema6_1d'] <= last_candle['avg_price']) & # # Buy low # (reduce(lambda x, y: x & y, conditions)) & # (dataframe['volume'] > 0) # Make sure Volume is not 0 # ), # 'buy'] = 1 return dataframe def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the sell signal for the given dataframe :param dataframe: DataFrame populated with indicators :param metadata: Additional information, like the currently traded pair :return: DataFrame with buy column """ dataframe.loc[ ( # # Signal: RSI crosses above sell_rsi # (dataframe['rsi'] >= 95) & # (dataframe['volume'] > 0) # Make sure Volume is not 0 ), 'sell'] = 0 return dataframe def custom_sell(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float, current_profit: float, **kwargs): """ Sell only when matching some criteria other than those used to generate the sell signal :return: str sell_reason, if any, otherwise None """ # get dataframe dataframe, _ = self.dp.get_analyzed_dataframe( pair=pair, timeframe=self.timeframe) """ SMA21-Check """ # get the current candle current_sma21_1d = dataframe['sma21_1d'].iloc[-1].squeeze() price_crossed_below_sma = qtpylib.crossed_below( dataframe['avg_price'], current_sma21_1d) # if price falls below SMA21 - 1d timeframe, sell if (price_crossed_below_sma.iloc[-1]): return "CUSTOM_SELL_BELOW_1dSMA21" # else, hold return None def custom_stoploss(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float, current_profit: float, **kwargs) -> float: #Return if profit is too low if (current_profit < self.min_stoploss_profit.value): return 1 dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Use parabolic sar as absolute stoploss price sar_1d_price = last_candle['sar_1d'] ema6_1d = last_candle['ema6_1d'] # Stoploss is 1 day EMA6-price, if 1 day parabolic sar and 1 day ema are below # current price. if ((sar_1d_price < current_rate) & (ema6_1d < current_rate)): # Convert absolute price to percentage relative to current_rate return (ema6_1d / current_rate)-1 # if condition above is not fulfilled, stoploss price is 1% below current price. return 0.001 # This is called when placing the initial order (opening trade) def custom_stake_amount(self, pair: str, current_rate: float, proposed_stake: float, min_stake: float, max_stake: float, **kwargs) -> float: # We need to leave most of the funds for possible further DCA orders # This also applies to fixed stakes new_proposed_stake = proposed_stake*0.75 logger.debug("Proposed Stake: %s", new_proposed_stake) return new_proposed_stake # return proposed_stake def adjust_trade_position(self, trade: Trade, current_time: 'datetime', current_rate: float, current_profit: float, min_stake: float, max_stake: float, **kwargs): """ Custom trade adjustment logic, returning the stake amount that a trade should be increased. This means extra buy orders with additional fees. :param trade: trade object. :param current_time: datetime object, containing the current datetime :param current_rate: Current buy rate. :param current_profit: Current profit (as ratio), calculated based on current_rate. :param min_stake: Minimal stake size allowed by exchange. :param max_stake: Balance available for trading. :param **kwargs: Ensure to keep this here so updates to this won't break your strategy. :return float: Stake amount to adjust your trade """ # 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() count_of_buys = trade.nr_of_successful_buys open_order_price = trade.open_rate # if ((count_of_buys == 1) & if ((last_candle['close'] > (open_order_price * 0.98))): return None if (count_of_buys > 2): return None if last_candle['ema9'] < last_candle['close']: return None filled_buys = trade.select_filled_orders('buy') # Allow up to 3 additional increasingly smaller buys (4 in total) # Initial buy is 1x # Hope you have a deep wallet! try: # This returns first order stake size stake_amount = filled_buys[0].cost new_stake_amount = (((stake_amount*100)/75)*(0.25/2)) # This then calculates current safety order size # stake_amount = stake_amount/(self.max_entry_position_adjustment+1) logger.debug('\nLast candle date: %s \n\ - count of buys: %s \n\ - max stake: %s \n\ - Stake Amount: %s \n\ - New Stake Amount: %s \n', last_candle['date'], count_of_buys, max_stake, stake_amount, new_stake_amount) return (new_stake_amount) except Exception as exception: return None return None