# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these libs --- import numpy as np import pandas as pd from pandas import DataFrame from datetime import datetime from typing import Optional from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, CategoricalParameter from freqtrade.strategy import stoploss_from_open import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from scipy.linalg import pinv from scipy.spatial.distance import cdist import warnings warnings.filterwarnings('ignore') ''' ______ __ __ __ __ ______ __ __ __ __ __ ______ / \ / | _/ | / | / | / \ / \ / | / | / | / | / \ /$$$$$$ |$$ |____ / $$ | _______ $$ | /$$/ /$$$$$$ |$$ \ $$ | _$$ |_ $$ | $$ | _______ /$$$$$$ | _______ $$ | $$/ $$ \ $$$$ | / |$$ |/$$/ $$ ___$$ |$$$ \$$ | / $$ | $$ |__$$ | / |$$$ \$$ | / | $$ | $$$$$$$ | $$ | /$$$$$$$/ $$ $$< / $$< $$$$ $$ | $$$$$$/ $$ $$ |/$$$$$$$/ $$$$ $$ |/$$$$$$$/ $$ | __ $$ | $$ | $$ | $$ | $$$$$ \ _$$$$$ |$$ $$ $$ | $$ | __$$$$$$$$ |$$ | $$ $$ $$ |$$ \ $$ \__/ |$$ | $$ | _$$ |_ $$ \_____ $$ |$$ \ / \__$$ |$$ |$$$$ | $$ |/ | $$ |$$ \_____ $$ \$$$$ | $$$$$$ | $$ $$/ $$ | $$ |/ $$ |$$ |$$ | $$ |$$ $$/ $$ | $$$ |______$$ $$/ $$ |$$ |$$ $$$/ / $$/ $$$$$$/ $$/ $$/ $$$$$$/ $$$$$$$/ $$/ $$/ $$$$$$/ $$/ $$// |$$$$/ $$/ $$$$$$$/ $$$$$$/ $$$$$$$/ $$$$$$/ ''' class GPR(IStrategy): """ Gaussian Process Regression Strategy This strategy uses Gaussian Process Regression to predict future price movements and generates buy/sell signals based on the forecasted trend direction. Original Pine Script by LuxAlgo - adapted for Freqtrade """ # Strategy interface version INTERFACE_VERSION = 3 # Timeframe for the strategy - works best on higher timeframes timeframe = '1h' # Can this strategy go short? can_short = False # Minimal ROI designed for the strategy minimal_roi = { "0": 0.10, "40": 0.05, "100": 0.02, "200": 0.01 } # Optimal stoploss stoploss = -0.05 # Trailing stoploss trailing_stop = True trailing_stop_positive = 0.01 trailing_stop_positive_offset = 0.02 trailing_only_offset_is_reached = True # Strategy parameters window = IntParameter(50, 200, default=200, space="buy", optimize=True) forecast_length = IntParameter(10, 50, default=20, space="buy", optimize=True) smooth_length = DecimalParameter(10.0, 50.0, default=20.0, space="buy", optimize=True) sigma = DecimalParameter(0.01, 0.1, default=0.51, decimals=2, space="buy", optimize=True) # Signal thresholds buy_threshold = DecimalParameter(0.001, 0.02, default=0.005, decimals=3, space="buy", optimize=True) sell_threshold = DecimalParameter(0.001, 0.02, default=0.005, decimals=3, space="sell", optimize=True) # Required startup candles startup_candle_count: int = 250 def __init__(self, config: dict) -> None: super().__init__(config) # Cache for storing model predictions self._model_cache = {} def rbf_kernel(self, x1: np.ndarray, x2: np.ndarray, length_scale: float) -> np.ndarray: """ Radial Basis Function (RBF) kernel implementation """ # Calculate squared Euclidean distances distances = cdist(x1.reshape(-1, 1), x2.reshape(-1, 1), 'sqeuclidean') return np.exp(-distances / (2.0 * length_scale ** 2)) def gaussian_process_regression(self, y_train: np.ndarray, window: int, forecast_length: int, smooth_length: float, sigma: float) -> tuple: """ Perform Gaussian Process Regression Returns: tuple: (fitted_values, forecast_values) """ try: # Create training and test indices x_train = np.arange(window) x_test = np.arange(window + forecast_length) # Compute kernel matrices K_train = self.rbf_kernel(x_train, x_train, smooth_length) K_star = self.rbf_kernel(x_train, x_test, smooth_length) # Add noise to diagonal (regularization) K_train_reg = K_train + sigma ** 2 * np.eye(window) # Compute inverse of regularized kernel matrix try: K_inv = pinv(K_train_reg) except: return None, None # Compute predictions mu = K_star.T @ K_inv @ y_train # Split into fitted and forecast parts fitted_values = mu[:window] forecast_values = mu[window:] if forecast_length > 0 else np.array([]) return fitted_values, forecast_values except Exception as e: self.dp.send_msg(f"GPR Error: {str(e)}") return None, None def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Add GPR indicators to the dataframe """ # Ensure we have enough data if len(dataframe) < self.window.value + self.forecast_length.value: return dataframe # Initialize columns dataframe['gpr_fitted'] = np.nan dataframe['gpr_forecast'] = np.nan dataframe['gpr_trend'] = 0 dataframe['gpr_signal'] = 0 dataframe['price_mean'] = ta.SMA(dataframe, timeperiod=self.window.value) # Calculate GPR for each row (only for recent data to save computation) start_idx = max(0, len(dataframe)-200) # Only calculate for last 100 bars for i in range(start_idx, len(dataframe)): if i < self.window.value: continue # Get training data end_idx = i + 1 start_idx_train = end_idx - self.window.value y_data = dataframe['close'].iloc[start_idx_train:end_idx].values price_mean = dataframe['price_mean'].iloc[i] if np.isnan(price_mean): continue # Normalize data (subtract mean) y_train = y_data - price_mean # Perform GPR fitted, forecast = self.gaussian_process_regression( y_train, self.window.value, self.forecast_length.value, self.smooth_length.value, self.sigma.value ) if fitted is not None and forecast is not None: # Store fitted value (add back the mean) dataframe.loc[dataframe.index[i], 'gpr_fitted'] = fitted[-1] + price_mean # Store forecast trend if len(forecast) > 0: forecast_trend = forecast[-1] - forecast[0] if len(forecast) > 1 else forecast[0] dataframe.loc[dataframe.index[i], 'gpr_forecast'] = forecast[-1] + price_mean dataframe.loc[dataframe.index[i], 'gpr_trend'] = forecast_trend # Generate signals based on forecast trend current_price = dataframe['close'].iloc[i] forecast_price = forecast[-1] + price_mean price_change_pct = (forecast_price - current_price) / current_price if price_change_pct > self.buy_threshold.value: dataframe.loc[dataframe.index[i], 'gpr_signal'] = 1 # Buy signal elif price_change_pct < -self.sell_threshold.value: dataframe.loc[dataframe.index[i], 'gpr_signal'] = -1 # Sell signal # Add additional indicators for confirmation dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=12) dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=26) dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on GPR analysis, populate the entry trend for buys and sells """ conditions_long = [ (dataframe['gpr_signal'] == 1), # GPR predicts upward movement (dataframe['gpr_trend'] > 0), # Positive trend (dataframe['rsi'] < 70), # Not overbought (dataframe['ema_fast'] > dataframe['ema_slow']), # EMA confirmation (dataframe['volume'] > dataframe['volume_sma']), # Volume confirmation ] conditions_short = [ (dataframe['gpr_signal'] == -1), # GPR predicts downward movement (dataframe['gpr_trend'] < 0), # Negative trend (dataframe['rsi'] > 30), # Not oversold (dataframe['ema_fast'] < dataframe['ema_slow']), # EMA confirmation (dataframe['volume'] > dataframe['volume_sma']), # Volume confirmation ] # Combine all conditions if conditions_long: dataframe.loc[ reduce(lambda x, y: x & y, conditions_long), 'enter_long' ] = 1 if conditions_short and self.can_short: dataframe.loc[ reduce(lambda x, y: x & y, conditions_short), 'enter_short' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on GPR analysis, populate the exit trend for buys and sells """ conditions_exit_long = [ (dataframe['gpr_signal'] == -1) | # GPR predicts downward movement (dataframe['gpr_trend'] < -0.001) | # Negative trend developing (dataframe['rsi'] > 75) # Overbought ] conditions_exit_short = [ (dataframe['gpr_signal'] == 1) | # GPR predicts upward movement (dataframe['gpr_trend'] > 0.001) | # Positive trend developing (dataframe['rsi'] < 25) # Oversold ] if conditions_exit_long: dataframe.loc[ reduce(lambda x, y: x | y, conditions_exit_long), 'exit_long' ] = 1 if conditions_exit_short and self.can_short: dataframe.loc[ reduce(lambda x, y: x | y, conditions_exit_short), 'exit_short' ] = 1 return dataframe def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Custom stoploss implementation using GPR signals """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Use GPR signal for dynamic stoploss if 'gpr_signal' in last_candle: if trade.is_short and last_candle['gpr_signal'] == 1: return 0.01 # Exit short position if strong buy signal elif not trade.is_short and last_candle['gpr_signal'] == -1: return 0.01 # Exit long position if strong sell signal return self.stoploss 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: """ Confirm trade entry based on GPR confidence """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return False last_candle = dataframe.iloc[-1].squeeze() # Check if we have valid GPR data if pd.isna(last_candle.get('gpr_signal', np.nan)): return False # Additional confirmation for entry if side == "long": return (last_candle['gpr_signal'] == 1 and last_candle['gpr_trend'] > 0) elif side == "short": return (last_candle['gpr_signal'] == -1 and last_candle['gpr_trend'] < 0) return True # Helper function for reducing conditions def reduce(function, iterable, initializer=None): """ Python's built-in reduce function """ it = iter(iterable) if initializer is None: value = next(it) else: value = initializer for element in it: value = function(value, element) return value