# source: https://raw.githubusercontent.com/BillGatesIII/Vires-in-Numeris/8b2428fcfe48a80049f19d24649087d38ebc29a3/vin.py from heapq import nlargest from itertools import count from math import ceil from pandas.core.series import Series from sqlalchemy import false from freqtrade.strategy import IStrategy, informative from freqtrade.exchange import timeframe_to_prev_date from freqtrade.persistence import Trade import logging import numpy as np from pandas import DataFrame, concat from functools import reduce from datetime import datetime, timedelta import locale locale.setlocale(category=locale.LC_ALL, locale='') log = logging.getLogger(__name__) class github_BillGatesIII_Vires_in_Numeris__vin__20220218_143232(IStrategy): INTERFACE_VERSION = 2 def version(self) -> str: return 'v5.0.1' min_day_listed: int = 5 top_volume: int = 80 df_market = DataFrame() df_top_vol = DataFrame() df_low = DataFrame() df_upp = DataFrame() custom_market_info = {} lb_market = (1, 2, 12) lb_buy_low = range(24, 49) lb_buy_upp = range(10, 35) minimal_roi = {"0": 100} stoploss = -1 stoploss_on_exchange = False trailing_stop = False use_custom_stoploss = False timeframe = '5m' process_only_new_candles = True use_sell_signal = True sell_profit_only = False startupp_candle_count: int = 48 @property def protections(self): return [ { "method": "CooldownPeriod", "stop_duration_candles": 36 } ] @informative('1d') def populate_indicators_1d(self, df: DataFrame, metadata: dict) -> DataFrame: i = self.min_day_listed df['candle_count'] = df['volume'].rolling(window=i, min_periods=i).count() return df def populate_indicators(self, df: DataFrame, metadata: dict) -> DataFrame: df['pair'] = metadata['pair'] df['green'] = df['close'].div(df['open']).ge(1) # bodysize = df['close'].div(df['open']).where(df['green'], df['open'].div(df['close'])) top = df['close'].where(df['green'], df['open']) bot = df['open'].where(df['green'], df['close']) # hi_tail = df['high'].div(top) # lo_tail = df['low'].rdiv(bot) # hi_adj = df['high'].div(hi_tail.pow(0.5)) # lo_adj = df['low'].mul(lo_tail.pow(0.5)) # hi_tail = df['high'].div(top) # lo_tail = df['low'].rdiv(bot) mid = top.add(bot).div(2) hi_tail_ratio = df['high'].sub(top).div(mid) hi_adj = top.add(pow(1 + hi_tail_ratio, 0.25).mul(df['high'])).div(2) lo_tail_ratio = df['low'].sub(bot).div(mid) lo_adj = bot.add(pow(1 - lo_tail_ratio, 0.25).mul(df['low'])).div(2) # hi_adj = df['close'].where(df['green'], df['open']) + (df['high'] - df['close']).where(df['green'], (df['high'] - df['open'])) / bodysize.pow(0.5) # low_adj = df['open'].where(df['green'], df['close']) - (df['open'] - df['low']).where(df['green'], (df['close'] - df['low'])) / bodysize.pow(0.5) # df['hlc3'] = hi_adj.add(low_adj).add(df['close']).div(3) df['hl2'] = hi_adj.add(lo_adj).div(2) # df['ohlc4'] = df['open'].add(hi_adj).add(lo_adj).add(df['close']).div(4) # hoc2 = hi_adj.add(df['close']).div(2) # lc2 = lo_adj.add(df['close']).div(2) df['hi_top'] = hi_adj.div(top) df['lo_bot'] = lo_adj.div(bot) # df['ho2o'] = hi_adj.add(df['open']).div(2).div(df['open']) # df_hlc3_ch: DataFrame = df['hlc3'].sub(df['hlc3'].shift(1)) # df_ohlc4_ch: DataFrame = df['ohlc4'].sub(df['ohlc4'].shift(1)) df_hl2_ch: DataFrame = df['hl2'].sub(df['hl2'].shift(1)) s = (1, 2, 3) for i in s: # df['updown'] = np.where(df_hlc3_ch.rolling(window=i, min_periods=i).sum().gt(0), 1, np.where(df_hlc3_ch.rolling(window=i, min_periods=i).sum().lt(0), -1, 0)) df['updown'] = np.where(df_hl2_ch.rolling(window=i, min_periods=i).sum().gt(0), 1, np.where(df_hl2_ch.rolling(window=i, min_periods=i).sum().lt(0), -1, 0)) df[f"str_{i}"] = df['updown'].groupby((df['updown'].ne(df['updown'].shift(1))).cumsum()).cumsum() df['str_min'] = df[[f"str_{i}" for i in s]].min(axis=1) df['str_max'] = df[[f"str_{i}" for i in s]].max(axis=1) # df['str_min_lc2_ch'] = lc2.div(lc2.to_numpy()[df.index.to_numpy() - df['str_min'].abs().to_numpy()]) # df['str_max_lc2_ch'] = lc2.div(lc2.to_numpy()[df.index.to_numpy() - df['str_max'].abs().to_numpy()]) # df['str_min_hc2_ch'] = hc2.div(hc2.to_numpy()[df.index.to_numpy() - df['str_min'].abs().to_numpy()]) # df['str_max_hc2_ch'] = hc2.div(hc2.to_numpy()[df.index.to_numpy() - df['str_max'].abs().to_numpy()]) df['str_min_hl2_ch'] = df['hl2'].div(df['hl2'].to_numpy()[df.index.to_numpy() - df['str_min'].abs().to_numpy()]) df['str_max_hl2_ch'] = df['hl2'].div(df['hl2'].to_numpy()[df.index.to_numpy() - df['str_max'].abs().to_numpy()]) # df['str_min_hlc3_ch'] = df['hlc3'].div(df['hlc3'].to_numpy()[df.index.to_numpy() - df['str_min'].abs().to_numpy()]) # df['str_max_hlc3_ch'] = df['hlc3'].div(df['hlc3'].to_numpy()[df.index.to_numpy() - df['str_max'].abs().to_numpy()]) # j = {*self.lb_buy_low, *self.lb_buy_upp} # for i in j: # df[f"vol_chg_{i}"] = df['volume'] / df['volume'].shift(1).rolling(window=i, min_periods=i).mean() # hlc3_pct = df['hlc3'].pct_change(i) # pct_mean: DataFrame = hlc3_pct.rolling(window=i, min_periods=i).mean() # pct_std: DataFrame = hlc3_pct.rolling(window=i, min_periods=i).std() # bb_pct_low = pct_mean.sub(pct_std.mul(2)) # bb_pct_upp = pct_mean.add(pct_std.mul(2)) # df[f"bb_w_{i}"] = bb_pct_up.sub(bb_pct_lo) # df[f"bb_low_{i}"] = hlc3_pct.sub(bb_pct_lo) # df[f"bb_upp_{i}"] = hlc3_pct.sub(bb_pct_up) # df['vol_chg_max'] = df[[f"vol_chg_{i}" for i in j]].max(axis=1) # df['bb_w_min'] = df[[f"bb_w_{i}" for i in j]].min(axis=1) # df['bb_w_max'] = df[[f"bb_w_{i}" for i in j]].max(axis=1) # df['bb_low_min'] = df[[f"bb_low_{i}" for i in j]].min(axis=1) # df['bb_upp_min'] = df[[f"bb_upp_{i}" for i in j]].min(axis=1) # df['bb_low_max'] = df[[f"bb_low_{i}" for i in j]].max(axis=1) # df['bb_upp_max'] = df[[f"bb_upp_{i}" for i in j]].max(axis=1) ef = [df] if __name__ not in ('upp', 'vol'): for i in self.lb_buy_low: ef.append(DataFrame({f"vol_chg_{i}": df['volume'] / df['volume'].shift(1).rolling(window=i, min_periods=i).mean()})) # df[f"lc2_pct_{i}"] = df['lc2'].pct_change(i) # pct_mean: DataFrame = df[f"lc2_pct_{i}"].rolling(window=i, min_periods=i).mean() # pct_std: DataFrame = df[f"lc2_pct_{i}"].rolling(window=i, min_periods=i).std() lo_pct = lo_adj.pct_change(i) pct_mean: DataFrame = lo_pct.rolling(window=i, min_periods=i).mean() pct_std: DataFrame = lo_pct.rolling(window=i, min_periods=i).std() bb_pct_upp = pct_mean.add(pct_std.mul(2)) bb_pct_low = pct_mean.sub(pct_std.mul(2)) ef.append(DataFrame({f"bb_w_low_{i}": bb_pct_upp.sub(bb_pct_low)})) # df[f"bb_low_{i}"] = df[f"lc2_pct_{i}"].sub(bb_pct_lo) ef.append(DataFrame({f"bb_low_{i}": lo_pct.sub(bb_pct_low)})) df = concat(ef, axis=1) df['vol_chg_max'] = df[[f"vol_chg_{i}" for i in self.lb_buy_low]].max(axis=1) df['bb_w_low_min'] = df[[f"bb_w_low_{i}" for i in self.lb_buy_low]].min(axis=1) df['bb_w_low_max'] = df[[f"bb_w_low_{i}" for i in self.lb_buy_low]].max(axis=1) df['bb_low_min'] = df[[f"bb_low_{i}" for i in self.lb_buy_low]].min(axis=1) df['bb_low_max'] = df[[f"bb_low_{i}" for i in self.lb_buy_low]].max(axis=1) if __name__ not in ('low', 'vol'): for i in self.lb_buy_upp: ef.append(DataFrame({f"vol_chg_{i}": df['volume'] / df['volume'].shift(1).rolling(window=i, min_periods=i).mean()})) # df[f"hc2_pct_{i}"] = df['hc2'].pct_change(i) # pct_mean: DataFrame = df[f"hc2_pct_{i}"].rolling(window=i, min_periods=i).mean() # pct_std: DataFrame = df[f"hc2_pct_{i}"].rolling(window=i, min_periods=i).std() hi_pct = hi_adj.pct_change(i) pct_mean: DataFrame = hi_pct.rolling(window=i, min_periods=i).mean() pct_std: DataFrame = hi_pct.rolling(window=i, min_periods=i).std() bb_pct_upp = pct_mean.add(pct_std.mul(2)) bb_pct_low = pct_mean.sub(pct_std.mul(2)) ef.append(DataFrame({f"bb_w_upp_{i}": bb_pct_upp.sub(bb_pct_low)})) # df[f"bb_upp_{i}"] = df[f"hc2_pct_{i}"].sub(bb_pct_up) ef.append(DataFrame({f"bb_upp_{i}": hi_pct.sub(bb_pct_upp)})) df = concat(ef, axis=1) df['vol_chg_max'] = df[[f"vol_chg_{i}" for i in self.lb_buy_upp]].max(axis=1) df['bb_w_upp_min'] = df[[f"bb_w_upp_{i}" for i in self.lb_buy_upp]].min(axis=1) df['bb_w_upp_max'] = df[[f"bb_w_upp_{i}" for i in self.lb_buy_upp]].max(axis=1) df['bb_upp_min'] = df[[f"bb_upp_{i}" for i in self.lb_buy_upp]].min(axis=1) df['bb_upp_max'] = df[[f"bb_upp_{i}" for i in self.lb_buy_upp]].max(axis=1) # hlc3_vol = df['hlc3'].mul(df['volume']) # ohlc4_vol = df['ohlc4'].mul(df['volume']) hl2_vol = df['hl2'].mul(df['volume']) j = self.lb_market for i in j: df[f"hl2_vol_{i}"] = hl2_vol.rolling(window=i, min_periods=i).sum() df[f"hl2_pct_{i}"] = df['hl2'].pct_change(i) self.df_market = concat([self.df_market, df[['date', 'pair', 'hl2', 'str_min', 'str_min_hl2_ch', 'str_max', 'str_max_hl2_ch', 'volume', f'hl2_vol_{j[0]}', f'hl2_vol_{j[1]}', f'hl2_vol_{j[2]}', f'hl2_pct_{j[0]}', f'hl2_pct_{j[1]}', f'hl2_pct_{j[2]}']]], ignore_index=True, copy=False) for i in j: df.drop(columns=[f"hl2_vol_{i}", f"hl2_pct_{i}"], inplace=True) df.drop(columns=['updown'], inplace=True) if len(self.dp.current_whitelist()) == 1: if __name__ not in ('upp', 'vol'): f = df['str_min'].between(-20, -1) & df['str_min_hl2_ch'].le(0.98) & df['bb_low_min'].lt(0) #& df['lc2c'].le(0.997) & df['vol_chg_max'].ge(0.5) & df['str_max'].le(1) #print(df.loc[f, ['pair', 'date', 'green', 'str_min_hl2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'bb_w_low_min', 'bb_w_low_max', 'bb_low_min', 'bb_low_max', 'close', 'hl2', 'volume']].to_string()) # #print(df.loc[f, [f"vol_chg_{i}" for i in self.lb_buy_low]].to_string()) # #print(df.loc[f, [f"bb_low_{i}" for i in self.lb_buy_low]].to_string()) # #print(df.loc[:, [f"pct_{i}" for i in self.lb_buy_low]].to_string()) # #print(df.loc[:, [f"bb_w_{i}" for i in self.lb_buy_low]].to_string()) if __name__ not in ('low', 'vol'): f = df['str_max'].between(1, 4) #& df['vol_chg_max'].ge(1) & df['str_min'].ge(1) & df['hc2c'].le(1.002) & df['bb_upp_max'].gt(0) #print(df.loc[:, ['pair', 'date', 'green', 'str_min_hl2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'bb_w_upp_min', 'bb_w_upp_max', 'bb_upp_min', 'bb_upp_max', 'close', 'hl2', 'volume']].to_string()) # #print(df.loc[f, [f"vol_chg_{i}" for i in self.lb_buy_upp]].to_string()) # #print(df.loc[f, [f"bb_upp_{i}" for i in self.lb_buy_upp]].to_string()) # #print(df.loc[:, [f"hc2_pct_{i}" for i in self.lb_buy_upp]].to_string()) # #print(df.loc[:, [f"bb_w_upp_{i}" for i in self.lb_buy_upp]].to_string()) else: b = 12 # f = df['ohlc4'].shift(-b).div(df['ohlc4']).le(0.95) f = df['close'].pct_change(b).le(-0.08) if not df.loc[f, :].empty: l = len(df.loc[f, :]) d = df.loc[f, 'date'].iloc[0] e1 = d - timedelta(minutes = b * 5) e2 = e1 e3 = e2 e4 = e3 if l >= 2: d = df.loc[f, 'date'].iloc[-1] e2 = d - timedelta(minutes = b * 5) if l >= 4: d = df.loc[f, 'date'].iloc[l // 2 - 1] e3 = d - timedelta(minutes = b * 5) d = df.loc[f, 'date'].iloc[-l // 2] e4 = d - timedelta(minutes = b * 5) if __name__ not in ('upp', 'vol'): f = [ ((df['date'] == e1) | (df['date'] == e2) | (df['date'] == e3) | (df['date'] == e4)), df['candle_count_1d'].ge(self.min_day_listed), df['str_min'].le(1), df['str_max'].le(2), df['str_min_hl2_ch'].between(0.795, 0.995), # df['str_max_hl2_ch'].between(0.8, 1), # df['lo_bot'].between(0.91, 0.99), df['bb_w_low_min'].ge(0.02), df['bb_w_low_max'].le(0.22), # df['bb_low_min'].ge(-0.4), # df['bb_low_max'].le(0.1), ] f = reduce(lambda x, y: x & y, f) if not df.loc[f].empty: # #print(df.loc[f, ['pair', 'date', 'green', 'str_min_hl2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'bb_w_low_min', 'bb_w_low_max', 'bb_low_min', 'bb_low_max', 'hl2', 'open', 'high', 'low', 'close', 'volume']].to_string()) self.df_low = concat([self.df_low, df.loc[f, ['pair', 'date', 'green', 'str_min_hl2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'bb_w_low_min', 'bb_w_low_max', 'bb_low_min', 'bb_low_max', 'hl2', 'open', 'high', 'low', 'close', 'volume']]]) if __name__ not in ('low', 'vol'): f = [ ((df['date'] == e1) | (df['date'] == e2) | (df['date'] == e3) | (df['date'] == e4)), df['candle_count_1d'].ge(self.min_day_listed), df['str_max'].ge(-1), df['str_min'].ge(-2), # df['str_min_hl2_ch'].between(1.01, 1.11), df['str_max_hl2_ch'].between(1.005, 1.135), df['hi_top'].between(1, 1.025), df['bb_w_upp_min'].ge(0.02), df['bb_w_upp_max'].le(0.22), # df['bb_upp_min'].ge(-0.1), df['bb_upp_max'].le(0.11), ] f = reduce(lambda x, y: x & y, f) if not df.loc[f].empty: # #print(df.loc[f, ['pair', 'date', 'green', 'str_min_hl2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'bb_w_upp_min', 'bb_w_upp_max', 'bb_upp_min', 'bb_upp_max', 'hl2', 'open', 'high', 'low', 'close', 'volume']].to_string()) self.df_upp = concat([self.df_upp, df.loc[f, ['pair', 'date', 'green', 'str_min_hl2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'bb_w_upp_min', 'bb_w_upp_max', 'bb_upp_min', 'bb_upp_max', 'hl2', 'open', 'high', 'low', 'close', 'volume']]]) return df def populate_buy_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[:, 'custom_buy_tag'] = '' # df.loc[:, 'buy_tag_bb_low'] = '' # df.loc[:, 'buy_tag_bb_upp'] = '' if __name__ not in ('upp', 'vol'): for i in self.lb_buy_low: # c = i / 300 - 0.01 buy_bb_low = [ df['candle_count_1d'].ge(self.min_day_listed), df['str_min'].le(1), df['str_max'].le(2), df['str_min_hl2_ch'].between(0.795, 0.995), # df['str_max_hl2_ch'].between(0.8, 1), # df['lo_bot'].between(0.91, 0.99), df['bb_w_low_min'].ge(0.02), df['bb_w_low_max'].le(0.22), df['bb_low_min'].ge(-0.11), # df['bb_low_max'].le(0.1), # df[f"lc2_pct_{i}"].between(-c, 0), # df[f"bb_w_low_{i}"].between(0.02, 0.32), # df[f"vol_chg_{i}"].between(0.5, 1.5), # df[f"bb_low_{i}"].shift(1).le(df[f"bb_low_{i}"]), df[f"bb_low_{i}"].shift(1).gt(0), df[f"bb_low_{i}"].between(-0.10, -0.02) ] df.loc[reduce(lambda x, y: x & y, buy_bb_low), 'custom_buy_tag'] += f"{i} " buy_bb_low = df['custom_buy_tag'].ne('') df.loc[buy_bb_low, 'custom_buy_tag'] = 'low ' + df['custom_buy_tag'].str.strip() if __name__ not in ('low', 'vol'): for i in self.lb_buy_upp: # c = i / 100 - 0.01 buy_bb_upp = [ df['candle_count_1d'].ge(self.min_day_listed), df['str_max'].ge(-1), df['str_min'].ge(-2), # df['str_min_hl2_ch'].between(1, 1.08), df['str_max_hl2_ch'].between(1.005, 1.205), df['hi_top'].between(1, 1.025), df['bb_w_upp_min'].ge(0.02), df['bb_w_upp_max'].le(0.22), # df['bb_upp_min'].ge(-0.1), df['bb_upp_max'].le(0.11), # df[f"hc2_pct_{i}"].between(-c, c), # df[f"bb_w_upp_{i}"].between(0.02, 0.32), # df[f"vol_chg_{i}"].between(1.5, 2.5), # df[f"bb_upp_{i}"].shift(1).le(df[f"bb_upp_{i}"]), df[f"bb_upp_{i}"].shift(1).lt(0), df[f"bb_upp_{i}"].between(0.05, 0.35) ] df.loc[reduce(lambda x, y: x & y, buy_bb_upp), 'custom_buy_tag'] += f"{i} " buy_bb_upp = df['custom_buy_tag'].ne('') df.loc[buy_bb_upp, 'custom_buy_tag'] = 'upp ' + df['custom_buy_tag'].str.strip() if __name__ not in ('low', 'upp'): buy_vol = [ df['candle_count_1d'].ge(self.min_day_listed), df['str_min'].ge(1), df['str_max_hl2_ch'].ge(1.5), df['volume'].div(df['volume'].shift(1)).ge(150) ] df.loc[reduce(lambda x, y: x & y, buy_vol), 'custom_buy_tag'] = 'vol' df.loc[df['custom_buy_tag'].ne(''), 'buy'] = True # buy_bb_low = df['buy_tag_bb_low'].ne('') & df['buy_tag_bb_upp'].eq('') # buy_bb_upp = df['buy_tag_bb_upp'].ne('') & df['buy_tag_bb_low'].eq('') # df.loc[buy_bb_low, 'custom_buy_tag'] = 'low ' + df['buy_tag_bb_low'].str.strip() # df.loc[buy_bb_upp, 'custom_buy_tag'] = 'upp ' + df['buy_tag_bb_upp'].str.strip() # df.loc[df['custom_buy_tag'].ne(''), 'buy'] = True df['buy_tag'] = df['custom_buy_tag'] if len(self.dp.current_whitelist()) <= 3: if __name__ not in ('upp', 'vol') and not df.loc[buy_bb_low, :].empty: lb = self.lb_buy_low #print(df.loc[buy_bb_low, ['pair', 'date', 'str_min_hl2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'bb_w_low_min', 'bb_w_low_max', 'bb_low_min', 'bb_low_max', 'close', 'hl2', 'volume', 'buy_tag']].to_string()) #print(df.loc[buy_bb_low, [f"vol_chg_{i}" for i in lb]].to_string()) # #print(df.loc[buy_bb_low, [f"lc2_pct_{i}" for i in self.lb_buy_lo]].to_string()) # #print(df.loc[buy_bb_low, [f"bb_w_low_{i}" for i in lb]].to_string()) #print(df.loc[buy_bb_low, [f"bb_low_{i}" for i in lb]].to_string()) if __name__ not in ('low', 'vol') and not df.loc[buy_bb_upp, :].empty: lb = self.lb_buy_upp #print(df.loc[buy_bb_upp, ['pair', 'date', 'str_min_hl2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'bb_w_upp_min', 'bb_w_upp_max', 'bb_upp_min', 'bb_upp_max', 'close', 'hl2', 'volume', 'buy_tag']].to_string()) #print(df.loc[buy_bb_upp, [f"vol_chg_{i}" for i in lb]].to_string()) # #print(df.loc[buy_bb_upp, [f"hc2_pct_{i}" for i in self.lb_buy_upp]].to_string()) # #print(df.loc[buy_bb_upp, [f"bb_w_upp_{i}" for i in self.lb_buy_upp]].to_string()) #print(df.loc[buy_bb_upp, [f"bb_upp_{i}" for i in lb]].to_string()) if __name__ not in ('low', 'upp'): #print(df.loc[df['buy_tag'] == 'vol', ['pair', 'date', 'str_min_hc2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'close', 'hl2', 'volume', 'buy_tag']].to_string()) return df def populate_sell_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[:, ['sell', 'exit_tag']] = (False, None) return df def custom_sell(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float, current_profit: float, **kwargs): df: DataFrame = self.dp.get_analyzed_dataframe(pair, self.timeframe)[0] trade_open_date = timeframe_to_prev_date(self.timeframe, trade.open_date_utc) df_trade = df.loc[df['date'].ge(trade_open_date)] trade_len = len(df_trade) candle_1 = df_trade.iloc[-1] sell_date = candle_1['date'] if trade_len <= 2: return None d = sell_date.strftime('%Y-%m-%d %H:%M') u: str = trade.buy_tag[:3] cp = (candle_1['close'] - trade.open_rate) / trade.open_rate df_m = self.df_market.loc[self.df_market['date'].eq(sell_date)] if df_m.empty: log.error(f"{sell_date} custom_sell: No volume info for sell candle.") return 'sell error' candle_2 = df_trade.iloc[-2] str_max = candle_1['str_1'] t = 1.04 - str_max * 0.0015 if candle_1['hi_top'] >= t and ((candle_1['green'] and candle_1['volume'] / candle_2['volume'] <= min(1, 0.5 + str_max * 0.05)) or (not candle_1['green'] and candle_1['volume'] / candle_2['volume'] >= 1.1)): log.info(f"{d} custom_sell: hito {u} {cp:.2f} for pair {pair} with trade len {trade_len}.") return f"hito {u}" candle_min = df_trade.loc[df_trade['close'] <= 1.001 * df_trade['close'].min()].iloc[-1] candle_max = df_trade.loc[df_trade['close'] >= 0.999 * df_trade['close'].max()].iloc[-1] rise = candle_max['close'] / candle_min['close'] if candle_max['date'] > candle_min['date'] else 1 fall = candle_max['close'] / candle_1['close'] if trade_len <= 18: max_fall = 1.04 if u != 'low' else 1.08 else: max_fall = max(1.08, pow(min(1.4, rise), 0.25)) if fall > max_fall and candle_1['lo_bot'] > 0.997: log.info(f"{d} custom_sell: fall {u} {cp:.2f} for pair {pair} with trade len {trade_len}.") return f"fall {u}" # if fall > 1.02: # i = self.lb_market # df_top_vol = df_m.nlargest(self.top_volume, f"hlc3_vol_{i}") # if len(df_m) >= 5 and self.market_down(df_top_vol, i): # log.info(f"{d} custom_sell: market down sell {cp:.2f} for pair {pair} with trade len {trade_len}.") # return f"market down {u}" i = 36 if trade_len > i: # ma = df_trade['hlc3'].tail(36).pct_change(1).max() # mi = df_trade['hlc3'].tail(36).pct_change(1).min() sk = df_trade['hl2'].tail(i).pct_change(1).skew() ku = df_trade['hl2'].tail(i).pct_change(1).kurt() # #print(f"{d} rise: {rise} fall:{fall} max fall:{max_fall} pct_max:{ma} pct_min:{mi} skew:{sk} kurt:{ku} cp: {cp:.2f} mm: {ma - mi}") if sk <= -0.33 and ku <= -0.33: log.info(f"{d} custom_sell: side {u} {cp:.2f} for pair {pair} with skew {sk:.2f}, kurtosis {ku:.2f} and trade len {trade_len}.") # if candle_1['str_min'] == -1 and candle_1['str_max'] == 1 and candle_2['str_min'] == -1 and candle_2['str_max'] == 1: # log.info(f"{d} custom_sell: side {u} {cp:.2f} for pair {pair} trade len {trade_len}.") return f"side {u}" return None def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, **kwargs) -> bool: # #print(self.df_low.to_string()) # #print(self.df_upp.to_string()) # exit() df = self.dp.get_analyzed_dataframe(pair, self.timeframe)[0] buy_date = df['date'].iloc[-1] buy_tag = df['custom_buy_tag'].iloc[-1] u: str = buy_tag[:3] d = buy_date.strftime('%Y-%m-%d %H:%M') df_m = self.df_market.loc[self.df_market['date'].eq(buy_date)] if df_m.empty: log.error(f"{buy_date} confirm_trade_entry: No volume info for buy candle.") return False elif len(df_m) >= 3: j = self.lb_market if buy_date not in self.custom_market_info: self.custom_market_info = {} self.custom_market_info[buy_date] = None self.custom_market_info['low'] = None self.custom_market_info['upp'] = None self.custom_market_info['vol'] = None self.df_top_vol = df_m.nlargest(self.top_volume, f"hl2_vol_{j[2]}") if u == 'low' else df_m.nlargest(self.top_volume, 'hl2_vol_1') if self.custom_market_info[f"{u}"] == None: self.custom_market_info[f"{u}"] = True md1 = self.market_down(self.df_top_vol, j[0]) or self.market_down(self.df_top_vol, j[1]) md2 = self.market_down(self.df_top_vol, j[2]) if (u == 'low' and md1 or md2) or (u == 'upp' and md1): log.info(f"{d} confirm_trade_entry: Cancel all buys. Market down.") self.custom_market_info[f"{u}"] = False # df_top_vol = df_top_vol.loc[df_m['pair'] != pair] # count_pairs = len(df_top_vol) # if u == 'upp': # n1_upp = df_top_vol.loc[df_top_vol['pct_8'] >= 0.001].count()[0] # if n1_upp <= count_pairs * 0.1: # log.info(f"{d} confirm_trade_entry: Cancel buy {u}. Not enough pairs up. There are {n1_up} of the {count_pairs} pairs over 0.1% upp in the last candle.") # self.custom_market_info[buy_date][f"{u}"] = False # return False # if u == 'low': # n1_low = df_top_vol.loc[df_top_vol['pct_8'] <= -0.02].count()[0] # n1_1_low = df_top_vol.loc[df_top_vol['pct_8'].shift(1) <= -0.02].count()[0] # if (n1_low >= count_pairs * 0.4) or (n1_1_low >= count_pairs * 0.4 and n1_low >= count_pairs * 0.2): # log.info(f"{d} confirm_trade_entry: Cancel buy {u}. Too much pairs down. There are {n1_lo} and {n1_1_lo} of the {count_pairs} pairs over 2% down in the last and previous candle.") # self.custom_market_info[buy_date][f"{u}"] = False # return False if self.custom_market_info[f"{u}"]: df_pair = self.df_top_vol.loc[self.df_top_vol['pair'] == pair] if df_pair.empty: # log.info(f"{d} confirm_trade_entry: Cancel buy {u} for pair {pair}. Volume not in top {self.top_volume}.") return False else: log.info(f"{d} confirm_trade_entry: Buy for pair {pair} with buy tag {buy_tag}.") return True else: return False log.info(f"{d} confirm_trade_entry: Buy for pair {pair} with buy tag {buy_tag}.") return True def market_down(self, df: DataFrame, i: int) -> bool: cnt_pairs = len(df) c = df[f"hl2_pct_{i}"] df_neg = df.loc[c <= -0.01] df_pos = df.loc[c >= 0.01] cnt_neg_pairs = len(df_neg) cnt_pos_pairs = len(df_pos) # sum_neg = df_neg[f"hlc3_vol_{i}"].mul(df_neg[f"hlc3_pct_{i}"]).sum() # sum_pos = df_pos[f"hlc3_vol_{i}"].mul(df_pos[f"hlc3_pct_{i}"]).sum() d = df['date'].iloc[-1].strftime('%Y-%m-%d %H:%M') # #print(f"{d} sum_pos: {sum_pos} sum_neg: {sum_neg} cnt_pairs: {cnt_pairs} cnt_pos_pairs: {cnt_pos_pairs} cnt_neg_pairs: {cnt_neg_pairs}") # log.info(f"{d} cnt_pairs: {cnt_pairs} cnt_pos_pairs: {cnt_pos_pairs} cnt_neg_pairs: {cnt_neg_pairs}") # return sum_pos < abs(sum_neg) and cnt_neg_pairs >= cnt_pairs // 2 and cnt_pos_pairs <= cnt_neg_pairs // 8 return cnt_pos_pairs <= cnt_neg_pairs #cnt_pos_pairs == 0 or cnt_neg_pairs > cnt_pos_pairs // 2 # (cnt_neg_pairs >= cnt_pairs // 2 and cnt_pos_pairs <= cnt_pairs // 40) or class low(github_BillGatesIII_Vires_in_Numeris__vin__20220218_143232): lb_buy_low = range(9, 49) lb_buy_upp = range(1, 1) # def market_down(self, df: DataFrame) -> bool: # return False class upp(github_BillGatesIII_Vires_in_Numeris__vin__20220218_143232): lb_buy_low = range(1, 1) lb_buy_upp = range(9, 49)