# source: https://raw.githubusercontent.com/BillGatesIII/Vires-in-Numeris/4df7b5dc07b1ad4b14a2cb495aca458ec7e5ddcf/vin.py 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__20220319_070112(IStrategy): INTERFACE_VERSION = 2 def version(self) -> str: return 'v7.0.0' min_day_listed = 5 top_volume = 180 df_market = DataFrame() df_top_vol = DataFrame() df_vin = DataFrame() custom_market_info = {} lb_market = (12, 36, 72) minimal_roi = {"0": 100} stoploss = -0.08 stoploss_on_exchange = False trailing_stop = True trailing_stop_positive = 0.02 trailing_only_offset_is_reached = True trailing_stop_positive_offset = 0.04 use_custom_stoploss = False timeframe = '5m' process_only_new_candles = True use_sell_signal = False sell_profit_only = False startup_candle_count = 72 @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'].sub(df['open']).ge(0) df['close_ch'] = df['close'].pct_change() df['volume_ch'] = df['volume'].pct_change() df['hlc3'] = df['high'].add(df['low']).add(df['close']).div(3) df['hlc3_vol'] = df['hlc3'].mul(df['volume']) df['vin_buy'] = ( # df['close'].shift(-1).pct_change(1).ge(0.02) & df['close'].shift(-3).pct_change(3).ge(0.08) & df['candle_count_1d'].ge(self.min_day_listed) & df['close_ch'].between(0.02, 0.04) & df['volume_ch'].ge(8) ) df['vin_sell'] = ( # df['close'].shift(-2).pct_change(1).lt(-0.01) & # df['close'].shift(-1).pct_change(1).le(-0.01) & df['close'].shift(-3).pct_change(3).le(-0.08) df['hl2_ch'].lt(0.01) & df['oc2_ch'].lt(0.001) & df['hl2_ch'].div(df['oc2_ch']).lt(1.01) # & df['volume_ch'].lt(2) ) df_pair = df[['date', 'pair', 'vin_buy', 'hlc3_vol']] for i in self.lb_market: df_pair = concat([df_pair, DataFrame({f"vol_ch_{i}": df['volume'].pct_change(i)}), DataFrame({f"cl_ch_{i}": df['close'].pct_change(i)})], axis=1) self.df_market = concat([self.df_market, df_pair], ignore_index=True) # self.df_vin = concat([self.df_vin, df.loc[df['vin_buy'], ['pair', 'date', 'vin_buy', 'vin_sell', 'open', 'high', 'low', 'close', 'oc2', 'oc2_ch', 'hl2', 'hl2_ch', 'volume', 'volume_ch']]]) if len(self.dp.current_whitelist()) <= 6: f = df['vin_buy'] | df['vin_sell'] #print(df.loc[f, ['pair', 'date', 'vin_buy', 'vin_sell', 'oc2', 'oc2_ch', 'hl2', 'hl2_ch', 'volume', 'volume_ch']].to_string()) return df def populate_buy_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[df['vin_buy'], ['buy', 'buy_tag']] = (True, 'vin_buy') return df def populate_sell_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[df['vin_sell'], ['sell', 'exit_tag']] = (True, 'vin_sell') return df 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_vin.to_string()) # exit() df: DataFrame = self.dp.get_analyzed_dataframe(pair, self.timeframe)[0] buy_date = df['date'].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 if len(df_m) <= 5: return True 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, 'hlc3_vol') if self.custom_market_info[buy_date] == None: self.custom_market_info[buy_date] = True # if self.market_down(self.df_top_vol, 'buy'): # log.info(f"{d} confirm_trade_entry: Cancel all buys. Market down.") # self.custom_market_info[buy_date] = False # else: # df_buy = self.df_top_vol.loc[self.df_top_vol['vin_buy']] # if len(df_buy) >= self.top_volume // 18: # log.info(f"{d} confirm_trade_entry: Cancel all buys. Too many buys in this candle.") # 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 for pair {pair}. Volume not in top {self.top_volume}.") return False else: log.info(f"{d} confirm_trade_entry: Buy for pair {pair}.") # self.df_vin = concat([self.df_vin, df[['pair', 'date', 'green', 'close', 'hl2_ch', 'volume', 'vol_ch', 'hl2_vol', 'hl2_vol_ch']].tail(1)]) # #print(self.df_vin.to_string()) return True else: df.loc[df['date'].eq(buy_date), ['buy', 'buy_tag']] = (False, None) return False def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, sell_reason: str, current_time: datetime, **kwargs) -> bool: df: DataFrame = self.dp.get_analyzed_dataframe(pair, self.timeframe)[0] sell_date = df['date'].iloc[-1] d = sell_date.strftime('%Y-%m-%d %H:%M') log.info(f"{d} confirm_trade_exit: Sell for pair {pair}.") return True def market_down(self, df: DataFrame, buy_sell: str) -> bool: # d = df['date'].iloc[0].strftime('%Y-%m-%d %H:%M') cnt_pairs_neg_price_change = {} cnt_pairs_pos_price_change = {} for i in self.lb_market: c = df[f"cl_ch_{i}"] df_neg = df.loc[c <= -0.01] df_pos = df.loc[c >= 0.01] cnt_pairs_neg_price_change[i] = len(df_neg) cnt_pairs_pos_price_change[i] = len(df_pos) down = False if buy_sell == 'buy': down = cnt_pairs_neg_price_change[i] >= cnt_pairs_pos_price_change[i] else: down = cnt_pairs_pos_price_change[i] == 0 if down: break return down