# source: https://raw.githubusercontent.com/project237/my_freqtrade_scripts/bf56425d9e420e99463f88a3a594d9f4e5062e03/cindicatorAB_method2.py # testing new repo | 2022-06-28 16:37 """ ok, so this will be the strategy file where we are gonna be running for each single iteration of the backtest loop. 1- Inside populate_indicators, we simply use M_dt column to the find the correct 15 min candle and insert the columns "buy", price1-3 and indicator as new column to the backtest dataframe. All but first columns will have the same value for each row (TODO - workaround) and the column "buy" will insert a 1 for an otherwise 0 on whichever row matches the the M_dt 2- Inside populate_buy_trend we just instruct it to buy when the column "buy" is 1. 3- Inside populate_sell_trend we instruct it to sell whenever either price2 or price3 is within the candle range. # todo - check the script that calls these methods to see if you need to create a copy of df within the methods. """ ## --- Do not remove these libs --- from ctypes.wintypes import BOOL from freqtrade.strategy import IStrategy, merge_informative_pair from typing import Dict, List from functools import reduce # my imports / un-imports here import pandas as pd import arrow # from pandas import pd.DataFrame ## -------------------------------- def first_is_recent_or_eq(ts1, ts2, use_arrow=True): """ compares two unix timestamps, if the timetamps might be of different units, keep use_arrow unchanged """ if use_arrow: return arrow.get(ts1) >= arrow.get(ts2) return ts1 >= ts2 def return_current_signal_as_dict(signals_df_C, candle_T): """ signals_df_C - dataframe that contains parsed signal values indexed by datetime candle_T - unix timestamp of the current candle start time !!WARNING - assumes candle_T is the STARTING timestamp of a given candle TODO - CHECK ASSUMPTION ABOVE """ # METHOD 2 | 2022-07-01 16:07 - # !! Unlike the last one, this requires iteration, so will be instersting to see which one is # !! faster # 1- start from first candle and first signal, go over the signals to find the latest one # that is earlier than the candle itself. # 2- If no such signal is found, assign empty dict, otherwise assign signals' values to that of the candle # 3- If that one is not the earliest signal, delete from df. # This way we'll always be assigning the data from the last signal, and avoid unnecessary comparisons index_earlier = [] # goes from eraliest to latest, for index, row in signals_df_C.iterrows(): # if the signal is later or eq to the candle timestamp if first_is_recent_or_eq(candle_T, row["M_dt"], use_arrow=False): index_earlier.append(index) # if the signal is later than the candle timestamp else: break len_earlier = len(index_earlier) # no valid signals, return empty dict if len_earlier == 0: return {} # more than 1 signals that is earlier, drop the rest from df elif len_earlier > 1: signals_df_C.drop(index_earlier[:-1], inplace=True) # proceed without dropping if exactly 1 signal last_dict = signals_df_C.loc[index_earlier[-1]].to_dict() return last_dict class bt(): """ This is our custom bactest class, this needs to be initialized inside the strategy class | 2022-07-06 20:49 """ def __init__(self) -> None: self.last_buy_signal_id = None self.last_sell_signal_id = None self.buys_set = set() self.sells_set = set() def is_actual_buy(self, current_signal_id): """ Called by populate_buy_trend whenever the nominal buying condition holds. If the nominal buy condition is produced by a new signal (and not for another that has already been bought) returns true after updating last_buy_signal_id to current signal, otherwise returns false. """ new_signal = (self.last_buy_signal_id != current_signal_id) if new_signal: self.last_buy_signal_id = current_signal_id # add the current signal to the set of bought signals self.buys_set.add(current_signal_id) return 1 # pass since others will be pass as well else: return 0 def is_actual_sell(self, current_signal_id): """ Called by populate_sell_trend whenever the nominal selling condition holds. If the nominal sell condition is produced by a new signal (and not for another that has already been sold) returns true after updating last_sell_signal_id to current signal, otherwise returns false. """ # check if the signal has been bought before # we check the set of bought signals, if the current signal is in the set, then it has been bought # if not, it has not been bought has_been_bought = (current_signal_id in self.buys_set) # check if the signal has not been sold before new_sell_signal = (self.last_sell_signal_id != current_signal_id) if (new_sell_signal and has_been_bought): self.last_sell_signal_id = current_signal_id # add the current signal to the set of sold signals self.sells_set.add(current_signal_id) return 1 # pass since others will be pass as well else: return 0 class github_project237_my_freqtrade_scripts__cindicatorAB_method2__20220818_141153(IStrategy): # So all the methods below are called only once per backtest and each time during the bot loop for # live trading and dry run modes # keep this in mind.. # initialize the custom bt class bt_obj = bt() # An informative dictionary mapping ticker df column to their meanings - 2022-06-30 21:07 ticker_df_column_ref = { "timestamp": 0, "open" : 1, "high" : 2, "low" : 3, "close" : 4, "volume" : 5 } # parameters that will go to hyperopt my_params = { "buy_buffer" : 1.01, "sell_buffer": 1.01, "min_indicator": 85 # these two will be directly multiplied with the raw signal price } signal_file = "/home/u237/projects/parsing_cindicator/data/CND_AB_parsed_fix1.json" # we import our cindictor AB sorted df here for use in populate_indicators signals = pd.read_json(signal_file) # filter signals df for only our TICKER and those with indicator greater than min_indicator signal_df_indicator_filtered = signals.loc[((signals["ticker"] == "ZEC/USD") & (signals["indicator"] > my_params["min_indicator"])), ['base', 'above', 'below',"indicator", "M_dt", "Mid"]] signal_df = signal_df_indicator_filtered.copy() ## Minimal ROI designed for the strategy. ## This attribute will be overridden if the config file contains "minimal_roi" minimal_roi = { "60": 0.01, "30": 0.03, "20": 0.04, "0": 0.05 } ## Optimal stoploss designed for the strategy ## This attribute will be overridden if the config file contains "stoploss" # !! leaving this ratio well below what the exit signal should be doing, # !! in effect, we're not using this stoploss = -0.25 ## Optimal timeframe for the strategy timeframe = '1h' ## trailing stoploss trailing_stop = False trailing_stop_positive = 0.02 trailing_stop_positive_offset = 0.04 ## run "populate_indicators" only for new candle ta_on_candle = False ## Experimental settings (configuration will overide these if set) use_exit_signal = True ignore_roi_if_entry_signal = True ## Optional order type mapping # !! leave these as is since these don't affect the bt order_types = { 'buy' : 'limit', 'sell' : 'limit', 'stoploss' : 'market', 'stoploss_on_exchange': False } # |*| 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: pd.DataFrame ) -> pd.DataFrame: """ !! row.loc[0] - the candle timestamp """ dataframe["signal"] = dataframe.progress_apply(lambda row: return_current_signal_as_dict(self.signal_df, row.loc[0]), axis='columns') dataframe["base"] = dataframe.apply(lambda row: row["signal"].get("base"), axis="columns") dataframe["above"] = dataframe.apply(lambda row: row["signal"].get("above"), axis="columns") dataframe["below"] = dataframe.apply(lambda row: row["signal"].get("below"), axis="columns") dataframe["indicator"] = dataframe.apply(lambda row: row["signal"].get("indicator"), axis="columns") dataframe["M_dt"] = dataframe.apply(lambda row: row["signal"].get("M_dt"), axis="columns") dataframe["Mid"] = dataframe.apply(lambda row: row["signal"].get("Mid"), axis="columns") return dataframe # |*| def populate_buy_trend(self, dataframe: pd.DataFrame) -> pd.DataFrame: """ Based on TA indicators, populates the buy signal for the given dataframe :param dataframe: pd.DataFrame :return: pd.DataFrame with buy column """ # returns the signals that might be a buy, we will then iterate over this subset # to see which ones are actual buys based on "is_actual_buy" df_copy = dataframe.assign(buy = 0) # FILTER FOR THE NOMINAL BUY CONDITION # we esentially buy whenever the (base price + buy_buffer) is within low price and high price # (this will go to hyperopt later) # todo - remove unnecessary computation here df_nominal_buy = df_copy.loc[ (df_copy.loc[:,3] <= (df_copy['base'] * self.my_params["buy_buffer"])) & \ ( (df_copy['base'] * self.my_params["buy_buffer"]) <= df_copy.loc[:,2]) ] # df_copy.assign(buy = lambda row: bt1.is_actual_buy(row.Mid)) df_copy["buy"] = df_nominal_buy.progress_apply(lambda row: self.bt.is_actual_buy(row.Mid), axis='columns') return df_copy # |*| def populate_sell_trend(self, dataframe: pd.DataFrame ) -> pd.DataFrame: """ Based on TA indicators, populates the sell signal for the given dataframe :param dataframe: pd.DataFrame :return: pd.DataFrame with buy column """ # add a new column "sell" to dataframe, initialize it to 0 dataframe = dataframe.assign(sell = 0) # filter for the nominal sell condition # we sell whenever the either above or below price is within the low and high of the candle df_nominal_sell = dataframe.loc[ (dataframe['above'] * self.my_params["sell_buffer"]).between(dataframe.loc[:,3], dataframe.loc[:,2]) | (dataframe['below'] * self.my_params["sell_buffer"]).between(dataframe.loc[:,3], dataframe.loc[:,2]) ] dataframe["sell"] = df_nominal_sell.progress_apply(lambda row: self.bt.is_actual_sell(row.Mid), axis='columns') return dataframe def number_of_signal_never_bought(self) -> int: """ Returns the number of signals that have never been bought :return: int """ # todo - test this function # get the set of all signals all_signals = set(self.signal_df.Mid) # get the set of all bought signals bought_signals = self.bt.buys_set # get the set of all signals that have never been bought never_bought_signals = all_signals - bought_signals # return the number of signals that have never been bought return len(never_bought_signals) def number_of_signal_never_sold(self) -> int: """ gets the length of difference between self.bt.sells_set and self.bt.buys_set :return: int """ # todo - test this function return len(self.bt.buys_set - self.bt.sells_set) def display_results(self): min_indicator = self.my_params["min_indicator"] #print("======THE BACKTEST OF IS OVER======") #print(f"The number of signals above {min_indicator} is {len(self.signal_df_indicator_filtered)}") # todo - sort the rest of these out #print(f"The number of total buys and sells is {len(df_sells)}") #print(f"The number of signals never bought is {number_of_signal_never_bought(signal_df_indicator_filtered, bt1)}") #print(f"The number of signals never sold is {number_of_signal_never_sold(bt1)}") #print("=====================================")