# source: https://raw.githubusercontent.com/DerSalvador/freqtrade-helm-chart/a669dc11b640b0eb63aa8f8b51e9f181fd7ee43c/chart/deployed_strategies/bot-mssm-08-k8s-namespace/extra_strategies/TaSearchLevelG15m_Short.py kubectl --context=gke_vaulted-gift-406223_europe-west1-b_private-cluster-3 -n bot-mssm-08 exec -it pod/freqtrade-bot-mssm-08-765d99b7b4-dl7gb -c freqtrade -- cat /extra_strategies/Github_DerSalvador_freqtrade_helm_chart__TaSearchLevelG15m_Short__20260115_122204.py import pandas as pd import numpy as np import talib.abstract as ta import logging from datetime import datetime from typing import Dict, List, Optional, Tuple, Union from scipy import signal from statistics import mean from freqtrade.persistence.trade_model import Trade from freqtrade.strategy.interface import IStrategy # "minimal_roi": { # "0": 0.05 # }, # "stoploss": -0.05, # # "trailing_stop": true, # "trailing_stop_positive": 0.05, # "trailing_stop_positive_offset": 0.2, # "trailing_only_offset_is_reached": true, class Github_DerSalvador_freqtrade_helm_chart__TaSearchLevelG15m_Short__20260115_122204(IStrategy): can_short: bool = True def populate_indicators(self, df: pd.DataFrame, metadata: dict) -> pd.DataFrame: pd.set_option('display.max_rows', 100000) pd.set_option('display.precision', 10) pd.set_option('mode.chained_assignment', None) df['rsi_7'] = ta.RSI(df['close'], timeperiod=7).round(2) df['min_level'] = 0 df['max_level'] = 0 df['buy_short'] = 0 df['buy_long'] = 0 df['buy_short2'] = 0 df['buy_long2'] = 0 df['i_close'] = 0 df['i_open'] = 0 df['i_low'] = 0 df['i_high'] = 0 logging.getLogger('freqtrade').info(str(metadata)) df = self.do_heikin_ashi(df) # Disable long signals # df = self.do_long(df) df = self.do_short(df) return df def do_long(self, df: pd.DataFrame) -> pd.DataFrame: n = 200 df['min_local'] = df.iloc[signal.argrelextrema(df.c.values, np.less_equal, order=n)[0]]['c'] min = df['c'].max() times = df.query(f'min_local > 0') prices = [] if len(times) > 0: for i in range(500, len(df)): if df['min_local'].loc[i] > 0: close = df['c'].loc[i] chunk = df[:i - 10] chunk['min_local'] = chunk.iloc[signal.argrelextrema(chunk.c.values, np.less_equal, order=n)[0]]['c'] time_chunk = chunk.query(f'min_local > 0') for x, row in time_chunk.iterrows(): close_chunk = time_chunk['c'].loc[x] prices.append(close_chunk) for p in prices: diff = self.diff_percentage(p, close) if 0 < diff < 0.3: logging.getLogger('freqtrade').info(str([i, '+++', diff, p])) df['min_level'].loc[i] = 1 for x in range(i - 2, i): # if candle green # if df['min_level'].loc[x] > 0 and df['o'].loc[i] < df['c'].loc[i]: if df['min_level'].loc[x] > 0: close = df['c'].loc[i] # if same value presented in past # and if value < average for long df_tail = df[i - 200: i].query(f'c < {close}') if len(df_tail) > 0: logging.getLogger('freqtrade').info( str([i, '++++', min, diff, '---', prices, 'long ++++++']) ) df['buy_long'].loc[i] = 1 df['i_low'].loc[i] = df['low'].loc[x] df['i_high'].loc[i] = df['high'].loc[x] df['i_open'].loc[i] = df['o'].loc[x] df['i_close'].loc[i] = df['c'].loc[x] for x in range(i - 100, i): if df['min_level'].loc[x] > 0: diff = self.diff_percentage(df['c'].loc[x], df['c'].loc[i]) # print('long ------- ', diff) # long # past < now # if df['c'].loc[x] < df['c'].loc[i] and 0 < diff < 1: if 0 < diff < 1: logging.getLogger('freqtrade').info( str([i, '---- long2 ----']) ) df['buy_long2'].loc[i] = 1 df['i_low'].loc[i] = df['low'].loc[x] df['i_high'].loc[i] = df['high'].loc[x] df['i_open'].loc[i] = df['o'].loc[x] df['i_close'].loc[i] = df['c'].loc[x] return df def do_short(self, df: pd.DataFrame) -> pd.DataFrame: n = 200 df['max_local'] = df.iloc[signal.argrelextrema(df.c.values, np.greater_equal, order=n)[0]]['c'] max = df['c'].min() times = df.query(f'max_local > 0') prices = [] if len(times) > 0: for i in range(500, len(df)): if df['max_local'].loc[i] > 0: close = df['c'].loc[i] chunk = df[:i - 10] chunk['max_x'] = chunk.iloc[signal.argrelextrema(chunk.c.values, np.greater_equal, order=n)[0]]['c'] time_chunk = chunk.query(f'max_x > 0') for x, row in time_chunk.iterrows(): close_chunk = time_chunk['c'].loc[x] prices.append(close_chunk) for p in prices: diff = self.diff_percentage(p, close) if 0 < diff < 0.3: logging.getLogger('freqtrade').info(str([i, '---', diff, p])) df['max_level'].loc[i] = 1 for x in range(i - 2, i): # if candle red # if df['max_level'].loc[x] > 0 and df['o'].loc[i] > df['c'].loc[i]: if df['max_level'].loc[x] > 0: close = df['c'].loc[i] # if same value presented in past # and if value > average for short df_tail = df[i - 200: i].query(f'c > {close}') if len(df_tail) > 0: logging.getLogger('freqtrade').info( str([i, '---', max, diff, '---', prices, 'short -----']) ) df['buy_short'].loc[i] = 1 df['i_low'].loc[i] = df['low'].loc[x] df['i_high'].loc[i] = df['high'].loc[x] df['i_open'].loc[i] = df['o'].loc[x] df['i_close'].loc[i] = df['c'].loc[x] for x in range(i - 100, i): if df['max_level'].loc[x] > 0: diff = self.diff_percentage(df['c'].loc[x], df['c'].loc[i]) # past > now # if df['c'].loc[x] > df['c'].loc[i] and 0 < diff < 1: if 0 < diff < 1: logging.getLogger('freqtrade').info( str([i, '---- short2 ----']) ) df['buy_short2'].loc[i] = 1 df['i_low'].loc[i] = df['low'].loc[x] df['i_high'].loc[i] = df['high'].loc[x] df['i_open'].loc[i] = df['o'].loc[x] df['i_close'].loc[i] = df['c'].loc[x] return df def populate_entry_trend(self, df: pd.DataFrame, metadata: dict) -> pd.DataFrame: df.loc[(df['buy_short'] > 0), 'enter_short'] = 1 df.loc[(df['buy_short2'] > 0), 'enter_short'] = 1 df.loc[(df['buy_long'] > 0), 'enter_long'] = 0 df.loc[(df['buy_long2'] > 0), 'enter_long'] = 0 return df def populate_exit_trend(self, df: pd.DataFrame, metadata: dict) -> pd.DataFrame: df.loc[ (df['rsi_7'] < 10), 'exit_short' ] = 1 df.loc[ (df['rsi_7'] > 90), 'exit_long' ] = 1 return df def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: return 3 def diff_percentage(self, v2, v1) -> float: diff = ((v2 - v1) / ((v2 + v1) / 2)) * 100 diff = np.round(diff, 4) return np.abs(diff) def do_heikin_ashi(self, df: pd.DataFrame) -> pd.DataFrame: """ https://www.investopedia.com/trading/heikin-ashi-better-candlestick/ """ df['h'] = df.apply(lambda x: max(x['high'], x['open'], x['close']), axis=1) df['l'] = df.apply(lambda x: min(x['low'], x['open'], x['close']), axis=1) for i in range(1, len(df)): df.loc[i, 'c'] = 1 / 4 * ( df['open'].iloc[i] + df['close'].iloc[i] + df['high'].iloc[i] + df['low'].iloc[i] ) df.loc[i, 'o'] = 1 / 2 * ( df['open'].iloc[i - 1] + df['close'].iloc[i - 1] ) return df def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: """ https://www.freqtrade.io/en/stable/strategy-callbacks/#trade-entry-buy-order-confirmation """ df, last_updated = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) print('------------------- confirm_trade_entry -------------', pair, rate) # mean = ( # df['i_open'].iat[-1] + # df['i_close'].iat[-1] # ) / 2 # # open = df['i_open'].iat[-1] # # close = df['i_close'].iat[-1] # min_ = min( # # df['i_high'].iat[-1], # # df['i_low'].iat[-1], # df['i_open'].iat[-1], # df['i_close'].iat[-1] # ) # max_ = max( # # df['i_high'].iat[-1], # # df['i_low'].iat[-1], # df['i_open'].iat[-1], # df['i_close'].iat[-1] # ) print (f"Temporarily disabled confirm entry, rate {rate} < mean {mean} = long and viceversa") return True # if mean == 0: # return False # if side == 'long': # return rate < mean # if side == 'short': # return rate > mean # return False # def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float, # current_profit: float, **kwargs): # if trade.is_short: # print(f'Strong Female Character goes Short for {pair}, fine') # return self.custom_exit_short(pair, trade, current_time, current_rate, current_profit) # # print(f'Strong Female Character tries to go Short for {pair}, No way... only short allowed') # # print(f' holds:{num_holds:.0f} ({hpct:.2f}%) ' + \ # # f'buys:{num_buys:.0f} ({bpct:.2f}%) sells:{num_sells:.0f} ({spct:.2f}%)') # # return False # else: # print(f'Strong Female Character tries to go Long for {pair}, No way... only short allowed here') # return False # def custom_entry_price(self, pair: str, current_time: datetime, proposed_rate: float, # entry_tag: Optional[str], side: str, **kwargs) -> float: # # # https://www.freqtrade.io/en/stable/strategy-callbacks/#custom-order-entry-and-exit-price-example # # df, last_updated = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) # avg = ( # # df['i_high'].iat[-1] + # # df['i_low'].iat[-1] + # df['i_open'].iat[-1] + # df['i_close'].iat[-1] # ) / 2 # # min_ = min( # df['i_high'].iat[-1], # df['i_low'].iat[-1], # df['i_open'].iat[-1], # df['i_close'].iat[-1] # ) # max_ = max( # df['i_high'].iat[-1], # df['i_low'].iat[-1], # df['i_open'].iat[-1], # df['i_close'].iat[-1] # ) # # print( # '------------------- confirm_trade_entry -------------', # pair, proposed_rate, side, '---', min_, max_, avg # ) # # if side == 'long': # return min(proposed_rate, avg) # # if side == 'short': # return max(proposed_rate, avg) # # return avg