# source: https://raw.githubusercontent.com/BillGatesIII/Vires-in-Numeris/bdc0fa5bf6805b3a606e9c5b618aa44d042a529c/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 talib.abstract as ta 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='') import logging log = logging.getLogger(__name__) class github_BillGatesIII_Vires_in_Numeris__vin__20220227_104014(IStrategy): INTERFACE_VERSION = 2 def version(self) -> str: return 'v6.0.0' min_day_listed = 5 top_volume = 80 df_market = DataFrame() df_top_vol = DataFrame() df_low = DataFrame() df_upp = DataFrame() custom_market_info = {} lb_market = (1, 18) minimal_roi = {"0": 100} stoploss = -1 stoploss_on_exchange = False trailing_stop = False use_custom_stoploss = False timeframe = '1h' process_only_new_candles = True use_sell_signal = True sell_profit_only = False startup_candle_count = 18 @property def protections(self): return [ { "method": "CooldownPeriod", "stop_duration_candles": 18 } ] @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']) df['wick_top'] = df['high'].div(top) df['wick_bot'] = bot.div(df['low']) 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()]) hl2_vol = df['hl2'].mul(df['volume']) df_pair = df[['date', 'pair', 'hl2', 'str_min', 'str_max', 'str_min_hl2_ch', 'str_max_hl2_ch', 'volume']] for i in self.lb_market: 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) 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: f = df['str_min'].le(-1) & df['str_max'].le(1) #print(df.loc[f, ['pair', 'date', 'green', 'str_1', 'str_2', 'str_3', ' str_min', 'str_max', 'str_min_hl2_ch', 'str_max_hl2_ch', 'wick_top', 'wick_bot', 'close', 'hl2', 'volume']].shift(1).to_string()) #print(df.loc[f, ['pair', 'date', 'green', 'str_1', 'str_2', 'str_3', ' str_min', 'str_max', 'str_min_hl2_ch', 'str_max_hl2_ch', 'wick_top', 'wick_bot', 'close', 'hl2', 'volume']].to_string()) return df def populate_buy_trend(self, df: DataFrame, metadata: dict) -> DataFrame: buy_bot = (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(-1, 1) # & df['str_max_hl2_ch'].between(-1, 1) & df['hl2'].div(df['hl2'].shift(1)).between(0.91, 1.01) & df['volume'].div(df['volume'].shift(1)).between(0.75, 1.05) & df['wick_bot'].gt(df['wick_top']) & df['wick_bot'].ge(1.005) ) df.loc[buy_bot, ['buy', 'buy_tag']] = (True, 'bot') if len(self.dp.current_whitelist()) <= 3: f = buy_bot #print(df.loc[f, ['pair', 'date', 'green', 'str_1', 'str_2', 'str_3', ' str_min', 'str_max', 'str_min_hl2_ch', 'str_max_hl2_ch', 'wick_top', 'wick_bot', 'close', 'hl2', 'volume']].to_string()) return df def populate_sell_trend(self, df: DataFrame, metadata: dict) -> DataFrame: if self.market_down(self.df_top_vol, min(self.lb_market)) or self.market_down(self.df_top_vol, max(self.lb_market)): log.info(f"populate_sell_trend: market down sell for pair {metadata['pair']}.") df.loc[sell_top, ['sell', 'exit_tag']] = (True, 'market down') else: sell_top = (df['str_min'].ge(-1) & df['str_max'].ge(1) # & df['str_min_hl2_ch'].between(-1, 1) # & df['str_max_hl2_ch'].between(-1, 1) & df['hl2'].div(df['hl2'].shift(1)).between(1.01, 1.11) & df['volume'].div(df['volume'].shift(1)).between(0.75, 1.05) & df['wick_top'].gt(df['wick_bot']) & df['wick_top'].ge(1.005) ) df.loc[sell_top, ['sell', 'exit_tag']] = (True, 'top') # 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) if trade_len <= 2: return None sell_date, str_max, close, hl2 = df_trade[['date', 'str_max', 'close', 'hl2']].iloc[-1] 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') cp = (close - trade.open_rate) / trade.open_rate 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'] / hl2 max_fall = max(1.04, pow(min(1.2, rise), 0.25)) if fall > max_fall and str_max <= -1: log.info(f"{d} custom_sell: fall {cp:.2f} for pair {pair} with trade len {trade_len}.") return f"fall" i = 12 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 {cp:.2f} for pair {pair} with skew {sk:.2f}, kurtosis {ku:.2f} and trade len {trade_len}.") return f"side" 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, buy_tag = df[['date', 'buy_tag']].iloc[-1] 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 = dict.fromkeys([buy_date], None) self.df_top_vol = df_m.nlargest(self.top_volume, 'hl2_vol_1') if self.custom_market_info[buy_date] == None: self.custom_market_info[buy_date] = True if self.market_down(self.df_top_vol, min(self.lb_market)) or 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[buy_date] = False if self.custom_market_info[buy_date]: 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}.") return True else: return False log.info(f"{d} confirm_trade_entry: Buy for pair {pair}.") 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']]])