# source: https://raw.githubusercontent.com/BillGatesIII/Vires-in-Numeris/d62f669313b7de8a8a510d87abf510e343ada0c7/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__20220225_115709(IStrategy): INTERFACE_VERSION = 2 def version(self) -> str: return 'v5.0.3' 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, 4, 24) 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().fillna(value=0) return df def populate_indicators(self, df: DataFrame, metadata: dict) -> DataFrame: df['pair'] = metadata['pair'] df['green'] = df['close'].div(df['open']).ge(1) top = df['close'].where(df['green'], df['open']) bot = df['open'].where(df['green'], df['close']) 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) * df['high']).div(2) lo_tail_ratio = df['low'].sub(bot).div(mid) lo_adj = bot.add(pow(1 - lo_tail_ratio, 0.25) * df['low']).div(2) df['hl2'] = hi_adj.add(lo_adj).div(2) df['hi_top'] = hi_adj.div(top) df['lo_bot'] = lo_adj.div(bot) df_hl2_ch: DataFrame = df['hl2'].sub(df['hl2'].shift(1)) s = (1, 2, 3) for i in s: 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_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()]) ef = [df] for i in {*self.lb_buy_low, *self.lb_buy_upp}: ef.append(DataFrame({f"vol_chg_{i}": df['volume'] / df['volume'].shift(1).rolling(window=i, min_periods=i).mean()})) if __name__ not in ('upp', 'vol'): for i in self.lb_buy_low: lo_pct = lo_adj.pct_change(i) pct_mean = lo_pct.rolling(window=i, min_periods=i).mean() pct_std = 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)})) ef.append(DataFrame({f"bb_low_{i}": lo_pct.sub(bb_pct_low)})) # df = concat(ef, 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: hi_pct = hi_adj.pct_change(i) pct_mean = hi_pct.rolling(window=i, min_periods=i).mean() pct_std = 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)})) ef.append(DataFrame({f"bb_upp_{i}": hi_pct.sub(bb_pct_upp)})) # df = concat(ef, 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) df = concat(ef, axis=1) hl2_vol = df['hl2'].mul(df['volume']) j = self.lb_market df_pair = df[['date', 'pair', 'hl2', 'str_min', 'str_min_hl2_ch', 'str_max', 'str_max_hl2_ch', 'volume']] for i in j: df_pair = concat([df_pair, DataFrame({f"hl2_vol_{i}": hl2_vol.rolling(window=i, min_periods=i).sum()}), DataFrame({f"hl2_pct_{i}": df['hl2'].pct_change(i)})], axis=1) # df_pair = concat([df_pair, DataFrame({f"hl2_pct_{i}": df['hl2'].pct_change(i)})], axis=1) self.df_market = concat([self.df_market, df_pair], ignore_index=True) df.drop(columns=['updown'], inplace=True) if len(self.dp.current_whitelist()) == 1: if __name__ not in ('upp', 'vol'): f = df['str_min'].le(1) & df['str_max'].le(2) & df['str_min_hl2_ch'].between(0.795, 0.995) & df['lo_bot'].le(0.9995) i = 11 #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', 'close', 'hl2', 'volume', f"bb_w_low_{i}", f"vol_chg_{i}", f"bb_low_{i}"]].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()) if __name__ not in ('low', 'vol'): f = df['str_max'].between(1, 4) #& df['str_min'].ge(1) & df['bb_upp_max'].gt(0) i = 11 #print(df.loc[:, ['pair', 'date', 'green', 'str_min_hl2_ch', 'str_max_hl2_ch', 'str_1', 'str_2', 'str_3', 'hi_top', 'lo_bot', 'close', 'hl2', 'volume', f"bb_w_upp_{i}", f"vol_chg_{i}", f"bb_upp_{i}"]].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()) return df def populate_buy_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[:, 'custom_buy_tag'] = '' if __name__ not in ('upp', 'vol'): for i in self.lb_buy_low: buy_bb_low = [ df['candle_count_1d'].ge(self.min_day_listed), df['str_min'].le(-1), df['str_max'].le(-1), df['str_min_hl2_ch'].between(0.795, 0.995), df['lo_bot'].le(0.9995), df[f"bb_w_low_{i}"].between(0.02, 0.07), df[f"vol_chg_{i}"].between(0.5, 1.5), df[f"bb_low_{i}"].shift(1).gt(0), df[f"bb_low_{i}"].between(-0.011, -0.001) ] 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: buy_bb_upp = [ df['candle_count_1d'].ge(self.min_day_listed), df['str_min'].ge(-1), df['str_max'].ge(2), df['str_max_hl2_ch'].between(1.005, 1.205), df['hi_top'].le(1.0005), df[f"bb_w_upp_{i}"].between(0.02, 0.17), df[f"vol_chg_{i}"].between(1.5, 2.5), df[f"bb_upp_{i}"].shift(1).lt(0), df[f"bb_upp_{i}"].between(0.005, 0.025) ] 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 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 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' 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 candle_2 = df_trade.iloc[-2] str_max = candle_1['str_max'] if candle_1['close'] / candle_2['close'] <= 0.98 and len(df_m) >= 3: md = True for i in self.lb_market: md = md and self.market_down(self.df_top_vol, i) if not md: break if md: log.info(f"{d} custom_sell: market down sell {cp:.2f} for pair {pair} with trade len {trade_len}.") return f"market down {u}" if trade_len <= 18 and cp >= -0.18: return None t = 1.04 - str_max * 0.002 if candle_1['hi_top'] >= t and candle_1['green'] and 0.8 <= candle_1['volume'] / candle_2['volume'] <= 1: log.info(f"{d} custom_sell: high top green {u} {cp:.2f} for pair {pair} with trade len {trade_len}.") return f"high top green {u}" t = 1.06 - str_max * 0.004 if candle_1['hi_top'] >= t and not candle_1['green'] and candle_1['volume'] / candle_2['volume'] <= 0.8 and str_max <= candle_2['str_max']: log.info(f"{d} custom_sell: high top red {u} {cp:.2f} for pair {pair} with trade len {trade_len}.") return f"high top red {u}" candle_min = df_trade.loc[df_trade['hl2'] <= 1.001 * df_trade['hl2'].min()].iloc[-1] candle_max = df_trade.loc[df_trade['hl2'] >= 0.999 * df_trade['hl2'].max()].iloc[-1] rise = candle_max['hl2'] / candle_min['hl2'] if candle_max['date'] > candle_min['date'] else 1 fall = candle_max['hl2'] / candle_1['hl2'] max_fall = max(1.04, pow(min(1.2, rise), 0.25)) if fall > max_fall and str_max <= -5: log.info(f"{d} custom_sell: fall {u} {cp:.2f} for pair {pair} with trade len {trade_len}.") return f"fall {u}" i = 36 if trade_len > i: 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}.") 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: 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_{max(self.lb_market)}") if self.custom_market_info[f"{u}"] == None: self.custom_market_info[f"{u}"] = True if self.market_down(self.df_top_vol, 1) or self.market_down(self.df_top_vol, 2): # or (u == 'low' and self.market_down(self.df_top_vol, max(self.lb_market))): log.info(f"{d} confirm_trade_entry: Cancel all buys. Market looks bearish.") self.custom_market_info[f"{u}"] = 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) # d = df['date'].iloc[0].strftime('%Y-%m-%d %H:%M') # log.info(f"{d} market_down: count negative pairs: {cnt_neg_pairs}, count positive pairs: {cnt_pos_pairs}") return cnt_pos_pairs == 0 #and cnt_neg_pairs >= cnt_pairs // 10 or cnt_neg_pairs >= max(1, cnt_pos_pairs) * 2 def temp(self, df: DataFrame): if len(self.dp.current_whitelist()) <= 3: b = 6 f = df['close'].pct_change(b).ge(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['str_min'].le(1), df['str_max'].le(2), df['str_min_hl2_ch'].between(0.795, 0.995) ] f = reduce(lambda x, y: x & y, f) if not df.loc[f].empty: 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', '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['str_max'].ge(-1), df['str_min'].ge(-2), df['str_max_hl2_ch'].between(1.005, 1.205) ] f = reduce(lambda x, y: x & y, f) if not df.loc[f].empty: 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', 'hl2', 'open', 'high', 'low', 'close', 'volume']]]) class low(github_BillGatesIII_Vires_in_Numeris__vin__20220225_115709): lb_buy_low = range(9, 49) lb_buy_upp = range(1, 1) class upp(github_BillGatesIII_Vires_in_Numeris__vin__20220225_115709): lb_buy_low = range(1, 1) lb_buy_upp = range(9, 49)