# source: https://raw.githubusercontent.com/vaskosmihaylov/nfi-custom-strategies/b973914112d7ff4f3b170f74e2723696056447f8/user_data/strategies/Tank/Tank1Modulus.py import logging from functools import reduce import datetime import talib.abstract as ta import pandas_ta as pta import logging import os import numpy as np import pandas as pd #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_vaskosmihaylov_nfi_custom_strategies__Tank1Modulus__20260416_123132(IStrategy): ''' ______ __ __ __ __ ______ __ __ __ __ __ ______ / \ / | _/ | / | / | / \ / \ / | / | / | / | / \ /$$$$$$ |$$ |____ / $$ | _______ $$ | /$$/ /$$$$$$ |$$ \ $$ | _$$ |_ $$ | $$ | _______ /$$$$$$ | _______ $$ | $$/ $$ \ $$$$ | / |$$ |/$$/ $$ ___$$ |$$$ \$$ | / $$ | $$ |__$$ | / |$$$ \$$ | / | $$ | $$$$$$$ | $$ | /$$$$$$$/ $$ $$< / $$< $$$$ $$ | $$$$$$/ $$ $$ |/$$$$$$$/ $$$$ $$ |/$$$$$$$/ $$ | __ $$ | $$ | $$ | $$ | $$$$$ \ _$$$$$ |$$ $$ $$ | $$ | __$$$$$$$$ |$$ | $$ $$ $$ |$$ \ $$ \__/ |$$ | $$ | _$$ |_ $$ \_____ $$ |$$ \ / \__$$ |$$ |$$$$ | $$ |/ | $$ |$$ \_____ $$ \$$$$ | $$$$$$ | $$ $$/ $$ | $$ |/ $$ |$$ |$$ | $$ |$$ $$/ $$ | $$$ |______$$ $$/ $$ |$$ |$$ $$$/ / $$/ $$$$$$/ $$/ $$/ $$$$$$/ $$$$$$$/ $$/ $$/ $$$$$$/ $$/ $$// |$$$$/ $$/ $$$$$$$/ $$$$$$/ $$$$$$$/ $$$$$$/ ''' exit_profit_only = True ### No selling at a loss use_custom_stoploss = True trailing_stop = False ignore_roi_if_entry_signal = True process_only_new_candles = True can_short = False use_exit_signal = True startup_candle_count: int = 200 stoploss = -0.99 locked_stoploss = {} timeframe = '1h' # DCA position_adjustment_enable = True max_epa = IntParameter(0, 5, default = 0 ,space='buy', optimize=True, load=True) # Of additional buys. max_dca_multiplier = DecimalParameter(low=1.2, high=4.0, default=1.2, decimals=1 ,space='buy', optimize=True, load=True) safety_order_reserve = IntParameter(0, 10, default=0.5, space='buy', optimize=True) max_entry_position_adjustment = max_epa.value ### Custom Functions # Modulus peaks = IntParameter(30, 60, default=32, space='buy', optimize=True) ### initial smallest window bull_bear = IntParameter(80, 120, default=90, space='buy', optimize=True) trend = DecimalParameter(low=25, high=40, default=26.2, decimals=1 ,space='buy', optimize=True, load=True) volatility = DecimalParameter(low=30, high=50, default=34.6, decimals=1 ,space='buy', optimize=True, load=True) sensitivity = IntParameter(7, 15, default=12, space='buy', optimize=True, load=True) atr = IntParameter(3, 7, default=5, space='buy', optimize=True, load=True) window = IntParameter(12, 70, default=16, space='buy', optimize=True) mod = IntParameter(180, 200, default=196, space='buy', optimize=True) # 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 # 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) locked_stoploss = {} minimal_roi = { } plot_config = {} @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 ### # This is called when placing the initial order (opening trade) # Let unlimited stakes leave funds open for DCA orders 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 proposed_stake = proposed_stake / (self.max_dca_multiplier.value + self.safety_order_reserve.value) # Leaving some reserve incase the market dumps!!! # This also applies to fixed staked return proposed_stake 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[float]: 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 current_candle = dataframe.iloc[-1].squeeze() TP0 = current_candle['move_mean'] * 0.618 TP0_5 = current_candle['move_mean'] TP1 = current_candle['move_mean'] * 1.618 TP2 = current_candle['move_mean'] * 2.618 TP3 = current_candle['move_mean'] * 3.618 display_profit = current_profit * 100 if current_profit is not None: logger.info(f"{trade.pair} - Current Profit: {display_profit:.3}% # of Entries: {trade.nr_of_successful_entries}") # Take Profit if m00n if current_profit > TP1 and trade.nr_of_successful_exits == 0: # Take quarter of the profit at average move fib% return -(trade.stake_amount / 4) if current_profit > TP2 and trade.nr_of_successful_exits == 1: # Take quarter of the profit at next fib% return -(trade.stake_amount / 3) if current_profit > TP3 and trade.nr_of_successful_exits == 2: # Take half of the profit at last fib% return -(trade.stake_amount / 2) # Profit Based DCA if current_profit > -TP0 and trade.nr_of_successful_entries == 1: return None if current_profit > -TP1 and trade.nr_of_successful_entries == 2: return None if current_profit > -TP2 and trade.nr_of_successful_entries == 3: return None if current_profit > -TP3 and trade.nr_of_successful_entries == 4: return None if current_profit > -0.25 and trade.nr_of_successful_entries == 5: return None try: # This returns first order stake size # Modify the following parameters to enable more levels or different buy size: # max_entry_position_adjustment = 3 # max_dca_multiplier = 3.5 stake_amount = filled_entries[0].cost # This then calculates current safety order size if count_of_entries == 1: stake_amount = stake_amount * 2 elif count_of_entries == 2: stake_amount = stake_amount * 4 elif count_of_entries == 3: stake_amount = stake_amount * 8 elif count_of_entries == 4: stake_amount = stake_amount * 16 elif count_of_entries == 5: stake_amount = stake_amount * 32 else: stake_amount = stake_amount return stake_amount except Exception as exception: return None return None ### Trailing Stop ### 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 SLT1 = current_candle['move_mean'] if trade_duration > 720 and trade_duration < 1080: SL1 = current_candle['move_mean'] * 0.4 if trade_duration > 1080 and trade_duration < 1440: SL1 = current_candle['move_mean'] * 0.3 else: SL1 = current_candle['move_mean'] * 0.5 SLT2 = current_candle['move_mean_x'] SL2 = current_candle['move_mean_x'] - current_candle['move_mean'] display_profit = current_profit * 100 slt1 = SLT1 * 100 sl1 = SL1 * 100 slt2 = SLT2 * 100 sl2 = SL2 * 100 # if len(self.locked_stoploss) > 0: # print(self.locked_stoploss) if current_candle['max_l'] != 0: # ignore stoploss if setting new highs if pair not in self.locked_stoploss: # No locked stoploss for this pair yet if SLT2 is not None and current_profit > SLT2: self.locked_stoploss[pair] = SL2 self.dp.send_msg(f'*** {pair} *** Profit {display_profit:.3f}% - {slt2:.3f}%/{sl2:.3f}% activated') logger.info(f'*** {pair} *** Profit {display_profit:.3f}% - {slt2:.3f}%/{sl2:.3f}% activated') return SL2 elif SLT1 is not None and current_profit > SLT1: self.locked_stoploss[pair] = SL1 self.dp.send_msg(f'*** {pair} *** Profit {display_profit:.3f}% - {slt1:.3f}%/{sl1:.3f}% activated') logger.info(f'*** {pair} *** Profit {display_profit:.3f}% - {slt1:.3f}%/{sl1:.3f}% activated') return SL1 else: return self.stoploss else: # Stoploss has been locked for this pair self.dp.send_msg(f'*** {pair} *** Profit {display_profit:.3f}% stoploss locked at {self.locked_stoploss[pair]:.4f}') logger.info(f'*** {pair} *** Profit {display_profit:.3f}% stoploss locked at {self.locked_stoploss[pair]:.4f}') return self.locked_stoploss[pair] if current_profit < -.01: if pair in self.locked_stoploss: del self.locked_stoploss[pair] self.dp.send_msg(f'*** {pair} *** Stoploss reset.') logger.info(f'*** {pair} *** Stoploss reset.') return self.stoploss 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 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.0001: # 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: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() if exit_reason == 'roi' and (last_candle['max_l'] < 0.003): return False # Handle freak events if exit_reason == 'roi' and trade.calc_profit_ratio(rate) < 0.003: logger.info(f"{trade.pair} ROI is below 0") self.dp.send_msg(f'{trade.pair} ROI is below 0') return False 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 populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata['pair'] dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["mfi"] = (ta.MFI(dataframe, timeperiod=89) - 50) * 2 dataframe["roc"] = ta.ROCR(dataframe, timeperiod=89) 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['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']) # 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(100).apply(lambda x: np.where(x >= 1, 1, 0).sum()).shift() dataframe['candle_2perc_50'] = dataframe['perc_change'].rolling(100).apply(lambda x: np.where(x >= 2, 1, 0).sum()).shift() dataframe['candle_3perc_50'] = dataframe['perc_change'].rolling(100).apply(lambda x: np.where(x >= 3, 1, 0).sum()).shift() dataframe['candle_5perc_50'] = dataframe['perc_change'].rolling(100).apply(lambda x: np.where(x >= 5, 1, 0).sum()).shift() dataframe['candle_-1perc_50'] = dataframe['perc_change'].rolling(100).apply(lambda x: np.where(x <= -1, -1, 0).sum()).shift() dataframe['candle_-2perc_50'] = dataframe['perc_change'].rolling(100).apply(lambda x: np.where(x <= -2, -1, 0).sum()).shift() dataframe['candle_-3perc_50'] = dataframe['perc_change'].rolling(100).apply(lambda x: np.where(x <= -3, -1, 0).sum()).shift() dataframe['candle_-5perc_50'] = dataframe['perc_change'].rolling(100).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 # Modulus 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 # # 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 - 9/12 ap = (0.333 * (heikinashi['high'] + heikinashi['low'] + heikinashi["close"])) dataframe['esa'] = ta.EMA(ap, timeperiod = 10) dataframe['d'] = ta.EMA(abs(ap - dataframe['esa']), timeperiod = 10) dataframe['wave_ci'] = (ap-dataframe['esa']) / (0.015 * dataframe['d']) dataframe['wave_t1'] = ta.EMA(dataframe['wave_ci'], timeperiod = 21) dataframe['wave_t2'] = ta.SMA(dataframe['wave_t1'], timeperiod = 4) dataframe.loc[dataframe['wave_t1'] > 0, "wave_t1_UP"] = dataframe['wave_t1'] dataframe.loc[dataframe['wave_t1'] < 0, "wave_t1_DN"] = dataframe['wave_t1'] dataframe['wave_t1_UP'].ffill() dataframe['wave_t1_DN'].ffill() dataframe['wave_t1_MEAN_UP'] = dataframe['wave_t1_UP'].mean() dataframe['wave_t1_MEAN_DN'] = dataframe['wave_t1_DN'].mean() dataframe['wave_t1_UP_FIB'] = dataframe['wave_t1_MEAN_UP'] * 1.618 dataframe['wave_t1_DN_FIB'] = dataframe['wave_t1_MEAN_DN'] * 1.618 # 200 SMA and distance dataframe['200sma'] = ta.SMA(dataframe, timeperiod = 200) dataframe['200sma_dist'] = get_distance(heikinashi["close"], dataframe['200sma']) dataframe['sma'] = ta.EMA(dataframe, timeperiod=self.mod.value) dataframe['sma_pc'] = abs((dataframe['sma'] - dataframe['sma'].shift()) / dataframe['sma']) * 100 dataframe['atr_pcnt'] = (qtpylib.atr(dataframe, window = self.atr.value)) / dataframe['ha_close'] dataframe['modulation'] = 1 + (dataframe['sma_pc'] * self.trend.value) + (dataframe['atr_pcnt'] * self.volatility.value) # Set minimum and maximum window size min_window = self.peaks.value max_window = self.bull_bear.value dataframe['order'] = (dataframe['modulation'] * self.peaks.value).round().fillna(self.bull_bear.value).astype(int) dataframe['order'] = np.where(dataframe['order'] > max_window, max_window, dataframe['order']) dataframe['order'] = np.where(dataframe['order'] < min_window, min_window, dataframe['order']) if not dataframe['order'].empty: order = dataframe['order'].iloc[-1] else: order = self.bear.value dataframe['zero'] = 0 # ### Manually Overriding order for testing... # order = 48 # print(pair, len(dataframe)) dataframe['extrema'] = 0 # Find the local minima and maxima min_peaks = argrelextrema(dataframe["low"].values, np.less, order=order) max_peaks = argrelextrema(dataframe["high"].values, np.greater, order=order) # Update the "extrema" column for minima for mp in min_peaks[0]: dataframe.at[mp, "extrema"] = -1 # Update the "extrema" column for maxima for mp in max_peaks[0]: dataframe.at[mp, "extrema"] = 1 # Save extrema to pickle file. if (self.dp.runmode.value in ('live', 'dry_run')): base = pair.split('/') path = os.path.join('pkl', f"{base[0]}.pkl") if os.path.exists(path): df_pkl = pd.read_pickle(path) # print(df_pkl.tail(5)) else: dataframe.to_pickle(path) df_pkl = dataframe.copy() for i in range(len(dataframe)): if i < 2: continue common_indices = dataframe.index.intersection(df_pkl.index) # Create a boolean mask where 'extrema' is not equal to 0 in df_pkl for common indices mask = (df_pkl.loc[common_indices, 'extrema'] != 0) # Use this mask to perform the assignment dataframe.loc[common_indices[mask], 'extrema'] = df_pkl.loc[common_indices[mask], 'extrema'] dataframe.to_pickle(path) # print(pair, len(df_pkl), len(dataframe)) # Update "minima" and "maxima" columns based on "extrema" dataframe["minima"] = np.where(dataframe["extrema"] == -1, 1, 0) dataframe["maxima"] = np.where(dataframe["extrema"] == 1, 1, 0) dataframe['max'] = dataframe["close"].rolling(order).max()/dataframe["close"] - 1 dataframe['min'] = abs(dataframe["close"].rolling(order).min()/dataframe["close"] - 1) dataframe['mm_width'] = dataframe['max'] - dataframe['min'] dataframe['atr_threshold'] = dataframe['atr_pcnt'].rolling(order).max() 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) if dataframe['maxima'].iloc[-1] == 1 and dataframe['maxima_check'].iloc[-1] == 0: self.dp.send_msg(f'*** {pair} *** Maxima Detected {order} - Potential Short!!!' ) if dataframe['minima'].iloc[-1] == 1 and dataframe['minima_check'].iloc[-1] == 0: self.dp.send_msg(f'*** {pair} *** Minima Detected {order} - Potential Long!!!' ) dataframe['OHLC4'] = (dataframe['open'] + dataframe['high'] + dataframe['low'] + dataframe['close']) / 4 # Check how far we are from min and max dataframe['max'] = dataframe['OHLC4'].rolling(4).max() / dataframe['OHLC4'] - 1 dataframe['min'] = abs(dataframe['OHLC4'].rolling(4).min() / dataframe['OHLC4'] - 1) dataframe['max_l'] = dataframe['OHLC4'].rolling(48).max() / dataframe['OHLC4'] - 1 dataframe['min_l'] = abs(dataframe['OHLC4'].rolling(48).min() / dataframe['OHLC4'] - 1) dataframe['max_x'] = dataframe['OHLC4'].rolling(336).max() / dataframe['OHLC4'] - 1 dataframe['min_x'] = abs(dataframe['OHLC4'].rolling(336).min() / dataframe['OHLC4'] - 1) # Apply rolling window operation to the 'OHLC4'column rolling_window = dataframe['OHLC4'].rolling(self.window.value) rolling_max = rolling_window.max() rolling_min = rolling_window.min() # Calculate the peak-to-peak value on the resulting rolling window data ptp_value = rolling_window.apply(lambda x: np.ptp(x)) # Assign the calculated peak-to-peak value to the DataFrame column dataframe['move'] = ptp_value / dataframe['OHLC4'] dataframe['move_mean'] = dataframe['move'].mean() dataframe['move_mean_x'] = dataframe['move'].mean() * 1.6 dataframe['exit_mean'] = rolling_min * (1 + dataframe['move_mean']) dataframe['exit_mean_x'] = rolling_min * (1 + dataframe['move_mean_x']) dataframe['enter_mean'] = rolling_max * (1 - dataframe['move_mean']) dataframe['enter_mean_x'] = rolling_max * (1 - dataframe['move_mean_x']) dataframe['atr_pcnt'] = (ta.ATR(dataframe, timeperiod=5) / dataframe['OHLC4']) dataframe['200sma_up'] = dataframe['200sma'] * (1 + dataframe['move_mean']) dataframe['200sma_dn'] = dataframe['200sma'] * (1 - dataframe['move_mean']) # Apply rolling window operation to the 'OHLC4'column rolling_window_x = dataframe['OHLC4'].rolling(200) rolling_max_x = rolling_window_x.max() rolling_min_x = rolling_window_x.min() # Calculate the peak-to-peak value on the resulting rolling window data ptp_value_x = rolling_window_x.apply(lambda x: np.ptp(x)) # Assign the calculated peak-to-peak value to the DataFrame column dataframe['move_l'] = ptp_value_x / dataframe['OHLC4'] dataframe['move_mean_l'] = dataframe['move_l'].mean() dataframe['move_mean_xl'] = dataframe['move_l'].mean() * 1.6 dataframe['exit_mean_l'] = rolling_min_x * (1 + dataframe['move_mean_l']) dataframe['exit_mean_xl'] = rolling_min_x * (1 + dataframe['move_mean_xl']) dataframe['enter_mean_l'] = rolling_max_x * (1 - dataframe['move_mean_l']) dataframe['enter_mean_xL'] = rolling_max_x * (1 - dataframe['move_mean_xl']) return dataframe def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: full_send1 = ( (df["minima_check"] == 0) & (df["minima_check"].shift(3) == 1) & (df["wave_t1"] < df["wave_t1_DN_FIB"]) & (df['volume'] > 0) # Make sure Volume is not 0 ) df.loc[full_send1, 'enter_long'] = 1 df.loc[full_send1, 'enter_tag'] = 'Full Send 1' full_send2 = ( (df["move"] > df['move_mean']) & (df["OHLC4"] < df['200sma']) & (df["wave_t1"] < df["wave_t1_DN_FIB"]) & (df['volume'] > 0) # Make sure Volume is not 0 ) df.loc[full_send2, 'enter_long'] = 1 df.loc[full_send2, 'enter_tag'] = 'Full Send 2' full_send3 = ( (df["minima_check"] == 0) & (df["minima_check"].shift(3) == 1) & (df['order'] == self.bull_bear.value) & (df["wave_t1"] < df["wave_t1_DN_FIB"]) & (df['volume'] > 0) # Make sure Volume is not 0 ) df.loc[full_send3, 'enter_long'] = 1 df.loc[full_send3, 'enter_tag'] = 'Full Send 3' full_send4 = ( (df["minima_check"] == 0) & (df["minima_check"].shift(3) == 1) & (df["wave_t1"] < df["wave_t1_DN_FIB"]) & (df['order'] < self.bull_bear.value) & (df['order'] > (self.bull_bear.value / 2)) & (df['volume'] > 0) # Make sure Volume is not 0 ) df.loc[full_send4, 'enter_long'] = 1 df.loc[full_send4, 'enter_tag'] = 'Full Send 4' full_send5 = ( (df["minima_check"] == 0) & (df["minima_check"].shift(3) == 1) & (df["wave_t1"] < df["wave_t1_MEAN_DN"]) & (df['order'] <= (self.bull_bear.value / 2)) & (df['order'] >= self.peaks.value) & (df['volume'] > 0) # Make sure Volume is not 0 ) df.loc[full_send4, 'enter_long'] = 1 df.loc[full_send4, 'enter_tag'] = 'Full Send 5' return df def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: profit_taker1 = ( (df["maxima_check"] == 0) & (df["wave_t1"] > df["wave_t1_UP_FIB"]) & (df['volume'] > 0) # Make sure Volume is not 0 ) df.loc[profit_taker1, 'exit_long'] = 1 df.loc[profit_taker1, 'exit_tag'] = 'Profit Taker1' profit_taker2 = ( (df["maxima_check"] == 0) & (df["maxima_check"].shift(3) == 1) & (df['order'] == self.bull_bear.value) & (df["wave_t1"] > df["wave_t1_UP_FIB"]) & (df['volume'] > 0) # Make sure Volume is not 0 ) df.loc[profit_taker2, 'exit_long'] = 1 df.loc[profit_taker2, 'exit_tag'] = 'Profit Taker2' profit_taker3 = ( (df["maxima_check"] == 0) & (df["maxima_check"].shift(3) == 1) & (df["wave_t1"] > df["wave_t1_UP_FIB"]) & (df['order'] < self.bull_bear.value) & (df['order'] > (self.bull_bear.value / 2)) & (df['volume'] > 0) # Make sure Volume is not 0 ) df.loc[profit_taker3, 'exit_long'] = 1 df.loc[profit_taker3, 'exit_tag'] = 'Profit Taker3' profit_taker4 = ( (df["maxima_check"] == 0) & (df["maxima_check"].shift(3) == 1) & (df["wave_t1"] > df["wave_t1_UP_FIB"]) & (df['order'] <= (self.bull_bear.value / 2)) & (df['order'] >= self.peaks.value) & (df['volume'] > 0) # Make sure Volume is not 0 ) df.loc[profit_taker4, 'exit_long'] = 1 df.loc[profit_taker4, 'exit_tag'] = 'Profit Taker4' 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 (p1) - (p2) def PC(dataframe, in1, in2): df = dataframe.copy() pc = ((in2-in1)/in1) * 100 return pc