# source: https://raw.githubusercontent.com/rmallarapu-bc/brahma/9287745fc036c2f4c00c586e598290885c2883b9/archive/OldSchoolOrignal1.py # --- Do not remove these libs --- from typing import Optional from pandas import DataFrame import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from datetime import datetime from freqtrade.persistence import Trade from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy from functools import reduce # -------------------------------- # Buy hyperspace params: buy_params = { "base_nb_candles_buy": 8, "ewo_high": 4.179, "ewo_low": -16.917, "ewo_high_2": -2.609, # value loaded from strategy "low_offset": 0.986, # value loaded from strategy "low_offset_2": 0.944, # value loaded from strategy "rsi_buy": 58, # value loaded from strategy } # Sell hyperspace params: sell_params = { "base_nb_candles_sell": 16, "high_offset": 1.054, "high_offset_2": 1.018, # value loaded from strategy } def EWO(dataframe, ema_length=5, ema2_length=35): df = dataframe.copy() ema1 = ta.EMA(df, timeperiod=ema_length) ema2 = ta.EMA(df, timeperiod=ema2_length) emadif = (ema1 - ema2) / df['low'] * 100 return emadif class Github_rmallarapu_bc_brahma__OldSchoolOrignal1__20240229_213751(IStrategy): INTERFACE_VERSION: int = 3 minimal_roi = {"0": 0.1} can_short = False # SMAOffset base_nb_candles_buy = IntParameter( 2, 20, default=buy_params['base_nb_candles_buy'], space='buy', optimize=False) base_nb_candles_sell = IntParameter( 10, 40, default=sell_params['base_nb_candles_sell'], space='sell', optimize=False) low_offset = DecimalParameter( 0.9, 0.99, default=buy_params['low_offset'], space='buy', optimize=False) low_offset_2 = DecimalParameter( 0.9, 0.99, default=buy_params['low_offset_2'], space='buy', optimize=False) high_offset = DecimalParameter( 0.95, 1.1, default=sell_params['high_offset'], space='sell', optimize=False) high_offset_2 = DecimalParameter( 0.99, 1.5, default=sell_params['high_offset_2'], space='sell', optimize=False) # Protection fast_ewo = 50 slow_ewo = 200 ewo_low = DecimalParameter(-20.0, -8.0, default=buy_params['ewo_low'], space='buy', optimize=False) ewo_high = DecimalParameter( 2.0, 12.0, default=buy_params['ewo_high'], space='buy', optimize=False) ewo_high_2 = DecimalParameter( -6.0, 12.0, default=buy_params['ewo_high_2'], space='buy', optimize=False) rsi_buy = IntParameter(30, 70, default=buy_params['rsi_buy'], space='buy', optimize=True) # Trailing stop: trailing_stop = True trailing_stop_positive = 0.0025 trailing_stop_positive_offset = 0.01 trailing_only_offset_is_reached = True # Sell signal use_exit_signal = True exit_profit_only = False exit_profit_offset = 0.01 ignore_roi_if_entry_signal = False # Optional order time in force. order_time_in_force = { 'entry': 'gtc', 'exit': 'gtc' } # # Trailing stoploss # trailing_stop = True # trailing_stop_positive = 0.005 # trailing_stop_positive_offset = 0.01 # trailing_only_offset_is_reached = True # # Optimal stoploss designed for the strategy stoploss = -0.72 # # use_exit_signal = False # ignore_roi_if_entry_signal = False # Optimal timeframe for the strategy # timeframe = '1h' timeframe = '5m' informative_timeframe = '15m' # Strategy parameters rsi_length = IntParameter(10, 40, default=14, space="buy") bb_window = IntParameter(10, 40, default=17, space="buy") bb_std = IntParameter(1, 6, default=1, space="buy") buy_rsi = IntParameter(10, 40, default=42, space="buy") sell_rsi = IntParameter(60, 90, default=76, space="sell") slippage_protection = { 'retries': 3, 'max_slippage': -0.01 } buy_signals = {} # timeframe = '5m' # informative_timeframe = '1h' process_only_new_candles = False startup_candle_count = 200 use_custom_stoploss = False def informative_pairs(self): pairs = self.dp.current_whitelist() informative_pairs = [(pair, self.informative_timeframe) for pair in pairs] return informative_pairs def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, sell_reason: str, current_time: datetime, **kwargs) -> bool: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1] if (last_candle is not None): if (sell_reason in ['sell_signal']): if (last_candle['hma_50'] * 1.149 > last_candle['ema_100']) and ( last_candle['close'] < last_candle['ema_100'] * 0.951): # *1.2 return False # slippage try: state = self.slippage_protection['__pair_retries'] except KeyError: state = self.slippage_protection['__pair_retries'] = {} candle = dataframe.iloc[-1].squeeze() slippage = (rate / candle['close']) - 1 if slippage < self.slippage_protection['max_slippage']: pair_retries = state.get(pair, 0) if pair_retries < self.slippage_protection['retries']: state[pair] = pair_retries + 1 return False state[pair] = 0 return True # def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_length.value) # # # Bollinger bands # bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=self.bb_window.value, # stds=self.bb_std.value) # dataframe['bb_lowerband'] = bollinger['lower'] # dataframe['bb_middleband'] = bollinger['mid'] # dataframe['bb_upperband'] = bollinger['upper'] # # return dataframe # # def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # dataframe.loc[ # ( # (dataframe['rsi'] < self.buy_rsi.value) & # (dataframe['close'] < dataframe['bb_lowerband']) # # ), # 'enter_long'] = 1 # dataframe.loc[ # ( # (dataframe['rsi'] > self.sell_rsi.value) & # (dataframe['close'] > dataframe['bb_upperband']) # # ), # 'enter_short'] = 1 # return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Calculate all ma_buy values for val in self.base_nb_candles_buy.range: dataframe[f'ma_buy_{val}'] = ta.EMA(dataframe, timeperiod=val) # Calculate all ma_sell values for val in self.base_nb_candles_sell.range: dataframe[f'ma_sell_{val}'] = ta.EMA(dataframe, timeperiod=val) dataframe['hma_50'] = qtpylib.hull_moving_average(dataframe['close'], window=50) dataframe['ema_100'] = ta.EMA(dataframe, timeperiod=100) dataframe['sma_9'] = ta.SMA(dataframe, timeperiod=9) # Elliot dataframe['EWO'] = EWO(dataframe, self.fast_ewo, self.slow_ewo) # RSI dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['rsi_fast'] = ta.RSI(dataframe, timeperiod=4) dataframe['rsi_slow'] = ta.RSI(dataframe, timeperiod=20) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( (dataframe['rsi_fast'] < 35) & (dataframe['close'] < ( dataframe[f'ma_buy_{self.base_nb_candles_buy.value}'] * self.low_offset.value)) & (dataframe['EWO'] > self.ewo_high.value) & (dataframe['rsi'] < self.rsi_buy.value) & (dataframe['volume'] > 0) & (dataframe['close'] < ( dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) ), ['buy', 'buy_tag']] = (1, 'ewo1') dataframe.loc[ ( (dataframe['rsi_fast'] < 35) & (dataframe['close'] < ( dataframe[f'ma_buy_{self.base_nb_candles_buy.value}'] * self.low_offset_2.value)) & (dataframe['EWO'] > self.ewo_high_2.value) & (dataframe['rsi'] < self.rsi_buy.value) & (dataframe['volume'] > 0) & (dataframe['close'] < ( dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) & (dataframe['rsi'] < self.buy_rsi.value) ), ['buy', 'buy_tag']] = (1, 'ewo2') dataframe.loc[ ( (dataframe['rsi_fast'] < 35) & (dataframe['close'] < ( dataframe[f'ma_buy_{self.base_nb_candles_buy.value}'] * self.low_offset.value)) & (dataframe['EWO'] < self.ewo_low.value) & (dataframe['volume'] > 0) & (dataframe['close'] < ( dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) ), ['buy', 'buy_tag']] = (1, 'ewolow') return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [] conditions.append( ((dataframe['close'] > dataframe['sma_9']) & (dataframe['close'] > ( dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset_2.value)) & (dataframe['rsi'] > 50) & (dataframe['volume'] > 0) & (dataframe['rsi_fast'] > dataframe['rsi_slow']) ) | ( (dataframe['close'] < dataframe['hma_50']) & (dataframe['close'] > ( dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) & (dataframe['volume'] > 0) & (dataframe['rsi_fast'] > dataframe['rsi_slow']) ) ) if conditions: dataframe.loc[ reduce(lambda x, y: x | y, conditions), 'sell' ] = 1 return dataframe initial_order_size = 0.5 position_adjustment_enable = True # Example specific variables max_entry_position_adjustment = 2 # This number is explained a bit further down max_dca_multiplier = 2 # 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: Optional[float], max_stake: float, # leverage: float, entry_tag: Optional[str], 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 custom_stake_amount_2(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: return proposed_stake * self.initial_order_size def adjust_trade_position_2(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: Optional[float], max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs) -> Optional[float]: """ 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. """ if current_profit > 0.05 and trade.nr_of_successful_exits == 0: # Take half of the profit at +5% return -(trade.stake_amount / 2) if current_profit > -0.05: return None 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 if current_profit > -0.01 and count_of_entries < self.max_entry_position_adjustment: # Obtain pair dataframe (just to show how to access it) # 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 # This then calculates current safety order size # stake_amount = stake_amount * (1 + (count_of_entries * 0.25)) stake_amount = stake_amount * (count_of_entries * 0.25) return stake_amount except Exception as exception: return None return None