"""Stacked tree-ensemble predictor for the 5-minute timeframe. A binary (up/down) classifier that stacks five gradient-boosted / bagged tree models with two meta-models and an HMM market-regime filter. """ import numpy as np import pandas as pd from sklearn.preprocessing import RobustScaler from sklearn.ensemble import RandomForestClassifier, ExtraTreesClassifier from sklearn.linear_model import LogisticRegression from sklearn.metrics import roc_auc_score from sklearn.model_selection import TimeSeriesSplit import xgboost as xgb import lightgbm as lgb from catboost import CatBoostClassifier from hmmlearn import hmm import talib class TreeEnsemblePredictor: def __init__(self, forecast_bars=24, confidence_threshold=0.65): """ Optimized gold price prediction model for 5-minute timeframe Parameters: forecast_bars (int): Number of future 5-min bars to predict (default=24, which is 2 hours) confidence_threshold (float): Minimum probability threshold for signal generation """ self.forecast_bars = forecast_bars self.confidence_threshold = confidence_threshold self.scaler = RobustScaler() self.regime_model = hmm.GaussianHMM(n_components=3, random_state=42) # Tree-based models with parameters optimized for 5-min timeframe self.models = { 'xgboost': xgb.XGBClassifier( colsample_bytree=0.85, learning_rate=0.05, max_depth=6, # Reduced depth for faster market dynamics min_child_weight=3, subsample=0.8, n_estimators=100, random_state=42, eval_metric='logloss', use_label_encoder=False ), 'lightgbm': lgb.LGBMClassifier( feature_fraction=0.9, learning_rate=0.02, max_depth=5, min_child_samples=20, # Smaller sample for 5-min data subsample=0.8, n_estimators=100, random_state=44, boosting_type='dart' # More robust to noise in high-frequency data ), 'catboost': CatBoostClassifier( depth=4, learning_rate=0.07, subsample=0.85, n_estimators=100, random_state=45, verbose=0 ), 'randomforest': RandomForestClassifier( n_estimators=100, max_depth=5, max_features='sqrt', min_samples_leaf=5, # Captures more granular patterns random_state=46 ), 'extratrees': ExtraTreesClassifier( n_estimators=100, max_depth=5, max_features='sqrt', min_samples_leaf=5, random_state=47 ) } # Primary meta-model self.meta_model = lgb.LGBMClassifier( n_estimators=100, max_depth=3, learning_rate=0.03, random_state=48, boosting_type='gbdt' ) # Secondary meta-model for consensus validation self.meta_model_backup = LogisticRegression( C=0.1, solver='liblinear', random_state=49 ) # Track feature importance self.feature_importances = {} self.feature_names = [] def detect_market_regime(self, data): """ Detects market regimes using HMM on returns and volatility Optimized for 5-minute data with shorter lookback windows Returns regime classifications (0=low vol, 1=normal, 2=high vol) """ # Calculate returns and volatility returns = np.log(data['close'] / data['close'].shift(1)) # Shorter window for 5-minute data (60 periods = 5 hours) volatility = returns.rolling(window=60).std() combined = pd.DataFrame({'returns': returns, 'volatility': volatility}).dropna() # Fit HMM model if we have enough data if len(combined) > 100: self.regime_model.fit(combined.values) regimes = self.regime_model.predict(combined.values) regime_series = pd.Series(index=data.index, dtype='float64') regime_series.iloc[len(data)-len(regimes):] = regimes return regime_series else: # Default to moderate regime if not enough data return pd.Series(1, index=data.index) def create_advanced_features(self, df): """ Create features optimized for 5-minute gold price prediction Features are organized in categories: 1. Market regime 2. Time-based features 3. Price action features 4. Volatility indicators 5. Momentum indicators 6. Volume indicators 7. Support/Resistance 8. Pattern recognition """ data = df.copy() # 1. Market regime detection data['market_regime'] = self.detect_market_regime(data) # 2. Time-based features for intraday seasonality data['hour'] = data.index.hour data['minute'] = data.index.minute data['day_of_week'] = data.index.dayofweek # Cyclical encoding of time (circular features) data['hour_sin'] = np.sin(2 * np.pi * data['hour']/24) data['hour_cos'] = np.cos(2 * np.pi * data['hour']/24) # 3. Short-term price action features # Moving averages adapted for 5-min timeframe for period in [12, 24, 48, 96, 144]: # 1h, 2h, 4h, 8h, 12h in 5-minute bars # Exponential moving averages data[f'ema_{period}'] = talib.EMA(data['close'], timeperiod=period) # Price relative to moving average (normalized distance) data[f'price_to_ema_{period}'] = data['close'] / data[f'ema_{period}'] - 1 # Trend strength data[f'trend_{period}'] = (data[f'ema_{period}'] - data[f'ema_{period}'].shift(period//4)) / data[f'ema_{period}'].shift(period//4) # 4. Volatility indicators # ATR with periods suitable for 5-minute bars for period in [12, 24, 48, 96]: # 1h, 2h, a data[f'atr_{period}'] = talib.ATR(data['high'], data['low'], data['close'], timeperiod=period) data[f'atr_ratio_{period}'] = data[f'atr_{period}'] / data['close'] # Bollinger Bands - essential for mean-reversion detection for period in [24, 48, 96]: upper, middle, lower = talib.BBANDS(data['close'], timeperiod=period, nbdevup=2, nbdevdn=2) data[f'bb_width_{period}'] = (upper - lower) / middle data[f'bb_position_{period}'] = (data['close'] - lower) / (upper - lower) # 5. Momentum indicators # RSI with different lookback periods for 5-min data for period in [12, 24, 48, 96]: data[f'rsi_{period}'] = talib.RSI(data['close'], timeperiod=period) # MACD for 5-minute data (faster parameters) macd, macd_signal, macd_hist = talib.MACD( data['close'], fastperiod=6, # Faster for 5-min data slowperiod=19, # Faster for 5-min data signalperiod=5 # Faster for 5-min data ) data['macd'] = macd data['macd_signal'] = macd_signal data['macd_hist'] = macd_hist # 6. Volume indicators (crucial for 5-minute signals) # Volume relative to moving average for period in [12, 24, 48]: data[f'volume_ma_{period}'] = talib.SMA(data['volume'], timeperiod=period) data[f'volume_ratio_{period}'] = data['volume'] / data[f'volume_ma_{period}'] # On-balance volume - good for measuring buying/selling pressure data['obv'] = talib.OBV(data['close'], data['volume']) data['obv_ma'] = talib.SMA(data['obv'], timeperiod=24) data['obv_ratio'] = data['obv'] / data['obv_ma'] # 7. Support/Resistance levels # Pivot points for 5-min (using 96 periods = 8 hours) data['pivot'] = (data['high'].rolling(96).max() + data['low'].rolling(96).min() + data['close'].rolling(96).mean()) / 3 data['dist_to_pivot'] = (data['close'] - data['pivot']) / data['close'] # 8. Candlestick pattern features # Candle size metrics data['candle_range'] = (data['high'] - data['low']) / data['close'] data['candle_body'] = abs(data['open'] - data['close']) / data['close'] # Rate of change - important for 5-min momentum for period in [6, 12, 24]: data[f'roc_{period}'] = talib.ROC(data['close'], timeperiod=period) # Target definition for 5-minute timeframe # Using appropriate thresholds for smaller price moves future_return = data['close'].shift(-self.forecast_bars) / data['close'] - 1 # Lower threshold for 5-minute bars (approximately 0.1-0.15% move) data['target'] = np.where(future_return > 0.0012, 1, np.where(future_return < -0.0012, 0, None)) # Drop rows with missing data return data.dropna() def prepare_features(self, data): """ Prepare and select optimal features for the model Features are grouped by category for easier selection """ # Most important features for 5-minute gold prediction feature_columns = [ # Market regime 'market_regime', # Time features for intraday patterns 'hour_sin', 'hour_cos', 'day_of_week', # Price action features 'price_to_ema_12', 'price_to_ema_24', 'price_to_ema_48', 'trend_24', 'trend_48', 'trend_96', # Volatility indicators 'atr_ratio_12', 'atr_ratio_24', 'bb_width_24', 'bb_width_48', 'bb_position_24', 'bb_position_48', # Momentum indicators 'rsi_12', 'rsi_24', 'rsi_48', 'macd', 'macd_hist', # Volume indicators 'volume_ratio_12', 'volume_ratio_24', 'obv_ratio', # Support/Resistance 'dist_to_pivot', # Pattern recognition 'candle_range', 'candle_body', 'roc_6', 'roc_12' ] X = data[feature_columns] if 'target' in data.columns: y = data['target'].astype(int) return X, y else: return X, None def fit(self, train_data): """Train the ensemble model on historical data""" print("Creating features...") processed_data = self.create_advanced_features(train_data) X, y = self.prepare_features(processed_data) # Store feature names for importance tracking self.feature_names = X.columns.tolist() # Scale features X_scaled = self.scaler.fit_transform(X) X_scaled = pd.DataFrame(X_scaled, columns=X.columns) # Time series cross-validation print("Performing time series cross-validation...") tscv = TimeSeriesSplit(n_splits=5) oof_preds = np.zeros((len(X_scaled), len(self.models))) for fold, (train_idx, val_idx) in enumerate(tscv.split(X_scaled)): X_train, X_val = X_scaled.iloc[train_idx], X_scaled.iloc[val_idx] y_train, y_val = y.iloc[train_idx], y.iloc[val_idx] for i, (name, model) in enumerate(self.models.items()): print(f"Training {name} for fold {fold+1}/5...") if name == 'xgboost': model.fit(X_train, y_train, eval_set=[(X_val, y_val)], verbose=False) elif name == 'lightgbm': model.fit(X_train, y_train, eval_set=[(X_val, y_val)]) elif name == 'catboost': model.fit(X_train, y_train, eval_set=(X_val, y_val)) else: model.fit(X_train, y_train) # Store out-of-fold predictions oof_preds[val_idx, i] = model.predict_proba(X_val)[:, 1] # Track feature importance if hasattr(model, 'feature_importances_'): if name not in self.feature_importances: self.feature_importances[name] = np.zeros(len(self.feature_names)) self.feature_importances[name] += model.feature_importances_ # Evaluate base models for i, name in enumerate(self.models.keys()): auc = roc_auc_score(y, oof_preds[:, i]) print(f"{name} Out-of-fold AUC: {auc:.4f}") # Train meta-models on out-of-fold predictions print("Training meta-models...") # Weight recent data more heavily for financial time series sample_weights = np.exp(np.linspace(-1, 0, len(y))) # Train primary meta-model self.meta_model.fit(oof_preds, y, sample_weight=sample_weights) # Train backup meta-model for consensus self.meta_model_backup.fit(oof_preds, y, sample_weight=sample_weights) # Display feature importance summary self._print_feature_importance() # Final training of all models on full dataset print("Final training on complete dataset...") for name, model in self.models.items(): if name in ['xgboost']: model.fit(X_scaled, y, eval_set=[(X_scaled, y)], verbose=False) elif name in ['lightgbm']: model.fit(X_scaled, y, eval_set=[(X_scaled, y)]) elif name in ['catboost']: model.fit(X_scaled, y, eval_set=(X_scaled, y)) else: model.fit(X_scaled, y) return processed_data def _print_feature_importance(self): """Display top features by importance for each model""" print("\n=== Feature Importance Analysis ===") for name, importances in self.feature_importances.items(): # Normalize importances to percentages importances = importances / np.sum(importances) * 100 # Sort by importance sorted_idx = np.argsort(importances)[::-1] print(f"\n{name.upper()} Top 10 Features:") for i in range(min(10, len(sorted_idx))): idx = sorted_idx[i] print(f" {self.feature_names[idx]}: {importances[idx]:.2f}%") def predict(self, data): """ Generate trading signals for 5-minute gold price data Returns: DataFrame with columns: - signal: Trading signal (-1=short, 0=neutral, 1=long) - strength: Signal strength (0-100%) - proba_up: Probability of price increase - proba_down: Probability of price decrease - model_agreement: Agreement level between meta-models - market_regime: Detected market regime """ processed_data = self.create_advanced_features(data) X, _ = self.prepare_features(processed_data) X_scaled = self.scaler.transform(X) X_scaled = pd.DataFrame(X_scaled, columns=X.columns) # Get predictions from base models base_preds = np.zeros((len(X_scaled), len(self.models))) for i, (name, model) in enumerate(self.models.items()): base_preds[:, i] = model.predict_proba(X_scaled)[:, 1] # Get predictions from both meta-models meta_proba = self.meta_model.predict_proba(base_preds) backup_proba = self.meta_model_backup.predict_proba(base_preds) # Calculate consensus level between meta-models (0-1) models_agreement = 1 - np.abs(meta_proba[:, 1] - backup_proba[:, 1]) # Initialize signals signals = pd.Series(0, index=processed_data.index) # Generate signals with sophisticated filtering # Long signal: high probability of price increase + high model agreement long_mask = (meta_proba[:, 1] > self.confidence_threshold) & (models_agreement > 0.8) # Short signal: high probability of price decrease + high model agreement short_mask = (meta_proba[:, 0] > self.confidence_threshold) & (models_agreement > 0.8) # Add market regime filter - only take signals in appropriate regimes market_regimes = processed_data['market_regime'] # Only generate long signals in trending or normal regimes (0 or 1) long_regime_mask = (market_regimes == 0) | (market_regimes == 1) # Only generate short signals in trending or high volatility regimes (0 or 2) short_regime_mask = (market_regimes == 0) | (market_regimes == 2) # Apply regime filters to signals signals.loc[long_mask & long_regime_mask] = 1 signals.loc[short_mask & short_regime_mask] = -1 # Calculate signal strength (0-100%) based on prediction confidence signal_strength = pd.Series(0.0, index=processed_data.index) signal_strength.loc[long_mask] = (meta_proba[long_mask, 1] - self.confidence_threshold) * (1 / (1 - self.confidence_threshold)) * 100 signal_strength.loc[short_mask] = (meta_proba[short_mask, 0] - self.confidence_threshold) * (1 / (1 - self.confidence_threshold)) * 100 # Risk management: additional signal filters # 1. Minimum signal duration (prevent rapid flipping) min_bars = 3 # Minimum 15 minutes for i in range(min_bars, len(signals)): if signals.iloc[i] != 0 and signals.iloc[i] == -signals.iloc[i-1]: # If signal flips too soon, maintain previous signal if sum(signals.iloc[i-min_bars:i] == signals.iloc[i-1]) >= min_bars-1: signals.iloc[i] = signals.iloc[i-1] # 2. Filter out signals during extreme volatility high_vol_mask = processed_data['atr_ratio_24'] > processed_data['atr_ratio_24'].quantile(0.95) signals.loc[high_vol_mask] = 0 # Combine results into a DataFrame results = pd.DataFrame({ 'signal': signals, 'strength': signal_strength, 'proba_up': meta_proba[:, 1], 'proba_down': meta_proba[:, 0], 'model_agreement': models_agreement, 'market_regime': processed_data['market_regime'] }, index=processed_data.index) return results def evaluate_performance(self, test_data): """ Evaluate model performance with trading simulation - corrected version """ results = self.predict(test_data) # Calculate forward returns for evaluation period close_prices = test_data['close'] forward_returns = close_prices.shift(-self.forecast_bars) / close_prices - 1 # Apply signals to returns (long = 1x return, short = -1x return) strategy_returns = results['signal'] * forward_returns # Properly handle NaN values that might appear from shifts strategy_returns = strategy_returns.dropna() # Calculate performance metrics total_return = strategy_returns.sum() # Correct annualization factor: 252 trading days, 12 hours per day, 12 bars per hour annualization_factor = np.sqrt(252 * 12 * 12) sharpe_ratio = strategy_returns.mean() / strategy_returns.std() * annualization_factor # Correct win rate calculation - account for signal direction wins = ((strategy_returns > 0) & (results['signal'] != 0)).sum() total_trades = (results['signal'] != 0).sum() win_rate = wins / total_trades if total_trades > 0 else 0 # Correct drawdown calculation cumulative_returns = strategy_returns.cumsum() drawdowns = cumulative_returns - cumulative_returns.cummax() max_drawdown = drawdowns.min() # Signal statistics signal_count = (results['signal'] != 0).sum() signal_changes = results['signal'].diff().abs() signal_changes = signal_changes[signal_changes > 0].sum() / 2 # Each change counts twice in diff # Correct trading days calculation (typically 5 days a week for forex) # Assuming 12 hours of active trading per day and 12 5-min bars per hour trading_days = len(results) / (12 * 12) avg_signals_per_day = signal_count / trading_days # Calculate profit factor profitable_trades = strategy_returns[strategy_returns > 0].sum() losing_trades = abs(strategy_returns[strategy_returns < 0].sum()) profit_factor = profitable_trades / losing_trades if losing_trades != 0 else float('inf') # Calculate average profit per trade avg_profit_per_trade = total_return / total_trades if total_trades > 0 else 0 # Compile metrics metrics = { 'total_return': total_return, 'annualized_return': total_return * (252 / trading_days), 'sharpe_ratio': sharpe_ratio, 'win_rate': win_rate, 'max_drawdown': max_drawdown, 'profit_factor': profit_factor, 'signal_count': signal_count, 'signal_changes': signal_changes, 'avg_signals_per_day': avg_signals_per_day, 'avg_profit_per_trade': avg_profit_per_trade } print("\n=== Performance Evaluation ===") print(f"Total Return: {total_return:.2%}") print(f"Annualized Return: {metrics['annualized_return']:.2%}") print(f"Annualized Sharpe Ratio: {sharpe_ratio:.2f}") print(f"Win Rate: {win_rate:.2%}") print(f"Maximum Drawdown: {max_drawdown:.2%}") print(f"Profit Factor: {profit_factor:.2f}") print(f"Total Signals: {signal_count}") print(f"Signal Changes: {signal_changes}") print(f"Average Signals Per Day: {avg_signals_per_day:.2f}") print(f"Average Profit Per Trade: {avg_profit_per_trade:.4%}") return metrics, results, strategy_returns # Example usage def run_model(data_path, forecast_bars=24, confidence_threshold=0.65): """ Run the model on input data Parameters: data_path: Path to CSV file with OHLCV data forecast_bars: Number of 5-min bars to forecast confidence_threshold: Threshold for signal generation Returns: predictor: Trained model metrics: Performance metrics results: Signal results returns: Strategy returns """ # Load and prepare data data = pd.read_csv(data_path) data['timestamp'] = pd.to_datetime(data['timestamp']) data = data.drop_duplicates(subset=['timestamp']) data.set_index('timestamp', inplace=True) # Split into train/test #pick the last 40% of data before the last 80% from 40% to 80% train_size = int(len(data) * 0.95) train_data = data[int(len(data) * 0.80):train_size].copy() test_data = data[train_size:].copy() # Create and train model predictor = TreeEnsemblePredictor( forecast_bars=forecast_bars, confidence_threshold=confidence_threshold ) print(f"Training model with forecast_bars={forecast_bars}, confidence_threshold={confidence_threshold}") predictor.fit(train_data) # Evaluate model metrics, results, returns = predictor.evaluate_performance(test_data) return predictor, metrics, results, returns