# source: https://raw.githubusercontent.com/gilaigenttech/freqtrade-user_data/91e5f711fbe037478e7709fcd3247b39b2c54694/strategies/AlexStrategyFinalV9.py import numpy as np import pandas_ta as pta from typing import Optional import math import freqtrade.vendor.qtpylib.indicators as qtpylib import logging from functools import reduce from typing import Dict from datetime import timedelta, datetime, timezone from freqtrade.persistence import Trade from pandas import DataFrame, Series import talib.abstract as ta from pandas import DataFrame from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy, BooleanParameter from technical.pivots_points import pivots_points from scipy.signal import argrelextrema import logging from functools import reduce import numpy as np import pandas as pd import talib.abstract as ta from pandas import DataFrame from technical import qtpylib from freqtrade.exchange.exchange_utils import * from freqtrade.strategy import IStrategy, RealParameter from technical.pivots_points import pivots_points logger = logging.getLogger(__name__) """ Github_gilaigenttech_freqtrade_user_data__AlexStrategyFinalV9__20251003_074033 - FreqAI Strategy with Advanced Technical Analysis This strategy combines multiple technical indicators with FreqAI predictions and includes dynamic features like: - Multi-indicator normalized scoring - Market regime filtering - Volatility-based adjustments - Dynamic trailing stops and leverage """ class Github_gilaigenttech_freqtrade_user_data__AlexStrategyFinalV9__20251003_074033(IStrategy): """ This is an example strategy that uses the LSTMRegressor model to predict the target score. Use at your own risk. This is a simple example strategy and should be used for educational purposes only. """ # Hyperspace parameters: buy_params = { "threshold_buy": 0.59453, "w0": 0.54347, "w1": 0.82226, "w2": 0.56675, "w3": 0.77918, "w4": 0.98488, "w5": 0.31368, "w6": 0.75916, "w7": 0.09226, "w8": 0.85667, } sell_params = { "threshold_sell": 0.80573, } # ROI table: minimal_roi = { "0": 0.239, "79": 0.058, "231": 0.029, "543": 0 } # Stoploss: stoploss = -0.305 # Were letting the model decide when to sell # Trailing stop: #trailing_stop = False #trailing_only_offset_is_reached = False timeframe = "1h" can_short = True use_exit_signal = True process_only_new_candles = True exit_profit_only = False exit_profit_offset = 0.01 ignore_roi_if_entry_signal = False startup_candle_count = 100 leverage_value = 10.0 threshold_buy = RealParameter(-1, 1, default=0, space='buy') threshold_sell = RealParameter(-1, 1, default=0, space='sell') # Weights for calculating the aggregate score - normalized to sum to 1 w0 = RealParameter(0, 1, default=0.10, space='buy') # moving average (normalized_ma) w1 = RealParameter(0, 1, default=0.15, space='buy') # MACD (normalized_macd) w2 = RealParameter(0, 1, default=0.10, space='buy') # Rate of Change (ROC) w3 = RealParameter(0, 1, default=0.15, space='buy') # RSI (normalized_rsi) w4 = RealParameter(0, 1, default=0.10, space='buy') # Bollinger Band width w5 = RealParameter(0, 1, default=0.10, space='buy') # CCI (normalized_cci) w6 = RealParameter(0, 1, default=0.10, space='buy') # OBV (normalized_obv) w7 = RealParameter(0, 1, default=0.05, space='buy') # ATR (normalized_atr) w8 = RealParameter(0, 1, default=0.10, space='buy') # Stochastic Oscillator (normalized_stoch) plot_config = { 'main_plot': { 'ma_100': {'color': 'blue'}, 'bb_upperband': {'color': 'orange'}, 'bb_middleband': {'color': 'yellow'}, 'bb_lowerband': {'color': 'orange'}, }, 'subplots': { "Target": { 'T': {'color': 'blue'}, '&-target': {'color': 'green'}, }, "Scores": { 'S': {'color': 'purple'}, }, "Filters": { 'R': {'color': 'red'}, 'R2': {'color': 'brown'}, 'V': {'color': 'green'}, 'V2': {'color': 'black'}, }, "Normalized Indicators": { 'normalized_ma': {'color': 'blue'}, 'normalized_macd': {'color': 'red'}, 'normalized_rsi': {'color': 'green'}, } } } def feature_engineering_expand_all(self, dataframe: DataFrame, period: int, metadata: Dict, **kwargs): """ Feature engineering for the AI model using rolling windows """ # Basic technical indicators with rolling periods dataframe["%-cci-period"] = ta.CCI(dataframe, timeperiod=20) dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=10) dataframe["%-momentum-period"] = ta.MOM(dataframe, timeperiod=4) dataframe['%-ma-period'] = ta.SMA(dataframe, timeperiod=10) dataframe['%-macd-period'], dataframe['%-macdsignal-period'], dataframe['%-macdhist-period'] = ta.MACD( dataframe['close'], slowperiod=12, fastperiod=26) dataframe['%-roc-period'] = ta.ROC(dataframe, timeperiod=2) # Bollinger Bands bollinger = qtpylib.bollinger_bands( qtpylib.typical_price(dataframe), window=period, stds=2.2 ) dataframe["bb_lowerband-period"] = bollinger["lower"] dataframe["bb_middleband-period"] = bollinger["mid"] dataframe["bb_upperband-period"] = bollinger["upper"] dataframe["%-bb_width-period"] = ( dataframe["bb_upperband-period"] - dataframe["bb_lowerband-period"] ) / dataframe["bb_middleband-period"] dataframe["%-close-bb_lower-period"] = ( dataframe["close"] / dataframe["bb_lowerband-period"] ) return dataframe def feature_engineering_expand_basic(self, dataframe: DataFrame, metadata: Dict, **kwargs): """ Basic feature engineering """ dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-raw_volume"] = dataframe["volume"] dataframe["%-raw_price"] = dataframe["close"] return dataframe def feature_engineering_standard(self, dataframe: DataFrame, metadata: Dict, **kwargs): """ Standard feature engineering with time-based features """ dataframe['date'] = pd.to_datetime(dataframe['date']) dataframe["%-day_of_week"] = dataframe["date"].dt.dayofweek dataframe["%-hour_of_day"] = dataframe["date"].dt.hour return dataframe def set_freqai_targets(self, dataframe: DataFrame, metadata: Dict, **kwargs) -> DataFrame: """ Calculate the target variable for FreqAI prediction This is the core of the strategy's logic """ # Calculate basic technical indicators dataframe['ma'] = ta.SMA(dataframe, timeperiod=10) dataframe['roc'] = ta.ROC(dataframe, timeperiod=2) dataframe['macd'], dataframe['macdsignal'], dataframe['macdhist'] = ta.MACD(dataframe['close'], slowperiod=12, fastperiod=26) dataframe['momentum'] = ta.MOM(dataframe, timeperiod=4) dataframe['rsi'] = ta.RSI(dataframe, timeperiod=10) bollinger = ta.BBANDS(dataframe, timeperiod=20) dataframe['bb_upperband'] = bollinger['upperband'] dataframe['bb_middleband'] = bollinger['middleband'] dataframe['bb_lowerband'] = bollinger['lowerband'] dataframe['cci'] = ta.CCI(dataframe, timeperiod=20) dataframe['stoch'] = ta.STOCH(dataframe)['slowk'] dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['obv'] = ta.OBV(dataframe) dataframe['ma_100'] = ta.SMA(dataframe, timeperiod=100) # Step 1: Normalize Indicators: # Why? Normalizing the indicators will make them comparable and allow us to assign weights to them. # How? We will calculate the z-score of each indicator by subtracting the rolling mean and dividing by the # rolling standard deviation. This will give us a normalized value that is centered around 0 with a standard # deviation of 1. dataframe['normalized_stoch'] = (dataframe['stoch'] - dataframe['stoch'].rolling(window=14).mean()) / dataframe['stoch'].rolling(window=14).std() dataframe['normalized_atr'] = (dataframe['atr'] - dataframe['atr'].rolling(window=14).mean()) / dataframe['atr'].rolling(window=14).std() dataframe['normalized_obv'] = (dataframe['obv'] - dataframe['obv'].rolling(window=14).mean()) / dataframe['obv'].rolling(window=14).std() dataframe['normalized_ma'] = (dataframe['close'] - dataframe['close'].rolling(window=10).mean()) / dataframe['close'].rolling(window=10).std() dataframe['normalized_macd'] = (dataframe['macd'] - dataframe['macd'].rolling(window=26).mean()) / dataframe['macd'].rolling(window=26).std() dataframe['normalized_roc'] = (dataframe['roc'] - dataframe['roc'].rolling(window=2).mean()) / dataframe['roc'].rolling(window=2).std() dataframe['normalized_momentum'] = (dataframe['momentum'] - dataframe['momentum'].rolling(window=4).mean()) / \ dataframe['momentum'].rolling(window=4).std() dataframe['normalized_rsi'] = (dataframe['rsi'] - dataframe['rsi'].rolling(window=10).mean()) / dataframe['rsi'].rolling(window=10).std() dataframe['normalized_bb_width'] = (dataframe['bb_upperband'] - dataframe['bb_lowerband']).rolling( window=20).mean() / (dataframe['bb_upperband'] - dataframe['bb_lowerband']).rolling(window=20).std() dataframe['normalized_cci'] = (dataframe['cci'] - dataframe['cci'].rolling(window=20).mean()) / dataframe['cci'].rolling(window=20).std() # Dynamic Weights Adjustment # Calculate trend strength as the absolute difference between MA and close price trend_strength = abs(dataframe['ma'] - dataframe['close']) # Calculate rolling mean and stddev once to avoid redundancy rolling_mean = trend_strength.rolling(window=14).mean() rolling_stddev = trend_strength.rolling(window=14).std() # Calculate a more dynamic strong trend threshold strong_trend_threshold = rolling_mean + 1.5 * rolling_stddev # Determine strong trend condition is_strong_trend = trend_strength > strong_trend_threshold # Apply dynamic weight adjustment, could also consider a more gradual adjustment dataframe['w_momentum'] = self.w3.value * (1 + 0.5 * (trend_strength / strong_trend_threshold)) # Optional: Clip the w_momentum values to prevent extreme cases dataframe['w_momentum'] = dataframe['w_momentum'].clip(lower=self.w3.value, upper=self.w3.value * 2) # Step 2: Calculate aggregate score S w = [self.w0.value, self.w1.value, self.w2.value, self.w3.value, self.w4.value, self.w5.value, self.w6.value, self.w7.value, self.w8.value] dataframe['S'] = w[0] * dataframe['normalized_ma'] + \ w[1] * dataframe['normalized_macd'] + \ w[2] * dataframe['normalized_roc'] + \ w[3] * dataframe['normalized_rsi'] + \ w[4] * dataframe['normalized_bb_width'] + \ w[5] * dataframe['normalized_cci'] + \ dataframe['w_momentum'] * dataframe['normalized_momentum'] + \ self.w8.value * dataframe['normalized_stoch'] + \ self.w7.value * dataframe['normalized_atr'] + \ self.w6.value * dataframe['normalized_obv'] # Step 3: Market Regime Filter R dataframe['R'] = 0 dataframe.loc[dataframe['close'] > dataframe['bb_upperband'], 'R'] = 1 dataframe.loc[dataframe['close'] < dataframe['bb_lowerband'], 'R'] = -1 buffer_pct = 0.01 # 1% buffer dataframe['R2'] = np.where(dataframe['close'] > dataframe['ma_100'] * (1 + buffer_pct), 1, np.where(dataframe['close'] < dataframe['ma_100'] * (1 - buffer_pct), -1, np.nan)) # Step 4: Volatility Adjustment V # EXPLANATION: Calculate the Bollinger Band width and assign it to V. The Bollinger Band width is the # difference between the upper and lower Bollinger Bands divided by the middle Bollinger Band. The idea is # that when the Bollinger Bands are wide, the market is volatile, and when the Bollinger Bands are narrow, # the market is less volatile. So we are using the Bollinger Band width as a measure of volatility. You can # use other indicators to measure volatility as well. For example, you can use the ATR (Average True Range) bb_width = (dataframe['bb_upperband'] - dataframe['bb_lowerband']) / dataframe['bb_middleband'] dataframe['V_mean'] = 1 / (bb_width + 1e-8) # Avoid division by zero # ATR-based Volatility Measure dataframe['V2_mean'] = 1 / (dataframe['atr'] + 1e-8) # Avoid division by zero # Rolling window size for adaptive normalization rolling_window = 50 # Normalize V_mean using a rolling window mean_v = dataframe['V_mean'].rolling(window=rolling_window).mean() std_v = dataframe['V_mean'].rolling(window=rolling_window).std() dataframe['V_norm'] = (dataframe['V_mean'] - mean_v) / std_v dataframe['V_norm'] = dataframe['V_norm'].fillna(0) # Normalize V2_mean using a rolling window mean_v2 = dataframe['V2_mean'].rolling(window=rolling_window).mean() std_v2 = dataframe['V2_mean'].rolling(window=rolling_window).std() dataframe['V2_norm'] = (dataframe['V2_mean'] - mean_v2) / std_v2 dataframe['V2_norm'] = dataframe['V2_norm'].fillna(0) # Signal assignment using hysteresis upper_threshold = 1.0 lower_threshold = -1.0 dataframe['V'] = np.where(dataframe['V_norm'] > upper_threshold, 1, np.where(dataframe['V_norm'] < lower_threshold, -1, np.nan)) dataframe['V2'] = np.where(dataframe['V2_norm'] > upper_threshold, 1, np.where(dataframe['V2_norm'] < lower_threshold, -1, np.nan)) # Forward-fill to maintain the last state of the signal dataframe['V'] = dataframe['V'].ffill() # Correct ffill usage dataframe['V2'] = dataframe['V2'].ffill() # Correct ffill usage # Get Final Target Score to incorporate new calculations dataframe['T'] = dataframe['S'] * dataframe['R'] * dataframe['R2'] * dataframe['V'] * dataframe['V2'] # Assign the target score T to the AI target column target_horizon = 1 # Define your prediction horizon here dataframe['&-target'] = dataframe['T'].shift(-target_horizon) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ This is where FreqAI is called to make predictions """ self.freqai_info = self.config["freqai"] dataframe = self.freqai.start(dataframe, metadata, self) # One can define indicators here if needed and add logic to populate_entry_trend and populate_exit_trend # dataframe["target_roi"] = dataframe["&-target_mean"] + dataframe["&-target_std"] * 1.25 # dataframe["sell_roi"] = dataframe["&-target_mean"] - dataframe["&-target_std"] * 1.25 return dataframe def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: """ Entry signal logic based on FreqAI predictions """ enter_long_conditions = [ df["do_predict"] == 1, df['&-target'] > self.threshold_buy.value, df['volume'] > 0 ] enter_short_conditions = [ df["do_predict"] == 1, df['&-target'] < self.threshold_sell.value, df["volume"] > 0 ] df.loc[ reduce(lambda x, y: x & y, enter_long_conditions), ["enter_long", "enter_tag"] ] = (1, "long") df.loc[ reduce(lambda x, y: x & y, enter_short_conditions), ["enter_short", "enter_tag"] ] = (1, "short") return df def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: """ Exit signal logic based on FreqAI predictions """ exit_long_conditions = [ df['&-target'] < self.threshold_sell.value ] exit_short_conditions = [ df['&-target'] > self.threshold_buy.value ] if exit_long_conditions: df.loc[ reduce(lambda x, y: x & y, exit_long_conditions), ["exit_long", "exit_tag"] ] = (1, "exit_long") if exit_short_conditions: df.loc[ reduce(lambda x, y: x & y, exit_short_conditions), ["exit_short", "exit_tag"] ] = (1, "exit_short") return df def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Calculate custom stoploss based on dynamic trailing stop logic. """ dataframe, row = self.dp.get_analyzed_dataframe(pair, self.timeframe) # If no valid dataframe is available (initial trades) if dataframe is None or row is None: return self.stoploss # Simple ATR-based dynamic stoploss (you can enhance this) try: atr = dataframe['atr'].iat[-1] if 'atr' in dataframe.columns else None if atr and not pd.isna(atr): # Dynamic stoploss based on ATR dynamic_stop = min(-0.02, -(atr * 2) / current_rate) # 2x ATR or 2% minimum return max(self.stoploss, dynamic_stop) except (IndexError, KeyError): pass return self.stoploss def leverage( self, pair: str, current_time: "datetime", current_rate: float, proposed_leverage: float, **kwargs, ) -> float: """ Dynamically adjust leverage based on volatility (ATR). """ # Get the analyzed dataframe for the pair and timeframe dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or dataframe.empty: return self.leverage_value # Fallback to default leverage # Get the latest ATR value from the last candle try: atr = dataframe['atr'].iat[-1] # Get the ATR of the most recent candle except (IndexError, KeyError): return self.leverage_value # Fallback if ATR is not available if pd.isna(atr): return self.leverage_value # Fallback if ATR is NaN # Example: Lower leverage when volatility (ATR) is high try: rolling_mean = dataframe['atr'].rolling(window=14).mean().iat[-1] rolling_stddev = dataframe['atr'].rolling(window=14).std().iat[-1] if atr > rolling_mean + 1.5 * rolling_stddev: return self.leverage_value * 0.5 # Reduce leverage in volatile markets except (IndexError, KeyError): pass return self.leverage_value def chaikin_mf(df, periods=20): """ Chaikin Money Flow indicator """ close = df["close"] low = df["low"] high = df["high"] volume = df["volume"] mfv = ((close - low) - (high - close)) / (high - low) mfv = mfv.fillna(0.0) mfv *= volume cmf = mfv.rolling(periods).sum() / volume.rolling(periods).sum() return Series(cmf, name="cmf") def top_percent_change(dataframe: DataFrame, length: int) -> float: """ Calculate top percent change over a given length """ if length == 0: return (dataframe["open"] - dataframe["close"]) / dataframe["close"] else: return (dataframe["open"].rolling(length).max() - dataframe["close"]) / dataframe["close"]