# AI Model Ensemble

> "Provides Combine multiple models for improved prediction accuracy and robustness"

- Skill: `paulpas/ai-model-ensemble` (Agent Skill)
- Install (CLI): `npx skillmds@latest add paulpas/ai-model-ensemble`
- Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/ai-model-ensemble/raw
- Safety review: PASS (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- License: MIT
- Author: paulpas (https://skillmd.com/u/paulpas)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/paulpas/ai-model-ensemble

---





**Role:** Design and implement ensemble methods that leverage multiple models for trading signals

**Philosophy:** Diversification reduces risk and improves stability. Prioritize models that are diverse in structure, training data, and error patterns to maximize ensemble benefits.

## Key Principles

1. **Diversity Over Individual Quality**: Prefer diverse weak models over a few strong ones
2. **Hierarchical Ensembles**: Use stacking for non-linear combination of base models
3. **Robust Aggregation**: Use median or trimmed mean to resist outlier predictions
4. **Dynamic Weighting**: Adjust model weights based on recent performance
5. **Failure Detection**: Monitor when ensemble underperforms individual models

## Implementation Guidelines

### Structure
- Core logic: `ensemble/models.py` - Ensemble classes
- Aggregators: `ensemble/aggregators.py` - Prediction combination methods
- Model pool: `ensemble/pool.py` - Model management
- Config: `config/ensemble_config.yaml` - Ensemble parameters

### Patterns to Follow
- Use different model families (tree, neural, linear)
- Train on different features or time periods
- Implement online learning for weight updates
- Include base model performance tracking

## Adherence Checklist
Before completing your task, verify:
- [ ] Base models use different algorithms and architectures
- [ ] Diversity measured and maintained across training
- [ ] Aggregation method appropriate for trading (median, weighted)
- [ ] Dynamic weighting based on recent performance
- [ ] Ensemble outperforms best individual model in backtest



## Code Examples

### Basic Ensemble Aggregation

```python
import numpy as np
from typing import List, Dict, Callable, Tuple
from abc import ABC, abstractmethod

class EnsembleAggregator(ABC):
    """Base class for ensemble aggregation methods."""
    
    @abstractmethod
    def aggregate(self, predictions: List[np.ndarray], weights: List[float] = None) -> np.ndarray:
        """Aggregate predictions from multiple models."""
        pass

class SimpleAverageAggregator(EnsembleAggregator):
    """Simple average of all model predictions."""
    
    def aggregate(self, predictions: List[np.ndarray], weights: List[float] = None) -> np.ndarray:
        """Average predictions equally."""
        if not predictions:
            return np.array([])
        
        predictions = np.array(predictions)
        return np.mean(predictions, axis=0)

class WeightedAverageAggregator(EnsembleAggregator):
    """Weighted average of predictions."""
    
    def __init__(self, normalize: bool = True):
        self.normalize = normalize
    
    def aggregate(self, predictions: List[np.ndarray], weights: List[float] = None) -> np.ndarray:
        """Weighted average of predictions."""
        if not predictions:
            return np.array([])
        
        predictions = np.array(predictions)
        
        if weights is None:
            weights = np.ones(len(predictions))
        
        if self.normalize:
            weights = np.array(weights) / np.sum(weights)
        
        return np.average(predictions, axis=0, weights=weights)

class MedianAggregator(EnsembleAggregator):
    """Median of predictions (robust to outliers)."""
    
    def aggregate(self, predictions: List[np.ndarray], weights: List[float] = None) -> np.ndarray:
        """Median prediction."""
        if not predictions:
            return np.array([])
        
        predictions = np.array(predictions)
        return np.median(predictions, axis=0)

class TrimmedMeanAggregator(EnsembleAggregator):
    """Trimmed mean (removes outliers before averaging)."""
    
    def __init__(self, trim_fraction: float = 0.1):
        self.trim_fraction = trim_fraction
    
    def aggregate(self, predictions: List[np.ndarray], weights: List[float] = None) -> np.ndarray:
        """Trimmed mean of predictions."""
        if not predictions:
            return np.array([])
        
        predictions = np.array(predictions)
        
        n = len(predictions)
        trim_count = int(n * self.trim_fraction)
        
        if trim_count == 0:
            return np.mean(predictions, axis=0)
        
        sorted_preds = np.sort(predictions, axis=0)
        trimmed = sorted_preds[trim_count:n-trim_count]
        
        return np.mean(trimmed, axis=0)
```

### Random Forest Ensemble

```python
import numpy as np
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.tree import DecisionTreeRegressor
from sklearn.linear_model import Ridge
from sklearn.preprocessing import StandardScaler
from typing import List, Tuple, Dict

class TreeEnsemble:
    """Ensemble of tree-based models."""
    
    def __init__(self, n_estimators: int = 10):
        self.n_estimators = n_estimators
        
        # Different model configurations for diversity
        self.models = [
            RandomForestRegressor(
                n_estimators=n_estimators // 3,
                max_depth=5,
                random_state=42,
                n_jobs=-1
            ),
            RandomForestRegressor(
                n_estimators=n_estimators // 3,
                max_depth=10,
                random_state=123,
                n_jobs=-1
            ),
            GradientBoostingRegressor(
                n_estimators=n_estimators // 3,
                max_depth=4,
                random_state=456,
                learning_rate=0.1
            )
        ]
        
        self.scaler = StandardScaler()
    
    def fit(self, X: np.ndarray, y: np.ndarray):
        """Fit all models in ensemble."""
        X_scaled = self.scaler.fit_transform(X)
        
        for model in self.models:
            model.fit(X_scaled, y)
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Get predictions from all models."""
        X_scaled = self.scaler.transform(X)
        
        predictions = []
        for model in self.models:
            pred = model.predict(X_scaled)
            predictions.append(pred)
        
        return predictions
    
    def ensemble_predict(self, X: np.ndarray, aggregator=None) -> np.ndarray:
        """Get ensemble prediction using specified aggregator."""
        predictions = self.predict(X)
        
        if aggregator is None:
            # Use median for robustness
            return np.median(predictions, axis=0)
        
        return aggregator.aggregate(predictions)
```

### Stacking Ensemble

```python
import numpy as np
from sklearn.linear_model import Ridge
from sklearn.preprocessing import StandardScaler
from typing import List, Tuple, Dict

class StackingEnsemble:
    """Stacking ensemble with meta-learner."""
    
    def __init__(self, base_models: List[object], meta_model=None):
        self.base_models = base_models
        self.meta_model = meta_model or Ridge(alpha=1.0)
        self.scaler = StandardScaler()
        self.is_fitted = False
    
    def _get_oof_predictions(self, X: np.ndarray, y: np.ndarray, n_folds: int = 5) -> np.ndarray:
        """Generate out-of-fold predictions for stacking."""
        n_samples = X.shape[0]
        n_models = len(self.base_models)
        oof_predictions = np.zeros((n_samples, n_models))
        
        fold_size = n_samples // n_folds
        
        for i, model in enumerate(self.base_models):
            for fold in range(n_folds):
                start_idx = fold * fold_size
                end_idx = start_idx + fold_size if fold < n_folds - 1 else n_samples
                
                # Train on all but this fold
                train_idx = np.concatenate([
                    np.arange(0, start_idx),
                    np.arange(end_idx, n_samples)
                ])
                
                model.fit(X[train_idx], y[train_idx])
                oof_predictions[end_idx - fold_size:end_idx, i] = model.predict(X[end_idx - fold_size:end_idx])
        
        return oof_predictions
    
    def fit(self, X: np.ndarray, y: np.ndarray):
        """Fit base models and meta-learner."""
        # Fit base models on full data
        for model in self.base_models:
            model.fit(X, y)
        
        # Generate OOF predictions for meta-learner
        oof_predictions = self._get_oof_predictions(X, y)
        oof_scaled = self.scaler.fit_transform(oof_predictions)
        
        # Fit meta-learner
        self.meta_model.fit(oof_scaled, y)
        self.is_fitted = True
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Get final prediction using ensemble."""
        # Get base model predictions
        base_predictions = []
        for model in self.base_models:
            pred = model.predict(X)
            base_predictions.append(pred)
        
        base_predictions = np.array(base_predictions).T
        
        # Apply meta-learner
        base_scaled = self.scaler.transform(base_predictions)
        meta_prediction = self.meta_model.predict(base_scaled)
        
        return meta_prediction
```

### Dynamic Weight Ensemble

```python
import numpy as np
from typing import List, Dict

class DynamicWeightEnsemble:
    """Ensemble with weights that adapt based on recent performance."""
    
    def __init__(self, models: List[object], learning_rate: float = 0.1):
        self.models = models
        self.n_models = len(models)
        self.learning_rate = learning_rate
        
        # Initialize equal weights
        self.weights = np.ones(self.n_models) / self.n_models
        
        # Track performance
        self.performance_history = []
        self.window_size = 20
    
    def fit(self, X: np.ndarray, y: np.ndarray):
        """Fit all base models."""
        for model in self.models:
            model.fit(X, y)
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Get weighted ensemble prediction."""
        predictions = []
        for model in self.models:
            pred = model.predict(X)
            predictions.append(pred)
        
        predictions = np.array(predictions)
        
        # Weighted average
        weights = self.weights / np.sum(self.weights)
        return np.average(predictions, axis=0, weights=weights)
    
    def update_weights(self, X: np.ndarray, y: np.ndarray, 
                       metric='mae', decay: float = 0.95):
        """Update weights based on recent performance."""
        predictions = []
        for model in self.models:
            pred = model.predict(X)
            predictions.append(pred)
        
        predictions = np.array(predictions)
        
        # Calculate individual model performance
        weights = self.weights / np.sum(self.weights)
        ensemble_pred = np.average(predictions, axis=0, weights=weights)
        
        model_errors = []
        for i, pred in enumerate(predictions):
            if metric == 'mae':
                error = np.mean(np.abs(pred - y))
            elif metric == 'mse':
                error = np.mean((pred - y) ** 2)
            else:
                error = np.mean(np.abs(pred - y))
            model_errors.append(error)
        
        # Convert to weights (lower error = higher weight)
        exp_errors = np.exp(-np.array(model_errors) / 0.1)  # Temperature scaling
        new_weights = exp_errors / np.sum(exp_errors)
        
        # Smooth weight updates
        self.weights = decay * self.weights + (1 - decay) * new_weights
        
        # Track performance
        self.performance_history.append({
            'predictions': predictions,
            'ensemble_prediction': ensemble_pred,
            'model_errors': model_errors,
            'weights': self.weights.copy()
        })
        
        return self.weights
    
    def get_weights_history(self) -> List[np.ndarray]:
        """Get weight evolution over time."""
        return [h['weights'] for h in self.performance_history]
```

### Failure Detection in Ensemble

```python
import numpy as np
from typing import List, Dict

class EnsembleFailureDetector:
    """Detect when ensemble underperforms individual models."""
    
    def __init__(self, threshold: float = 1.2):
        self.threshold = threshold  # 20% worse than best model
    
    def check_ensemble_quality(self, ensemble_prediction: np.ndarray,
                              individual_predictions: List[np.ndarray],
                              actual: np.ndarray) -> Dict:
        """Check if ensemble performs adequately."""
        # Calculate individual errors
        individual_errors = []
        for pred in individual_predictions:
            error = np.mean(np.abs(pred - actual))
            individual_errors.append(error)
        
        # Calculate ensemble error
        ensemble_error = np.mean(np.abs(ensemble_prediction - actual))
        
        # Find best individual error
        best_individual_error = min(individual_errors)
        
        # Check if ensemble is significantly worse
        is_degraded = ensemble_error > self.threshold * best_individual_error
        
        return {
            'ensemble_error': ensemble_error,
            'best_individual_error': best_individual_error,
            'ratio': ensemble_error / (best_individual_error + 1e-8),
            'is_degraded': is_degraded,
            'individual_errors': individual_errors,
            'recommendation': 'retrain' if is_degraded else 'continue'
        }
    
    def check_diversity(self, individual_predictions: List[np.ndarray]) -> Dict:
        """Check if predictions are diverse enough."""
        if len(individual_predictions) < 2:
            return {'diversity_score': 1.0, 'is_diverse': True}
        
        predictions = np.array(individual_predictions)
        
        # Pairwise correlation
        correlations = []
        for i in range(len(predictions)):
            for j in range(i+1, len(predictions)):
                corr = np.corrcoef(predictions[i], predictions[j])[0, 1]
                correlations.append(corr)
        
        avg_correlation = np.mean(correlations)
        diversity_score = 1 - abs(avg_correlation)
        
        return {
            'diversity_score': diversity_score,
            'is_diverse': diversity_score > 0.3,
            'avg_correlation': avg_correlation,
            'correlation_distribution': correlations
        }
```

---

## Constraints

### MUST DO
- Validate input feature distributions against training data baselines; flag drift exceeding 2 standard deviations
- Implement model versioning with reproducibility tags — every prediction must be traceable to the exact model artifact and config
- Include confidence intervals or probability estimates alongside all point predictions, never return raw scores without context
- Log all model inputs, outputs, and metadata to enable post-hoc analysis of prediction failures
- Implement feature computation consistently between training and inference — use the same transformation pipeline for both

### MUST NOT DO
- Do not train models on look-ahead biased features (e.g., using future prices or events in training data)
- Avoid deploying a new model version without shadow-testing against the current production model first
- Never retrain a model on a data window that includes regime changes without explicit regime-aware validation
- Do not use accuracy as the primary metric for imbalanced datasets — use precision/recall, F1, or AUC-ROC
- Avoid hardcoding feature names; load them from a schema or config file to prevent mismatches between training and inference


## Live References

> Authoritative documentation links for this domain. The model follows markdown links at load time to resolve external references and inline content.

- [Ensemble Learning: A Survey (Dietterich, 2000)](https://link.springer.com/article/10.1023/A:1010933404324) — Thomas Dietterich's seminal paper on ensemble learning methods for improving prediction accuracy
- [scikit-learn Ensemble Methods](https://scikit-learn.org/stable/modules/ensemble.html) — Official scikit-learn documentation on Bagging, Random Forests, Gradient Boosting, and AdaBoost
- [Stacking and Blending Ensemble Techniques (Kaggle Guide)](https://www.kaggle.com/code/lextoumbourou/stacking-demo/notebook) — Kaggle's guide to stacking and blending ensemble methods for competitive machine learning
- [Quantitative Trading with ML (Ernest Chan)](https://epchan.blogspot.com/) — Ernest Chan's blog on applying machine learning ensemble methods to quantitative trading strategies
- [XGBoost Documentation](https://xgboost.readthedocs.io/) — Official XGBoost documentation for gradient-boosted decision trees, a dominant model in financial prediction