# source: https://raw.githubusercontent.com/rainfrog123/blue/5461cb8077703b14c959d6971e86f1bebc238863/freq/project/user_data/strategies/trendat.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Standard Library Imports --- from datetime import datetime from typing import Optional, Union # --- Third Party Imports --- import numpy as np import pandas as pd from pandas import DataFrame import talib.abstract as ta # --- Freqtrade Imports --- from freqtrade.persistence import Trade from freqtrade.strategy import ( IStrategy, IntParameter, CategoricalParameter, DecimalParameter ) # --- Custom Strategy Class --- class Github_rainfrog123_blue__trendat__20250811_180631(IStrategy): """ Strategy for labeling trend reversals in trading data for machine learning tasks. """ INTERFACE_VERSION = 3 timeframe = '5s' can_short: bool = True # We use custom stoploss and exit logic - these are just placeholders minimal_roi = {"0": 100} # Very high ROI means we rely on custom_exit stoploss = -0.99 # This is a placeholder, real stoploss is calculated dynamically trailing_stop = False use_custom_stoploss = True startup_candle_count: int = 170 # ATR period parameter atr_period = 14 def informative_pairs(self): """ Define pairs for additional, informative data. Currently returns an empty list. """ return [] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Add all necessary technical indicators to the DataFrame. Currently includes Triple Exponential Moving Average (TEMA) and ATR. """ # Debug print to check incoming data pair = metadata['pair'] if not dataframe.empty: newest_candle = dataframe['date'].iloc[-1] candle_count = len(dataframe) print(f"DEBUG - {pair} - populate_indicators received {candle_count} candles. Newest: {newest_candle}") print(f"DEBUG - {pair} - Last 3 candles timestamps: {dataframe['date'].tail(3).values}") print(f"DEBUG - {pair} - Last 3 candles OHLCV: \n{dataframe[['date', 'open', 'high', 'low', 'close', 'volume']].tail(3)}") # Print data from approximately 10 minutes ago (120 candles for 5s timeframe) current_time = pd.to_datetime(newest_candle) target_time = current_time - pd.Timedelta(minutes=10) # Find candles closest to 10 minutes ago historical_candles = dataframe[dataframe['date'] > target_time].head(5) if not historical_candles.empty: print(f"DEBUG - {pair} - Historical data from ~10 minutes ago (after {target_time}):") print(f"DEBUG - {pair} - Historical OHLCV data: \n{historical_candles[['date', 'open', 'high', 'low', 'close', 'volume']]}") else: print(f"DEBUG - {pair} - No historical data found from ~10 minutes ago") else: print(f"DEBUG - {pair} - populate_indicators received EMPTY dataframe") # Ensure we are getting 5s data if that's what this strategy part expects # If self.timeframe is '5s', this will just get the regular dataframe. # If self.timeframe was a shim (e.g., '1m'), this would fetch the 5s data. df_5s = dataframe # Default to the passed dataframe if self.dp and hasattr(self.dp, 'get_analyzed_dataframe') and self.timeframe != '5s': # This block would be relevant if we used a shim timeframe # For now, assuming self.timeframe IS '5s' due to the patch intention pass # df_5s, _ = self.dp.get_analyzed_dataframe(metadata['pair'], '5s') # TEMA indicator for trend direction tema_period = 50 # Increased for 5s timeframe to maintain similar time window df_5s['tema'] = ta.TEMA(df_5s['close'], timeperiod=tema_period) df_5s['trend'] = np.where(df_5s['tema'] > df_5s['tema'].shift(1), 'UP', 'DOWN') df_5s['trend_duration'] = (df_5s['trend'] != df_5s['trend'].shift(1)).cumsum() df_5s['trend_count'] = df_5s.groupby('trend_duration').cumcount() + 1 # ATR calculation for stop loss and take profit df_5s['atr'] = ta.ATR(df_5s['high'], df_5s['low'], df_5s['close'], timeperiod=self.atr_period) return df_5s # Return the (potentially 5s) dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Populate the entry signals based on trend reversal logic. Signals for long and short entries based on trend direction changes. """ # Debug print for entry signals if not dataframe.empty: # Check if we have any entry signals in the last 10 candles last_10 = dataframe.tail(10).copy() # Fixed boolean operations long_signals = (last_10['trend'] == 'UP') & (last_10['trend'].shift(1) == 'DOWN') short_signals = (last_10['trend'] == 'DOWN') & (last_10['trend'].shift(1) == 'UP') if long_signals.any() or short_signals.any(): print(f"DEBUG - {metadata['pair']} - Entry signals detected in last 10 candles:") if long_signals.any(): print(f"DEBUG - Long entry at: {last_10.loc[long_signals, 'date'].values}") if short_signals.any(): print(f"DEBUG - Short entry at: {last_10.loc[short_signals, 'date'].values}") # Apply long entry conditions long_conditions = (dataframe['trend'] == 'UP') & (dataframe['trend'].shift(1) == 'DOWN') dataframe.loc[long_conditions, 'enter_long'] = 1 # Short entry signals short_conditions = (dataframe['trend'] == 'DOWN') & (dataframe['trend'].shift(1) == 'UP') dataframe.loc[short_conditions, 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Populate the exit signals for the strategy. This method is left empty as we use custom_stoploss and custom_exit for exits. """ return dataframe def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Custom stoploss logic. Returns the stoploss percentage as a positive value between 0.0 and 1.0. 0.01 would be a 1% loss. """ # Explicitly request 5s data from the dataprovider dataframe_5s, _ = self.dp.get_analyzed_dataframe(pair, '5s') # Debug print to check dataframe in custom_stoploss print(f"DEBUG - {pair} - custom_stoploss called at {current_time}") if dataframe_5s is not None and not dataframe_5s.empty: newest_candle = dataframe_5s['date'].iloc[-1] candle_count = len(dataframe_5s) time_diff = (pd.to_datetime(current_time) - pd.to_datetime(newest_candle)).total_seconds() print(f"DEBUG - {pair} - custom_stoploss has {candle_count} candles. Newest: {newest_candle}") print(f"DEBUG - {pair} - Data age: {time_diff:.1f} seconds") print(f"DEBUG - {pair} - Last candle: {dataframe_5s[['date', 'open', 'high', 'low', 'close', 'volume', 'atr']].iloc[-1].to_dict()}") else: print(f"DEBUG - {pair} - custom_stoploss received EMPTY dataframe") if dataframe_5s is not None and len(dataframe_5s) > 0: # Get current ATR value from the 5s dataframe current_atr = dataframe_5s['atr'].iloc[-1] # Calculate stoploss percentage based on entry price and ATR # 0.25 * ATR defines our risk (stoploss distance) atr_stoploss_distance = 0.25 * current_atr # Convert from absolute price distance to percentage stoploss_percent = atr_stoploss_distance / trade.open_rate return stoploss_percent # Default fallback (1% stoploss) return 0.01 def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs): """ Custom exit logic for take profit based on 0.5 * ATR (2:1 reward:risk ratio). Returns exit tag (string) if exit should occur, or None otherwise. """ # Explicitly request 5s data from the dataprovider dataframe_5s, _ = self.dp.get_analyzed_dataframe(pair, '5s') # Debug print to check dataframe in custom_exit print(f"DEBUG - {pair} - custom_exit called at {current_time}") if dataframe_5s is not None and not dataframe_5s.empty: newest_candle = dataframe_5s['date'].iloc[-1] candle_count = len(dataframe_5s) time_diff = (pd.to_datetime(current_time) - pd.to_datetime(newest_candle)).total_seconds() print(f"DEBUG - {pair} - custom_exit has {candle_count} candles. Newest: {newest_candle}") print(f"DEBUG - {pair} - Data age: {time_diff:.1f} seconds") print(f"DEBUG - {pair} - Last candle: {dataframe_5s[['date', 'open', 'high', 'low', 'close', 'volume', 'atr']].iloc[-1].to_dict()}") else: print(f"DEBUG - {pair} - custom_exit received EMPTY dataframe") if dataframe_5s is not None and len(dataframe_5s) > 0: # Get current ATR value from the 5s dataframe current_atr = dataframe_5s['atr'].iloc[-1] # Calculate take profit distance (0.5 * ATR) atr_take_profit_distance = 0.5 * current_atr # Convert distance to percentage take_profit_percent = atr_take_profit_distance / trade.open_rate # Check if current profit exceeds our take profit level # This gives us 2:1 reward:risk (0.5 ATR vs 0.25 ATR) if current_profit >= take_profit_percent: print(f"DEBUG - {pair} - Taking profit at {current_profit:.4f}% (target: {take_profit_percent:.4f}%)") return 'take_profit_atr' return None 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 the leverage to use for a trade. """ return max_leverage # return 100.0 def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float | None, max_stake: float, leverage: float, entry_tag: str | None, side: str, **kwargs) -> float: """ Custom stake amount logic - uses 75% of the total available balance. """ # Fetch the total balance in the stake currency # total_balance = self.wallets.get_total_balance(self.config['stake_currency']) total_balance = self.wallets.get_total_stake_amount() # Calculate 3/4 (75%) of the total balance stake_amount = total_balance * 0.75 # Ensure the stake amount is within allowed limits return stake_amount