# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file from freqtrade.strategy import IStrategy from pandas import DataFrame import logging from datetime import datetime, timedelta import talib.abstract as ta import pandas as pd logger = logging.getLogger(__name__) class Test5s(IStrategy): INTERFACE_VERSION = 3 timeframe = '5s' can_short: bool = True # Only process new candles, not all historical data process_only_new_candles = True # More realistic ROI: 1% profit target minimal_roi = {"0": 0.01, "10": 0.005, "30": 0} # Tighter stop loss for 5s timeframe stoploss = -0.01 trailing_stop = False # We'll use custom trailing stop use_custom_stoploss = True # Enable ATR-based trailing stop startup_candle_count: int = 60 # Need more candles for TEMA calculation (50 + buffer) # Manual confirmation system pending_confirmations = {} # Track when we last sent notifications confirmation_cooldown = timedelta(minutes=1) # Prevent spam - 1 minute between notifications def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # TEMA calculation (Triple Exponential Moving Average) tema_period = 50 # Calculate the three EMAs for TEMA ema1 = ta.EMA(dataframe['close'], timeperiod=tema_period) ema2 = ta.EMA(ema1, timeperiod=tema_period) ema3 = ta.EMA(ema2, timeperiod=tema_period) # TEMA formula: 3 * EMA1 - 3 * EMA2 + EMA3 dataframe['tema'] = 3 * ema1 - 3 * ema2 + ema3 # Calculate trend direction dataframe['tema_prev'] = dataframe['tema'].shift(1) dataframe['trend_up'] = dataframe['tema'] > dataframe['tema_prev'] dataframe['trend_down'] = dataframe['tema'] < dataframe['tema_prev'] dataframe['trend_stable'] = (dataframe['tema'] == dataframe['tema_prev']) # Track trend changes dataframe['prev_trend_up'] = dataframe['trend_up'].shift(1) dataframe['prev_trend_down'] = dataframe['trend_down'].shift(1) # Detect trend reversals dataframe['reversal_to_up'] = (~dataframe['prev_trend_up']) & dataframe['trend_up'] dataframe['reversal_to_down'] = (~dataframe['prev_trend_down']) & dataframe['trend_down'] # ATR for trailing stop calculation dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Get pair name for logging pair = metadata['pair'] # MANUAL CONFIRMATION SYSTEM - No automatic entries dataframe['enter_long'] = 0 dataframe['enter_short'] = 0 # Check for TEMA reversal signals on the latest candle only if not dataframe.empty and len(dataframe) > 1: current_time = datetime.now() last_notification = self.pending_confirmations.get(pair) # Check for reversal signals long_signal = dataframe['reversal_to_up'].iloc[-1] if not dataframe['reversal_to_up'].isna().iloc[-1] else False short_signal = dataframe['reversal_to_down'].iloc[-1] if not dataframe['reversal_to_down'].isna().iloc[-1] else False if (long_signal or short_signal) and (last_notification is None or (current_time - last_notification > self.confirmation_cooldown)): # Get signal details current_price = dataframe['close'].iloc[-1] current_tema = dataframe['tema'].iloc[-1] if not dataframe['tema'].isna().iloc[-1] else 0 prev_tema = dataframe['tema_prev'].iloc[-1] if not dataframe['tema_prev'].isna().iloc[-1] else 0 tema_change = current_tema - prev_tema if prev_tema != 0 else 0 # Determine signal type signal_type = "LONG" if long_signal else "SHORT" signal_emoji = "🟢" if long_signal else "🔴" trend_direction = "UP" if long_signal else "DOWN" command = f"/forcelong {pair}" if long_signal else f"/forceshort {pair}" # Create confirmation message message = ( f"{signal_emoji} TEMA REVERSAL SIGNAL {signal_emoji}\n" f"📊 Pair: {pair}\n" f"💰 Price: ${current_price:.4f}\n" f"📈 TEMA: {current_tema:.4f}\n" f"🔄 Change: {tema_change:.4f}\n" f"📊 Trend: {trend_direction}\n" f"🎯 Signal: {signal_type}\n" f"⏰ Time: {current_time.strftime('%H:%M:%S')}\n\n" f"💡 To enter trade, copy and send:\n" f"`{command}`" ) # Send Telegram notification if self.dp and hasattr(self.dp, 'send_msg'): try: self.dp.send_msg(message) self.pending_confirmations[pair] = current_time logger.info(f"🎯 Sent TEMA {signal_type} confirmation request for {pair} (TEMA: {current_tema:.4f}, Change: {tema_change:.4f})") except Exception as e: logger.error(f"Failed to send Telegram message: {e}") else: logger.warning("DataProvider send_msg not available - check Telegram config") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # No exit signals - using only ATR-based trailing stop dataframe['exit_long'] = 0 dataframe['exit_short'] = 0 return dataframe def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Custom ATR-based trailing stoploss. Uses ATR * 0.5 as the trailing stop distance. """ # Get the latest analyzed dataframe dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or dataframe.empty: # Fallback to initial stoploss if no data return self.stoploss # Get the latest ATR value latest_atr = dataframe['atr'].iloc[-1] if pd.isna(latest_atr) or latest_atr <= 0: # Fallback to initial stoploss if ATR is invalid return self.stoploss # Calculate ATR-based stop distance (ATR * 0.5) atr_stop_distance = latest_atr * 0.5 # Convert to percentage of current rate stop_percentage = atr_stop_distance / current_rate # For long positions, return negative percentage (stop below current price) # For short positions, return positive percentage (stop above current price) if trade.is_short: return stop_percentage else: return -stop_percentage def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str, **kwargs) -> float: """ Customize leverage for each new trade. This method is only called in futures mode. """ return max_leverage