""" ML Ensemble for Signal Prediction SVM, Random Forest, Gradient Boosting, Logistic Regression with PCA and KMeans """ import numpy as np import pandas as pd from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier from sklearn.svm import SVC from sklearn.linear_model import LogisticRegression from sklearn.decomposition import PCA from sklearn.cluster import KMeans from sklearn.preprocessing import StandardScaler from sklearn.model_selection import cross_val_score from typing import Dict, List, Tuple, Optional import pickle import hashlib from datetime import datetime class FeatureEngineer: """Build technical features from OHLCV data""" def __init__(self, volatility_window: int = 20, rsi_period: int = 14): self.volatility_window = volatility_window self.rsi_period = rsi_period self.scaler = StandardScaler() def build_features(self, df: pd.DataFrame, mode: str = 'training') -> pd.DataFrame: """ Build features from candle data mode: 'training' (requires fwd_ret) or 'inference' (no fwd_ret needed) """ df = df.copy() df = df.sort_values('timestamp') # Returns over multiple windows for window in [1, 2, 4, 8, 24]: df[f'ret_{window}h'] = df['close'].pct_change(window) # Volatility df['volatility'] = df['close'].pct_change().rolling(self.volatility_window).std() # RSI df['rsi'] = self._calculate_rsi(df['close'], self.rsi_period) # MACD df['macd'], df['macd_signal'] = self._calculate_macd(df['close']) # Volume z-score df['volume_zscore'] = (df['volume'] - df['volume'].rolling(24).mean()) / df['volume'].rolling(24).std() # Position in 24h range df['range_position'] = (df['close'] - df['low'].rolling(24).min()) / (df['high'].rolling(24).max() - df['low'].rolling(24).min()) # Skew df['price_skew'] = df['close'].rolling(24).skew() # Cross-sectional features (if multiple symbols) if 'symbol' in df.columns: df['ret_rank'] = df.groupby('timestamp')['ret_1h'].rank(pct=True) df['volume_rank'] = df.groupby('timestamp')['volume'].rank(pct=True) # Forward return (only for training mode) if mode == 'training': df['fwd_ret'] = df['close'].shift(-1) / df['close'] - 1 df['target'] = (df['fwd_ret'] > 0).astype(int) # Drop NaN if mode == 'training': df = df.dropna(subset=['fwd_ret', 'target']) else: # For inference, keep the last row even if it has NaN in some features df = df.dropna(subset=['ret_1h', 'volatility', 'rsi']) return df def _calculate_rsi(self, prices: pd.Series, period: int) -> pd.Series: delta = prices.diff() gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() rs = gain / loss return 100 - (100 / (1 + rs)) def _calculate_macd(self, prices: pd.Series) -> Tuple[pd.Series, pd.Series]: ema_12 = prices.ewm(span=12).mean() ema_26 = prices.ewm(span=26).mean() macd = ema_12 - ema_26 signal = macd.ewm(span=9).mean() return macd, signal def get_feature_columns(self) -> List[str]: return [ 'ret_1h', 'ret_2h', 'ret_4h', 'ret_8h', 'ret_24h', 'volatility', 'rsi', 'macd', 'macd_signal', 'volume_zscore', 'range_position', 'price_skew' ] class MLEnsemble: """Ensemble of ML models with PCA and KMeans""" def __init__(self, n_components: int = 10, n_clusters: int = 5): self.n_components = n_components self.n_clusters = n_clusters # Models self.svm = SVC(probability=True, random_state=42) self.rf = RandomForestClassifier(n_estimators=100, random_state=42) self.gbm = GradientBoostingClassifier(n_estimators=100, random_state=42) self.logreg = LogisticRegression(random_state=42) # Dimensionality reduction self.pca = PCA(n_components=n_components) # Regime labeling self.kmeans = KMeans(n_clusters=n_clusters, random_state=42) # Ensemble weights (learned by genetic algorithm) self.weights = { 'svm': 0.25, 'rf': 0.25, 'gbm': 0.25, 'logreg': 0.25 } self.feature_columns = None self.fitted = False def fit(self, X: np.ndarray, y: np.ndarray) -> Dict: """Fit all models and learn ensemble weights""" # Fit PCA X_pca = self.pca.fit_transform(X) # Fit KMeans for regime labeling regimes = self.kmeans.fit_predict(X_pca) # Fit individual models self.svm.fit(X_pca, y) self.rf.fit(X_pca, y) self.gbm.fit(X_pca, y) self.logreg.fit(X_pca, y) # Get cross-validation scores cv_scores = { 'svm': cross_val_score(self.svm, X_pca, y, cv=5).mean(), 'rf': cross_val_score(self.rf, X_pca, y, cv=5).mean(), 'gbm': cross_val_score(self.gbm, X_pca, y, cv=5).mean(), 'logreg': cross_val_score(self.logreg, X_pca, y, cv=5).mean() } # Simple weight optimization based on CV scores total_score = sum(cv_scores.values()) for model in self.weights: self.weights[model] = cv_scores[model] / total_score self.fitted = True return { 'cv_scores': cv_scores, 'weights': self.weights, 'regime_distribution': np.bincount(regimes).tolist() } def predict_proba(self, X: np.ndarray) -> np.ndarray: """Predict probability of up movement""" if not self.fitted: raise ValueError("Model not fitted") X_pca = self.pca.transform(X) # Get individual predictions proba_svm = self.svm.predict_proba(X_pca)[:, 1] proba_rf = self.rf.predict_proba(X_pca)[:, 1] proba_gbm = self.gbm.predict_proba(X_pca)[:, 1] proba_logreg = self.logreg.predict_proba(X_pca)[:, 1] # Weighted ensemble ensemble_proba = ( self.weights['svm'] * proba_svm + self.weights['rf'] * proba_rf + self.weights['gbm'] * proba_gbm + self.weights['logreg'] * proba_logreg ) return ensemble_proba def predict(self, X: np.ndarray, threshold: float = 0.5) -> np.ndarray: """Predict direction with confidence-based sizing""" proba = self.predict_proba(X) # LinUCB-style contextual bandit for position sizing confidence = np.abs(proba - 0.5) * 2 # 0 to 1 # Direction and position sizing direction = np.where(proba > threshold, 1, -1) position = direction * confidence return { 'direction': direction, 'probability_up': proba, 'confidence': confidence, 'position': position } def get_regime(self, X: np.ndarray) -> np.ndarray: """Get market regime for each sample""" X_pca = self.pca.transform(X) return self.kmeans.predict(X_pca) def save(self, filepath: str): """Save model to file""" model_data = { 'svm': self.svm, 'rf': self.rf, 'gbm': self.gbm, 'logreg': self.logreg, 'pca': self.pca, 'kmeans': self.kmeans, 'weights': self.weights, 'feature_columns': self.feature_columns, 'fitted': self.fitted, 'n_components': self.n_components, 'n_clusters': self.n_clusters } with open(filepath, 'wb') as f: pickle.dump(model_data, f) def load(self, filepath: str): """Load model from file""" with open(filepath, 'rb') as f: model_data = pickle.load(f) self.svm = model_data['svm'] self.rf = model_data['rf'] self.gbm = model_data['gbm'] self.logreg = model_data['logreg'] self.pca = model_data['pca'] self.kmeans = model_data['kmeans'] self.weights = model_data['weights'] self.feature_columns = model_data['feature_columns'] self.fitted = model_data['fitted'] self.n_components = model_data['n_components'] self.n_clusters = model_data['n_clusters'] def compute_feature_hash(self, features: np.ndarray) -> str: """Compute hash of feature vector for reproducibility""" feature_bytes = features.tobytes() return hashlib.sha256(feature_bytes).hexdigest()