# source: https://raw.githubusercontent.com/apanton2001/prometheus-bot/03a61e8f3ec4126c7d359d8d9a43ef3b61471b4b/trading/ml_xgboost_strategy.py import pandas as pd import numpy as np # import talib # Removed TA-Lib dependency from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, CategoricalParameter import joblib # For loading the model import os # Define thresholds for target variable calculation (used during OFFLINE training) # These are hyperparameters for the problem definition TARGET_THRESHOLD_PCT = 0.005 # e.g., 0.5% price movement LOOKAHEAD_PERIODS = 4 # e.g., predict for 4 candles ahead class Github_apanton2001_prometheus_bot__ml_xgboost_strategy__20250408_054730(IStrategy): """ Trading strategy using a pre-trained XGBoost classifier. Features are calculated using standard technical indicators. The target variable (Buy/Sell/Hold based on future price movement) is calculated OFFLINE during model training. This strategy LOADS a pre-trained model and uses it for predictions. """ # --- Strategy Hyperparameters --- # Define model path as a parameter for flexibility model_path = CategoricalParameter( ['user_data/models/xgb_model.joblib', '/path/to/your/model.joblib'], default='user_data/models/xgb_model.joblib', space='buy' ) # Add other strategy hyperparameters from EnhancedMAStrategy if needed for reference # (These are NOT used by the ML model directly but might be useful context or for hybrid approaches) # --- Removed TA-Lib dependent parameters --- # fast_ma_period = IntParameter(5, 20, default=9, space='buy') # medium_ma_period = IntParameter(15, 30, default=21, space='buy') # slow_ma_period = IntParameter(40, 60, default=50, space='buy') # adx_period = IntParameter(10, 20, default=14, space='buy') # adx_threshold = IntParameter(20, 35, default=25, space='buy') # volume_ma_period = IntParameter(15, 30, default=20, space='buy') # Keeping volume MA - not TA-Lib # volume_factor = DecimalParameter(1.0, 2.5, default=1.5, space='buy') # Keeping volume factor # atr_period = IntParameter(10, 20, default=14, space='buy') # rsi_period = IntParameter(10, 20, default=14, space='buy') # --- Minimal ROI and Stoploss (Required by Freqtrade) --- minimal_roi = { "0": 0.10, # Example: 10% profit after 0 minutes (uses exit signal mainly) "60": 0.05, # Example: 5% after 1 hour "120": 0.02 # Example: 2% after 2 hours } stoploss = -0.10 # Example: 10% stoploss # Timeframe needs to be defined timeframe = '1h' # Example timeframe - should match data used for training # --- Model Loading --- _model = None @property def model(self): """Load the model lazily.""" if self._model is None: model_file = self.model_path.value if os.path.exists(model_file): try: self._model = joblib.load(model_file) print(f"Successfully loaded model from {model_file}") except Exception as e: print(f"Error loading model from {model_file}: {e}") # Decide how to handle failure: stop bot, fall back, etc. # For now, we'll let it fail later during prediction else: print(f"Model file not found at {model_file}. Prediction will fail.") return self._model # --- Feature Engineering (Indicators) --- def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: """ Calculate all technical indicators that will be used as features for the ML model. REMOVED TA-LIB INDICATORS. """ # --- Basic OHLCV features (Example - Expand this!) --- dataframe['price_change_pct'] = dataframe['close'].pct_change() dataframe['high_low_diff'] = dataframe['high'] - dataframe['low'] dataframe['close_open_diff'] = dataframe['close'] - dataframe['open'] # --- Volume features (Not TA-Lib based) --- dataframe['volume_ma_20'] = dataframe['volume'].rolling(window=20).mean() dataframe['volume_ratio'] = np.where( dataframe['volume_ma_20'] > 0, dataframe['volume'] / dataframe['volume_ma_20'], 1 ) # ---- Add more NON-TA-LIB features here! Examples: ---- # Rolling window features dataframe['rolling_mean_close_5'] = dataframe['close'].rolling(window=5).mean() dataframe['rolling_std_close_20'] = dataframe['close'].rolling(window=20).std() # Time-based features dataframe['hour_of_day'] = dataframe.index.hour dataframe['day_of_week'] = dataframe.index.dayofweek return dataframe # --- Prediction Logic (Buy/Sell Signals) --- def populate_buy_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: """ Use the loaded ML model to predict the buy signal. Target classes (example): 0=Hold, 1=Buy, 2=Sell """ dataframe['buy'] = 0 # Default to no buy signal if self.model is None: print("Model not loaded, cannot generate buy signals.") return dataframe # Prepare features for the model (ensure order and scaling matches training) # This needs to be robust and match the *exact* preprocessing used during training # UPDATE THIS LIST TO MATCH populate_indicators (NO TA-LIB) feature_columns = [ 'price_change_pct', 'high_low_diff', 'close_open_diff', 'volume_ma_20', 'volume_ratio', 'rolling_mean_close_5', 'rolling_std_close_20', 'hour_of_day', 'day_of_week' # Add all features used in training ] # Check if all required columns exist and handle potential NaNs if not all(col in dataframe.columns for col in feature_columns): print(f"Missing required feature columns for prediction: {dataframe.columns.tolist()} vs {feature_columns}") return dataframe # Select only the features the model was trained on features = dataframe[feature_columns] # Handle any remaining NaNs (e.g., fill with 0 or a mean, MUST match training) features = features.fillna(0) # Example: Fill NaNs with 0 - ADJUST AS NEEDED # Predict try: predictions = self.model.predict(features) # Use predict_proba for probabilities if needed # Assuming target 1 means "Buy" dataframe.loc[predictions == 1, 'buy'] = 1 except Exception as e: print(f"Error during model prediction (buy): {e}") # Potentially add error handling or fallback logic # print(f"Buy signals generated: {dataframe['buy'].sum()}") # Basic logging return dataframe def populate_sell_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: """ Use the loaded ML model to predict the sell signal. Target classes (example): 0=Hold, 1=Buy, 2=Sell """ dataframe['sell'] = 0 # Default to no sell signal if self.model is None: print("Model not loaded, cannot generate sell signals.") return dataframe # Prepare features (same process as in populate_buy_trend) # UPDATE THIS LIST TO MATCH populate_indicators (NO TA-LIB) feature_columns = [ 'price_change_pct', 'high_low_diff', 'close_open_diff', 'volume_ma_20', 'volume_ratio', 'rolling_mean_close_5', 'rolling_std_close_20', 'hour_of_day', 'day_of_week' # Add all features used in training ] if not all(col in dataframe.columns for col in feature_columns): print(f"Missing required feature columns for prediction: {dataframe.columns.tolist()} vs {feature_columns}") return dataframe features = dataframe[feature_columns].fillna(0) # MUST match training # Predict try: predictions = self.model.predict(features) # Assuming target 2 means "Sell" dataframe.loc[predictions == 2, 'sell'] = 1 except Exception as e: print(f"Error during model prediction (sell): {e}") # print(f"Sell signals generated: {dataframe['sell'].sum()}") # Basic logging return dataframe # --- Optional: Custom Exit Logic (Can be ML-based too) --- # def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float, # current_profit: float, **kwargs): # """ # Optional: Implement custom exit logic, potentially using another ML model # or specific conditions based on the entry signal model's prediction confidence. # """ # # Example: Exit if prediction confidence drops below a threshold # return None # Return string reason to exit, or None to keep trade