# source: https://raw.githubusercontent.com/RoboticAutomations/freqtradestrategies/52c4bd40414645f079a980cf4486dc7af4061efb/dirty%20LA/from-AIRA/battle-tank/TankAi15__3.py # directory_url: https://github.com/RoboticAutomations/freqtradestrategies/blob/main/dirty LA/from-AIRA/battle-tank/ # User: RoboticAutomations # Repository: freqtradestrategies # --------------------import logging from functools import reduce import datetime import talib.abstract as ta import pandas_ta as pta import logging import numpy as np import pandas as pd from finta import TA import pandas_ta as pta import ta as tastc #from murrey_math import calculate_murrey_math_levels import freqtrade.vendor.qtpylib.indicators as qtpylib from technical import qtpylib from datetime import timedelta, datetime, timezone from pandas import DataFrame, Series from technical import qtpylib from typing import Optional from freqtrade.strategy.interface import IStrategy from technical.pivots_points import pivots_points from freqtrade.exchange import timeframe_to_prev_date, timeframe_to_minutes from freqtrade.persistence import Trade from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter, RealParameter, merge_informative_pair) from scipy.signal import argrelextrema from typing import Optional from functools import reduce import warnings import math warnings.simplefilter(action="ignore", category=pd.errors.PerformanceWarning) logger = logging.getLogger(__name__) class Github_RoboticAutomations_freqtradestrategies__TankAi15__3__20260915_233254(IStrategy): exit_profit_only = True # No selling at a loss use_custom_stoploss = False trailing_stop = False ignore_roi_if_entry_signal = True process_only_new_candles = True can_short = True use_exit_signal = True startup_candle_count: int = 200 stoploss = -0.99 timeframe = '15m' # DCA max_dca_multiplier = 3 position_adjustment_enable = True max_entry_position_adjustment = 2 max_safety_orders = IntParameter(1, 6, default=3, space='buy', optimize=True) #initial_safety_order_trigger = DecimalParameter(low=-0.02, high=-0.01, default=-0.018, decimals=3 ,space='buy', optimize=True, load=True) #safety_order_step_scale = DecimalParameter(low=1.05, high=1.5, default=1.25, decimals=2 ,space='buy', optimize=True, load=True) #safety_order_volume_scale = DecimalParameter(low=1.1, high=2, default=1.4, decimals=1 ,space='buy', optimize=True, load=True) #safety_order_reserve = IntParameter(1, 10, default=1, space='buy', optimize=True) ### Custom Functions # Custom Entry increment = DecimalParameter(low=1.0005, high=1.002, default=1.001, decimals=4 ,space='buy', optimize=True, load=True) last_entry_price = None #trailing stop loss optimiziation tsl_target6 = DecimalParameter(low=0.45, high=0.55, default=0.51, decimals=2, space='sell', optimize=True, load=True) ts6 = DecimalParameter(low=0.05, high=0.07, default=0.06,decimals=2, space='sell', optimize=True, load=True) tsl_target5 = DecimalParameter(low=0.4, high=0.5, default=0.46, decimals=2, space='sell', optimize=True, load=True) ts5 = DecimalParameter(low=0.04, high=0.06, default=0.05,decimals=2, space='sell', optimize=True, load=True) tsl_target4 = DecimalParameter(low=0.36, high=0.44, default=0.41, decimals=2, space='sell', optimize=True, load=True) ts4 = DecimalParameter(low=0.04, high=0.05, default=0.045, decimals=3, space='sell', optimize=True, load=True) tsl_target3 = DecimalParameter(low=0.27, high=0.33, default=0.31, decimals=2,space='sell', optimize=True, load=True) ts3 = DecimalParameter(low=0.03, high=0.04, default=0.035, decimals=3,space='sell', optimize=True, load=True) tsl_target2 = DecimalParameter(low=0.17, high=0.23, default=0.21, decimals=2,space='sell', optimize=True, load=True) ts2 = DecimalParameter(low=0.025, high=0.035, default=0.03, decimals=3,space='sell', optimize=True, load=True) tsl_target1 = DecimalParameter(low=0.08, high=0.12, default=0.11, decimals=2,space='sell', optimize=True, load=True) ts1 = DecimalParameter(low=0.015, high=0.03, default=0.02, decimals=2,space='sell', optimize=True, load=True) tsl_target0 = DecimalParameter(low=0.04, high=0.06, default=0.055, decimals=3,space='sell', optimize=True, load=True) ts0 = DecimalParameter(low=0.01, high=0.016, default=0.013, decimals=3,space='sell', optimize=True, load=True) # protections cooldown_lookback = IntParameter(2, 48, default=1, space="protection", optimize=True) stop_duration = IntParameter(12, 200, default=4, space="protection", optimize=True) use_stop_protection = BooleanParameter(default=True, space="protection", optimize=True) minimal_roi = { '0': 0.50, '15': 0.45, '30': 0.40, '60': 0.30, '90': 0.20, '120': 0.10, '180': 0.05, '240': 0.001, } plot_config = { "main_plot": {}, "subplots": { "extrema": { "&s-extrema": { "color": "#f53580", "type": "line" }, "&s-minima_sort_threshold": { "color": "#4ae747", "type": "line" }, "&s-maxima_sort_threshold": { "color": "#5b5e4b", "type": "line" } }, "CCI": { "cci1": { "color": "#f53580", "type": "line" }, "cci2": { "color": "#4ae747", "type": "line" } }, "STC": { 'stc5': {'color': 'red'}, 'rsx': {'color': 'blue'} }, "AI": { "do_predict": { "color": "#4f227c", "type": "bar" }, "minima": { "color": "#2a643c", "type": "line" }, "maxima": { "color": "#e277b7", "type": "line" }, "minima_check": { "color": "#2a643c", "type": "line" }, "maxima_check": { "color": "#e277b7", "type": "line" } } } } @property def protections(self): prot = [] prot.append({ "method": "CooldownPeriod", "stop_duration_candles": self.cooldown_lookback.value }) if self.use_stop_protection.value: prot.append({ "method": "StoplossGuard", "lookback_period_candles": 24 * 3, "trade_limit": 2, "stop_duration_candles": self.stop_duration.value, "only_per_pair": False }) return prot ### Custom Functions ### ### Dollar Cost Averaging ### # 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: """ Determine the custom stake amount for the initial order. """ # We need to leave most of the funds for possible further DCA orders return proposed_stake / self.max_dca_multiplier # This also applies to fixed staked return proposed_stake def custom_entry_price(self, pair: str, trade: Optional['Trade'], current_time: datetime, proposed_rate: float, entry_tag: Optional[str], side: str, **kwargs) -> float: dataframe, last_updated = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) entry_price = (dataframe['close'].iat[-1] + dataframe['open'].iat[-1] + proposed_rate + proposed_rate) / 4 if proposed_rate < entry_price: entry_price = proposed_rate logger.info(f"{pair} Using Entry Price: {entry_price} | close: {dataframe['close'].iat[-1]} open: {dataframe['open'].iat[-1]} proposed_rate: {proposed_rate}") # Check if there is a stored last entry price and if it matches the proposed entry price if self.last_entry_price is not None and abs(entry_price - self.last_entry_price) < 0.001: # Tolerance for floating-point comparison entry_price *= self.increment.value # Increment by 0.2% logger.info(f"{pair} Incremented entry price: {entry_price} based on previous entry price : {self.last_entry_price}.") # Update the last entry price self.last_entry_price = entry_price return entry_price 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: if exit_reason == 'partial_exit' and trade.calc_profit_ratio(rate) < 0: logger.info(f"{trade.pair} partial exit is below 0") self.dp.send_msg(f'{trade.pair} partial exit is below 0') return False if exit_reason == 'trailing_stop_loss' and trade.calc_profit_ratio(rate) < 0: logger.info(f"{trade.pair} trailing stop price is below 0") self.dp.send_msg(f'{trade.pair} trailing stop price is below 0') return False return True def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float, current_profit: float, **kwargs): if current_profit <= -0.85: # Se la perdita รจ inferiore al -85% logger.info(f'{trade.pair} Loss is greater than 85%. Exiting position.') self.dp.send_msg(f'{trade.pair} Loss is greater than 85%. Exiting position.') return True # Chiudi la posizione return False # Altrimenti, mantieni la posizione aperta def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) current_candle = dataframe.iloc[-1].squeeze() trade_duration = (current_time - trade.open_date_utc).seconds / 60 total_trade_days = (current_time - trade.open_date_utc).days if trade_duration > 0: if (current_profit > self.tsl_target6.range): self.dp.send_msg(f'*** {pair} *** Profit: {current_profit} - lvl5 {stop5}/{stop5a} activated') return self.ts6.range if trade_duration > 15: if (current_profit > self.tsl_target5.range): self.dp.send_msg(f'*** {pair} *** Profit: {current_profit} - lvl5 {stop5}/{stop5a} activated') return self.ts5.range if trade_duration > 30: if (current_profit > self.tsl_target4.range): self.dp.send_msg(f'*** {pair} *** Profit {current_profit} - lvl4 {stop4}/{stop4a} activated') return self.ts4.range if trade_duration > 60: if (current_profit > self.tsl_target3.range): self.dp.send_msg(f'*** {pair} *** Profit {current_profit} - lvl3 {stop3}/{stop3a} activated') return self.ts3.range if trade_duration > 90: if (current_profit > self.tsl_target2.range): self.dp.send_msg(f'*** {pair} *** Profit {current_profit} - lvl2 {stop2}/{stop2a} activated') return self.ts2.range if trade_duration > 120: if (current_profit > self.tsl_target1.range): self.dp.send_msg(f'*** {pair} *** Profit {current_profit} - lvl1 {stop2}/{stop2a} activated') return self.ts1.range if trade_duration > 180: if (current_profit > self.tsl_target0.range): self.dp.send_msg(f'*** {pair} *** Profit {current_profit} - lvl0 {stop2}/{stop2a} activated') return self.ts0.range return self.stoploss ''' ### if trade is past maturity or quick profit engage lower trailing stops to either exit or lock in profits if total_trade_days > 1: logging.info(f"{pair}: 2nd trailing stop engaged {trade_duration}m : {total_trade_days}d elapsed") for stop1 in self.tsl_target1.range: if (current_profit > stop1): for stop1a in self.ts1.range: self.dp.send_msg(f'*** {pair} *** Profit {current_profit} - lvl1 {stop1}/{stop1a} activated') return stop1a if total_trade_days < 1 and trade_duration < 360: logging.info(f"{pair}: 2nd trailing stop engaged {trade_duration}m : {total_trade_days}d elapsed") for stop1 in self.tsl_target1.range: if (current_profit > stop1): for stop1a in self.ts1.range: self.dp.send_msg(f'*** {pair} *** Profit {current_profit} - lvl1 {stop1}/{stop1a} activated') return stop1a if total_trade_days > 2: logging.info(f"{pair} 1st trailing stop engaged {trade_duration} elapsed : {total_trade_days}d elapsed") for stop0 in self.tsl_target0.range: if (current_profit > stop0): for stop0a in self.ts0.range: self.dp.send_msg(f'*** {pair} *** Profit {current_profit} - lvl0 {stop0}/{stop0a} activated') return stop0a if total_trade_days < 1 and trade_duration < 360: logging.info(f"{pair} 1st trailing stop engaged {trade_duration} elapsed : {total_trade_days}d elapsed") for stop0 in self.tsl_target0.range: if (current_profit > stop0): for stop0a in self.ts0.range: self.dp.send_msg(f'*** {pair} *** Profit {current_profit} - lvl0 {stop0}/{stop0a} activated') return stop0a else: for stop0 in self.tsl_target0.range: if (current_profit > stop0): self.dp.send_msg(f'*** {pair} *** Profit {current_profit} SWINGING FOR THE MOON!!!') return 0.99 return self.stoploss ''' def adjust_trade_position(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[str]: """ Adjust the trade position based on the current profit and the number of successful entries. """ dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) filled_entries = trade.select_filled_orders(trade.entry_side) count_of_entries = trade.nr_of_successful_entries trade_duration = (current_time - trade.open_date_utc).seconds / 60 # Take Profit if current_profit > 0.25 and trade.nr_of_successful_exits == 0: # Take half of the profit at +25% return -(trade.stake_amount / 4) if current_profit > 0.40 and trade.nr_of_successful_exits == 1: # Take half of the profit at +40% return -(trade.stake_amount / 3) # Profit Based DCA if current_profit > -0.15 and trade.nr_of_successful_entries == 1: return None if current_profit > -0.3 and trade.nr_of_successful_entries == 2: return None if current_profit > -0.6 and trade.nr_of_successful_entries == 3: return None # Check if max number of safety orders is reached if trade.nr_of_successful_entries >= self.max_safety_orders.value: return None 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 return stake_amount, except Exception as exception: # Log the exception if needed print(f"Error adjusting trade position: {exception}") return None return None #Leverage def leverage(self, pair: str, current_time: 'datetime', current_rate: float, proposed_leverage: float, max_leverage: float, side: str, **kwargs) -> float: """ Customize leverage for each new trade. :param pair: Pair that's currently analyzed :param current_time: datetime object, containing the current datetime :param current_rate: Rate, calculated based on pricing settings in exit_pricing. :param proposed_leverage: A leverage proposed by the bot. :param max_leverage: Max leverage allowed on this pair :param side: 'long' or 'short' - indicating the direction of the proposed trade :return: A leverage amount, which is between 1.0 and max_leverage. """ window_size = 50 # Obtain historical candle data for the given pair and timeframe dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) # Extract required historical data for indicators historical_close_prices = dataframe['close'].tail(window_size) historical_high_prices = dataframe['high'].tail(window_size) historical_low_prices = dataframe['low'].tail(window_size) # Set base leverage base_leverage = 15 # Calculate RSI and ATR based on historical data using TA-Lib rsi_values = ta.RSI(historical_close_prices, timeperiod=14) # Adjust the time period as needed atr_values = ta.ATR(historical_high_prices, historical_low_prices, historical_close_prices, timeperiod=14) # Adjust the time period as needed # Calculate MACD and SMA based on historical data macd_line, signal_line, _ = ta.MACD(historical_close_prices, fastperiod=12, slowperiod=26, signalperiod=9) sma_values = ta.SMA(historical_close_prices, timeperiod=20) # Get the current RSI and ATR values from the last data point in the historical window current_rsi = rsi_values[-1] if len(rsi_values) > 0 else 50.0 # Default value if no data available current_atr = atr_values[-1] if len(atr_values) > 0 else 0.0 # Default value if no data available # Get the current MACD and SMA values current_macd = macd_line[-1] - signal_line[-1] if len(macd_line) > 0 and len(signal_line) > 0 else 0.0 current_sma = sma_values[-1] if len(sma_values) > 0 else 0.0 # Define dynamic thresholds for RSI and ATR for leverage adjustments # Set default values or use non-NaN values if available dynamic_rsi_low = np.nanmin(rsi_values) if len(rsi_values) > 0 and not np.isnan(np.nanmin(rsi_values)) else 30.0 dynamic_rsi_high = np.nanmax(rsi_values) if len(rsi_values) > 0 and not np.isnan(np.nanmax(rsi_values)) else 70.0 dynamic_atr_low = np.nanmin(atr_values) if len(atr_values) > 0 and not np.isnan(np.nanmin(atr_values)) else 0.002 dynamic_atr_high = np.nanmax(atr_values) if len(atr_values) > 0 and not np.isnan(np.nanmax(atr_values)) else 0.005 # Print variables for debugging print("Historical Close Prices:", historical_close_prices) print("RSI Values:", rsi_values) print("ATR Values:", atr_values) print("Current RSI:", current_rsi) print("Current ATR:", current_atr) print("Current MACD:", current_macd) print("Current SMA:", current_sma) print("Dynamic RSI Low:", dynamic_rsi_low) print("Dynamic RSI High:", dynamic_rsi_high) print("Dynamic ATR Low:", dynamic_atr_low) print("Dynamic ATR High:", dynamic_atr_high) # Leverage adjustment factors long_increase_factor = 22 # Increase the base leverage for long positions long_decrease_factor = 12 # Decrease the base leverage for long positions short_increase_factor = 22 # Increase base leverage for short positions short_decrease_factor = 12 # Decrease the base leverage for short positions volatility_decrease_factor = 8 # Decrease to 10 the base leverage when volatility is high # Adjust leverage for long trades if side == 'long': # Adjust leverage for short trades based on dynamic thresholds and current RSI if current_rsi < dynamic_rsi_low: base_leverage = long_increase_factor elif current_rsi > dynamic_rsi_high: base_leverage = long_decrease_factor if current_atr > (current_rate * 0.03): base_leverage = volatility_decrease_factor # Adjust leverage based on MACD and SMA if current_macd > 0: base_leverage = long_increase_factor if current_rate < current_sma: base_leverage = long_decrease_factor # Adjust leverage for short trades elif side == 'short': # Adjust leverage for short trades based on dynamic thresholds and current RSI if current_rsi > dynamic_rsi_high: base_leverage = short_increase_factor elif current_rsi < dynamic_rsi_low: base_leverage = short_decrease_factor if current_atr > (current_rate * 0.03): base_leverage = volatility_decrease_factor # Adjust leverage based on MACD and SMA if current_macd < 0: base_leverage = short_increase_factor # Increase leverage for potential downward movement if current_rate > current_sma: base_leverage = short_decrease_factor # Decrease leverage if price is above the moving average else: return proposed_leverage # Return the proposed leverage if side is neither 'long' nor 'short' # Apply maximum and minimum limits to the adjusted leverage adjusted_leverage = max(min(base_leverage, max_leverage), 1.0) # Apply max and min limits # Print variables for debugging print("Proposed Leverage:", proposed_leverage) print("Adjusted Leverage:", adjusted_leverage) return adjusted_leverage # Return the adjusted leverage def feature_engineering_expand_all(self, dataframe, period, **kwargs): dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period) dataframe["%-mfi-period"] = ta.MFI(dataframe, timeperiod=period) dataframe["%-rocr-period"] = ta.ROCR(dataframe, timeperiod=period) dataframe["%-cmf-period"] = chaikin_mf(dataframe, periods=period) dataframe["%-chop-period"] = qtpylib.chopiness(dataframe, period) dataframe["%-linear-period"] = ta.LINEARREG_ANGLE( dataframe['close'], timeperiod=period) dataframe["%-atr-period"] = ta.ATR(dataframe, timeperiod=period) dataframe["%-atr-periodp"] = dataframe["%-atr-period"] / \ dataframe['close'] * 1000 return dataframe def feature_engineering_expand_basic(self, dataframe, metadata, **kwargs): dataframe["%-raw_volume"] = dataframe["volume"] dataframe["%-obv"] = ta.OBV(dataframe) dataframe["dpo"] = pta.dpo(dataframe['close'], length=40, centered=False) dataframe["%-dpo"] = dataframe["dpo"] # Williams R% dataframe['%-willr14'] = pta.willr(dataframe['high'], dataframe['low'], dataframe['close']) # VWAP vwap_low, vwap, vwap_high = VWAPB(dataframe, 20, 1) dataframe['%-vwap_upperband'] = vwap_high dataframe['%-vwap_middleband'] = vwap dataframe['%-vwap_lowerband'] = vwap_low dataframe['%-vwap_width'] = ((dataframe['%-vwap_upperband'] - dataframe['%-vwap_lowerband']) / dataframe['%-vwap_middleband']) * 100 dataframe = dataframe.copy() dataframe['%-dist_to_vwap_upperband'] = get_distance(dataframe['close'], dataframe['%-vwap_upperband']) dataframe['%-dist_to_vwap_middleband'] = get_distance(dataframe['close'], dataframe['%-vwap_middleband']) dataframe['%-dist_to_vwap_lowerband'] = get_distance(dataframe['close'], dataframe['%-vwap_lowerband']) dataframe['%-tail'] = (dataframe['close'] - dataframe['low']).abs() dataframe['%-wick'] = (dataframe['high'] - dataframe['close']).abs() dataframe['%-rawclose'] = dataframe['close'] dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-raw_volume"] = dataframe["volume"] dataframe["%-raw_price"] = dataframe["close"] dataframe["%-raw_open"] = dataframe["open"] dataframe["%-raw_low"] = dataframe["low"] dataframe["%-raw_high"] = dataframe["high"] heikinashi = qtpylib.heikinashi(dataframe) dataframe['%-ha_open'] = heikinashi['open'] dataframe['%-ha_close'] = heikinashi['close'] dataframe['ha_open'] = heikinashi['open'] dataframe['ha_close'] = heikinashi['close'] dataframe['%-ha_high'] = heikinashi['high'] dataframe['%-ha_low'] = heikinashi['low'] dataframe['%-ha_closedelta'] = (heikinashi['close'] - heikinashi['close'].shift()) dataframe['%-ha_tail'] = (heikinashi['close'] - heikinashi['low']) dataframe['%-ha_wick'] = (heikinashi['high'] - heikinashi['close']) dataframe['%-HLC3'] = (heikinashi['high'] + heikinashi['low'] + heikinashi['close'])/3 murrey_math_levels = calculate_murrey_math_levels(dataframe) for level, value in murrey_math_levels.items(): dataframe[level] = value dataframe["%-+3/8"] = dataframe["[+3/8]P"] dataframe["%-+2/8"] = dataframe["[+2/8]P"] dataframe["%-+1/8"] = dataframe["[+1/8]P"] dataframe["%-8/8"] = dataframe["[8/8]P"] dataframe["%-7/8"] = dataframe["[7/8]P"] dataframe["%-6/8"] = dataframe["[6/8]P"] dataframe["%-5/8"] = dataframe["[5/8]P"] dataframe["%-4/8"] = dataframe["[4/8]P"] dataframe["%-3/8"] = dataframe["[3/8]P"] dataframe["%-2/8"] = dataframe["[2/8]P"] dataframe["%-1/8"] = dataframe["[1/8]P"] dataframe["%-0/8"] = dataframe["[0/8]P"] dataframe["%--1/8"] = dataframe["[-1/8]P"] dataframe["%--2/8"] = dataframe["[-2/8]P"] dataframe["%--3/8"] = dataframe["[-3/8]P"] dataframe['ema_2'] = ta.EMA(dataframe, timeperiod=2) dataframe['%-distema2'] = get_distance(dataframe['ema_2'], dataframe['[+3/8]P']) dataframe['%-distema2'] = get_distance(dataframe['ema_2'], dataframe['[+2/8]P']) dataframe['%-distema2'] = get_distance(dataframe['ema_2'], dataframe['[+1/8]P']) dataframe['%-distema2'] = get_distance(dataframe['ema_2'], dataframe['[8/8]P']) dataframe['%-distema2'] = get_distance(dataframe['ema_2'], dataframe['[4/8]P']) dataframe['%-distema2'] = get_distance(dataframe['ema_2'], dataframe['[0/8]P']) dataframe['%-distema2'] = get_distance(dataframe['ema_2'], dataframe['[-1/8]P']) dataframe['%-distema2'] = get_distance(dataframe['ema_2'], dataframe['[-2/8]P']) dataframe['%-distema2'] = get_distance(dataframe['ema_2'], dataframe['[-3/8]P']) dataframe['%-entrythreshold4'] = (dataframe['%-tail'] - dataframe['[0/8]P']) dataframe["%-entrythreshold5"] = (dataframe['%-tail'] - dataframe['[-1/8]P']) dataframe["%-entrythreshold6"] = (dataframe['%-tail'] - dataframe['[-2/8]P']) dataframe["%-entrythreshold7"] = (dataframe['%-tail'] - dataframe['[-3/8]P']) dataframe["%-exitthreshold4"] = (dataframe['%-wick'] - dataframe['[8/8]P']) dataframe["%-exitthreshold5"] = (dataframe['%-wick'] - dataframe['[+1/8]P']) dataframe["%-exitthreshold6"] = (dataframe['%-wick'] - dataframe['[+2/8]P']) dataframe["%-exitthreshold7"] = (dataframe['%-wick'] - dataframe['[+3/8]P']) dataframe['mmlextreme_oscillator'] = 100 * ((dataframe['close'] - dataframe["[-3/8]P"]) / (dataframe["[+3/8]P"] - dataframe["[-3/8]P"])) dataframe['%-mmlextreme_oscillator'] = dataframe['mmlextreme_oscillator'] # Calculate the percentage change between the high and open prices for each 5-minute candle dataframe['%-perc_change'] = (dataframe['high'] / dataframe['open'] - 1) * 100 # Create a custom indicator that checks if any of the past 100 5-minute candles' high price is 3% or more above the open price dataframe['%-candle_1perc_50'] = dataframe['%-perc_change'].rolling(50).apply(lambda x: np.where(x >= 1, 1, 0).sum()).shift() dataframe['%-candle_2perc_50'] = dataframe['%-perc_change'].rolling(50).apply(lambda x: np.where(x >= 2, 1, 0).sum()).shift() dataframe['%-candle_3perc_50'] = dataframe['%-perc_change'].rolling(50).apply(lambda x: np.where(x >= 3, 1, 0).sum()).shift() dataframe['%-candle_5perc_50'] = dataframe['%-perc_change'].rolling(50).apply(lambda x: np.where(x >= 5, 1, 0).sum()).shift() dataframe['%-candle_-1perc_50'] = dataframe['%-perc_change'].rolling(50).apply(lambda x: np.where(x <= -1, -1, 0).sum()).shift() dataframe['%-candle_-2perc_50'] = dataframe['%-perc_change'].rolling(50).apply(lambda x: np.where(x <= -2, -1, 0).sum()).shift() dataframe['%-candle_-3perc_50'] = dataframe['%-perc_change'].rolling(50).apply(lambda x: np.where(x <= -3, -1, 0).sum()).shift() dataframe['%-candle_-5perc_50'] = dataframe['%-perc_change'].rolling(50).apply(lambda x: np.where(x <= -5, -1, 0).sum()).shift() # Calculate the percentage of the current candle's range where the close price is dataframe['%-close_percentage'] = (dataframe['close'] - dataframe['low']) / (dataframe['high'] - dataframe['low']) dataframe['%-body_size'] = abs(dataframe['open'] - dataframe['close']) dataframe['%-range_size'] = dataframe['high'] - dataframe['low'] dataframe['%-body_range_ratio'] = dataframe['%-body_size'] / dataframe['%-range_size'] dataframe['%-upper_wick_size'] = dataframe['high'] - dataframe[['open', 'close']].max(axis=1) dataframe['%-upper_wick_range_ratio'] = dataframe['%-upper_wick_size'] / dataframe['%-range_size'] lookback_period = 10 dataframe['%-max_high'] = dataframe['high'].rolling(50).max() dataframe['%-min_low'] = dataframe['low'].rolling(50).min() dataframe['%-close_position'] = (dataframe['close'] - dataframe['%-min_low']) / (dataframe['%-max_high'] - dataframe['%-min_low']) dataframe['%-current_candle_perc_change'] = (dataframe['high'] / dataframe['open'] - 1) * 100 # Lazy Bear Impulse Macd dataframe['%-hi'] = ta.SMA(dataframe['high'], timeperiod = 28) dataframe['%-lo'] = ta.SMA(dataframe['low'], timeperiod = 28) dataframe['%-ema1'] = ta.EMA(dataframe['%-HLC3'], timeperiod = 28) dataframe['%-ema2'] = ta.EMA(dataframe['%-ema1'], timeperiod = 28) dataframe['%-d'] = dataframe['%-ema1'] - dataframe['%-ema2'] dataframe['%-mi'] = dataframe['%-ema1'] + dataframe['%-d'] dataframe['%-md'] = np.where(dataframe['%-mi'] > dataframe['%-hi'], dataframe['%-mi'] - dataframe['%-hi'], np.where(dataframe['%-mi'] < dataframe['%-lo'], dataframe['%-mi'] - dataframe['%-lo'], 0)) dataframe['%-sb'] = ta.SMA(dataframe['%-md'], timeperiod = 8) dataframe['%-sh'] = dataframe['%-md'] - dataframe['%-sb'] # WaveTrend using OHLC4 or HA close - 3/21 ap = (0.333 * (heikinashi['high'] + heikinashi['low'] + heikinashi["close"])) dataframe['esa'] = ta.EMA(ap, timeperiod = 9) dataframe['d'] = ta.EMA(abs(ap - dataframe['esa']), timeperiod = 9) dataframe['%-wave_ci'] = (ap-dataframe['esa']) / (0.015 * dataframe['d']) dataframe['%-wave_t1'] = ta.EMA(dataframe['%-wave_ci'], timeperiod = 12) dataframe['%-wave_t2'] = ta.SMA(dataframe['%-wave_t1'], timeperiod = 4) # 200 SMA and distance dataframe['%-200sma'] = ta.SMA(dataframe, timeperiod = 200) dataframe['%-200sma_dist'] = get_distance(heikinashi["close"], dataframe['%-200sma']) return dataframe def feature_engineering_standard(self, dataframe, **kwargs): dataframe["%-day_of_week"] = (dataframe["date"].dt.dayofweek + 1) / 7 dataframe["%-hour_of_day"] = (dataframe["date"].dt.hour + 1) / 25 return dataframe def set_freqai_targets(self, dataframe, **kwargs): dataframe["&s-extrema"] = 0 min_peaks = argrelextrema( dataframe["open"].values, np.less, order=self.freqai_info["feature_parameters"]["label_period_candles"] ) max_peaks = argrelextrema( dataframe["close"].values, np.greater, order=self.freqai_info["feature_parameters"]["label_period_candles"] ) for mp in min_peaks[0]: dataframe.at[mp, "&s-extrema"] = -1 for mp in max_peaks[0]: dataframe.at[mp, "&s-extrema"] = 1 dataframe["minima"] = np.where(dataframe["&s-extrema"] == -1, 1, 0) dataframe["maxima"] = np.where(dataframe["&s-extrema"] == 1, 1, 0) dataframe['&s-extrema'] = dataframe['&s-extrema'].rolling( window=5, win_type='gaussian', center=True).mean(std=0.5) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['rsi'] = ta.RSI(dataframe) dataframe['HCL3'] = (dataframe['high'] + dataframe['close'] + dataframe['low']) / 3 dataframe['rsx'] = pta.rsx(dataframe['HCL3'],length=14, drift=1, offset=0) stc = tastc.trend.stc(dataframe['close'], window_slow=120, window_fast=18, cycle=12) dataframe["stc5"] = stc dataframe['200sma'] = ta.SMA(dataframe, timeperiod=200) dataframe = self.freqai.start(dataframe, metadata, self) murrey_math_levels = calculate_murrey_math_levels(dataframe) for level, value in murrey_math_levels.items(): dataframe[level] = value dataframe['mmlextreme_oscillator'] = 100 * ((dataframe['close'] - dataframe["[4/8]P"]) / (dataframe["[+3/8]P"] - dataframe["[-3/8]P"])) dataframe["DI_catch"] = np.where( dataframe["DI_values"] > dataframe["DI_cutoff"], 0, 1, ) dataframe["minima_sort_threshold"] = dataframe["&s-minima_sort_threshold"] dataframe["maxima_sort_threshold"] = dataframe["&s-maxima_sort_threshold"] dataframe['min_threshold_mean'] = dataframe["minima_sort_threshold"].expanding().mean() dataframe['max_threshold_mean'] = dataframe["maxima_sort_threshold"].expanding().mean() dataframe['maxima_check'] = dataframe['maxima'].rolling(3).apply(lambda x: int((x != 1).all()), raw=True).fillna(0) dataframe['minima_check'] = dataframe['minima'].rolling(3).apply(lambda x: int((x != 1).all()), raw=True).fillna(0) pair = metadata['pair'] if dataframe['maxima'].iloc[-3] == 1 and dataframe['maxima_check'].iloc[-1] == 0: self.dp.send_msg(f'*** {pair} *** Maxima Detected - Potential Short!!!' ) if dataframe['minima'].iloc[-3] == 1 and dataframe['minima_check'].iloc[-1] == 0: self.dp.send_msg(f'*** {pair} *** Minima Detected - Potential Long!!!' ) # CCI dataframe['cci2'] = ta.CCI(dataframe, timeperiod=170) dataframe['cci1'] = ta.CCI(dataframe, timeperiod=34) return dataframe def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[ ( (df["minima_check"].shift(1) == 0) & (df["minima_check"] == 0) & (df["&s-extrema"] < 0) & #(df['rsx'] < 50)& #(df['stc5'] < 5)& (df['volume'] > 0) # Make sure Volume is not 0 ), ['enter_long', 'enter_tag']] = (1, 'Minima Full Send') df.loc[ ( (df["minima_check"].shift(2) == 0) & (df["minima_check"] == 0) & (df["&s-extrema"] < 0) & #(df['rsx'] < 50)& #(df['stc5'] < 5)& (df['volume'] > 0) # Make sure Volume is not 0 ), ['enter_long', 'enter_tag']] = (1, 'Minima Full Send2') ''' df.loc[ ( (df["do_predict"] == 1) & # Guard: tema is raising (df["DI_catch"] == 1) & (df["maxima_check"] == 1) & (df["&s-extrema"] < 0) & (df["minima"].shift(1) == 1) & (df['volume'] > 0) # Make sure Volume is not 0 ), ['enter_long', 'enter_tag']] = (1, 'Minima') ### Comment Out this signal after Training is completed. df.loc[ ( (df["minima_check"] == 0) & (df['volume'] > 0) # Make sure Volume is not 0 ), ['enter_long', 'enter_tag']] = (1, 'Minima Full Send') df.loc[ ( (df["do_predict"] == 1) & # Guard: tema is raising (df["DI_catch"] == 1) & (df["minima_check"] == 0) & (df["minima_check"].shift(5) == 1) & (df['volume'] > 0) # Make sure Volume is not 0 ), ['enter_long', 'enter_tag']] = (1, 'Minima Check') ''' ########### short df.loc[ ( (df["maxima_check"].shift(1) == 0) & (df["maxima_check"] == 0) & (df["&s-extrema"] > 0) & #(df['rsx'] > 50)& #(df['stc5'] > 95)& (df['volume'] > 0) # Make sure Volume is not 0 ), ['enter_short', 'enter_tag']] = (1, 'Maxima Full Send') df.loc[ ( (df["maxima_check"].shift(2) == 0) & (df["maxima_check"] == 0) & (df["&s-extrema"] > 0) & #(df['rsx'] > 50)& #(df['stc5'] > 95)& (df['volume'] > 0) # Make sure Volume is not 0 ), ['enter_short', 'enter_tag']] = (1, 'Maxima Full Send2') # df.loc[ # ( # (df["do_predict"] == 1) & # Guard: tema is raising # (df["DI_catch"] == 1) & # (df["minima_check"] == 1) & # (df["&s-extrema"] > 0) & # (df["maxima"].shift(1) == 1) & # (df['volume'] > 0) # Make sure Volume is not 0 # ), # ['enter_short', 'enter_tag']] = (1, 'Maxima') # df.loc[ # ( # # (df["do_predict"] == 1) & # Guard: tema is raising # # (df["DI_catch"] == 1) & # # (df["maxima_check"] == 1) & # # (df["&s-extrema"] < 0) & # (df["maxima_check"] == 0) & # (df['volume'] > 0) # Make sure Volume is not 0 # ), # ['enter_short', 'enter_tag']] = (1, 'Maxima Full Send') # df.loc[ # ( # (df["do_predict"] == 1) & # Guard: tema is raising # (df["DI_catch"] == 1) & # (df["maxima_check"] == 0) & # (df["maxima_check"].shift(5) == 1) & # (df['volume'] > 0) # Make sure Volume is not 0 # ), # ['enter_short', 'enter_tag']] = (1, 'Maxima Check') return df def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[ ( (df["maxima_check"].shift(1) == 0) & (df["maxima_check"] == 0) & (df["&s-extrema"] > 0) & (df['volume'] > 0) # Make sure Volume is not 0 ), ['exit_long', 'exit_tag']] = (1, 'Maxima Check Complete') df.loc[ ( (df["maxima_check"].shift(2) == 0) & (df["maxima_check"] == 0) & (df["&s-extrema"] > 0) & (df['volume'] > 0) # Make sure Volume is not 0 ), ['exit_long', 'exit_tag']] = (1, 'Maxima Check Complete') # ##### short exits df.loc[ ( (df["minima_check"].shift(1) == 0) & (df["minima_check"] == 0) & (df["&s-extrema"] < 0) & (df['volume'] > 0) # Make sure Volume is not 0 ), ['exit_short', 'exit_tag']] = (1, 'Minima Check') df.loc[ ( (df["minima_check"].shift(2) == 0) & (df["minima_check"] == 0) & (df["&s-extrema"] < 0) & (df['volume'] > 0) # Make sure Volume is not 0 ), ['exit_short', 'exit_tag']] = (1, 'Minima Check') # df.loc[ # ( # (df["do_predict"] == 1) & # Guard: tema is raising # (df["DI_catch"] == 1) & # (df["&s-extrema"] < 0) & # (df["minima"].shift(1) == 1) & # (df['volume'] > 0) # Make sure Volume is not 0 # ), # ['exit_short', 'exit_tag']] = (1, 'Minima') # df.loc[ # ( # # (df["do_predict"] == 1) & # Guard: tema is raising # # (df["DI_catch"] == 1) & # # (df["&s-extrema"] > 0) & # (df["minima_check"] == 0) & # (df['volume'] > 0) # Make sure Volume is not 0 # ), # ['exit_short', 'exit_tag']] = (1, 'Minima Full Send') return df def top_percent_change(dataframe: DataFrame, length: int) -> float: """ Percentage change of the current close from the range maximum Open price :param dataframe: DataFrame The original OHLC dataframe :param length: int The length to look back """ if length == 0: return (dataframe['open'] - dataframe['close']) / dataframe['close'] else: return (dataframe['open'].rolling(length).max() - dataframe['close']) / dataframe['close'] def chaikin_mf(df, periods=20): close = df['close'] low = df['low'] high = df['high'] volume = df['volume'] mfv = ((close - low) - (high - close)) / (high - low) mfv = mfv.fillna(0.0) mfv *= volume cmf = mfv.rolling(periods).sum() / volume.rolling(periods).sum() return Series(cmf, name='cmf') def VWAPB(dataframe, window_size=20, num_of_std=1): df = dataframe.copy() df['vwap'] = qtpylib.rolling_vwap(df, window=window_size) rolling_std = df['vwap'].rolling(window=window_size).std() df['vwap_low'] = df['vwap'] - (rolling_std * num_of_std) df['vwap_high'] = df['vwap'] + (rolling_std * num_of_std) return df['vwap_low'], df['vwap'], df['vwap_high'] def get_distance(p1, p2): return abs((p1) - (p2)) def calculate_murrey_math_levels(df, window_size=64): #df = df.iloc[-window_size:] # Calculate rolling 64-bar maximum and minimum values rolling_max_H = df['high'].rolling(window=window_size).max() rolling_min_L = df['low'].rolling(window=window_size).min() max_H = rolling_max_H min_L = rolling_min_L range_HL = max_H - min_L def calculate_fractal(v2): fractal = 0 if 25000 < v2 <= 250000: fractal = 100000 elif 2500 < v2 <= 25000: fractal = 10000 elif 250 < v2 <= 2500: fractal = 1000 elif 25 < v2 <= 250: fractal = 100 elif 12.5 < v2 <= 25: fractal = 12.5 elif 6.25 < v2 <= 12.5: fractal = 12.5 elif 3.125 < v2 <= 6.25: fractal = 6.25 elif 1.5625 < v2 <= 3.125: fractal = 3.125 elif 0.390625 < v2 <= 1.5625: fractal = 1.5625 elif 0 < v2 <= 0.390625: fractal = 0.1953125 return fractal def calculate_octave(v1, v2, mn, mx): range_ = v2 - v1 sum_ = np.floor(np.log(calculate_fractal(v1) / range_) / np.log(2)) octave = calculate_fractal(v1) * (0.5 ** sum_) mn = np.floor(v1 / octave) * octave if mn + octave > v2: mx = mn + octave else: mx = mn + (2 * octave) return mx def calculate_x_values(v1, v2, mn, mx): dmml = (v2 - v1) / 8 x_values = [] # Calculate the midpoints of each segment midpoints = [mn + i * dmml for i in range(8)] for i in range(7): x_i = (midpoints[i] + midpoints[i + 1]) / 2 x_values.append(x_i) finalH = max(x_values) # Maximum of the x_values is the finalH return x_values, finalH def calculate_y_values(x_values, mn): y_values = [] for x in x_values: if x > 0: y = mn else: y = 0 y_values.append(y) return y_values def calculate_mml(mn, finalH, mx): dmml = ((finalH - finalL) / 8) * 1.0699 mml = (float([mx][0]) * 0.99875) + (dmml * 3) # mml = (float([mx]) * 0.99875) + (dmml * 3) ml = [] for i in range(0, 16): calc = mml - (dmml * (i)) ml.append(calc) murrey_math_levels = { "[-3/8]P": ml[14], "[-2/8]P": ml[13], "[-1/8]P": ml[12], "[0/8]P": ml[11], "[1/8]P": ml[10], "[2/8]P": ml[9], "[3/8]P": ml[8], "[4/8]P": ml[7], "[5/8]P": ml[6], "[6/8]P": ml[5], "[7/8]P": ml[4], "[8/8]P": ml[3], "[+1/8]P": ml[2], "[+2/8]P": ml[1], "[+3/8]P": ml[0] } return mml, murrey_math_levels for i in range(len(df)): mn = np.min(min_L.iloc[:i + 1]) mx = np.max(max_H.iloc[:i + 1]) x_values, finalH = calculate_x_values(mn, mx, mn, mx) y_values = calculate_y_values(x_values, mn) finalL = np.min(y_values) mml, murrey_math_levels = calculate_mml(finalL, finalH, mx) # Add Murrey Math levels to the DataFrame at each time step for level, value in murrey_math_levels.items(): df.at[df.index[i], level] = value return df def PC(dataframe, in1, in2): df = dataframe.copy() pc = ((in2-in1)/in1) * 100 return pc