← Back to Home
Collective Intelligence An Ensemble Strategy with Fixed Weights and Rolling Backtest Analysis

Collective Intelligence An Ensemble Strategy with Fixed Weights and Rolling Backtest Analysis

In quantitative trading, relying on a single indicator can be fragile. Markets are complex, and what works in one environment might fail in another. Ensemble methods address this by combining multiple independent trading signals, aiming to leverage their collective intelligence and improve overall robustness. This article explores an Ensemble Strategy with Fixed Weights, where signals from distinct sub-models are aggregated, and evaluates its performance using a rolling backtest to assess consistency over time.

The Method & Theory: The Power of Weighted Consensus

The core idea behind this ensemble strategy is to derive a final trading decision from a weighted vote of several independent sub-models. Each sub-model provides a directional signal (long, short, or neutral), and these signals are then multiplied by pre-defined, fixed weights. The sum of these weighted votes forms a “consensus signal.”

The benefits of this approach include:

The strategy incorporates three widely used sub-models:

  1. Simple Moving Average (SMA) Crossover: This classic trend-following indicator generates a long signal when a fast SMA crosses above a slow SMA, and a short signal when the fast SMA crosses below the slow SMA. It captures sustained price direction.
  2. Relative Strength Index (RSI) Trend: The RSI measures the speed and change of price movements. In this context, it’s used as a trend filter: an RSI above 50 suggests an upward bias, while below 50 suggests a downward bias.
  3. Moving Average Convergence Divergence (MACD) Crossover: MACD identifies changes in the strength, direction, momentum, and duration of a trend. A buy signal typically occurs when the MACD line crosses above the signal line, and a sell signal when it crosses below.

Each sub-model generates a signal: +1 for a bullish indication, -1 for a bearish indication, and 0 for neutral or no clear signal. These signals are then combined:

\text{WeightedVoteSum} = (S_{SMA} \times W_{SMA}) + (S_{RSI} \times W_{RSI}) + (S_{MACD} \times W_{MACD})

where S represents the signal from each sub-model and W represents its fixed weight. A consensus_threshold determines if this WeightedVoteSum is strong enough to trigger a trade. For example, if consensus_threshold is 0.1, a sum of 0.2 would be a buy, -0.15 a sell, and 0.05 would be neutral.

The Strategy: EnsembleStrategyWithFixedWeights

The Backtrader implementation provides a clear structure for this ensemble approach:

import backtrader as bt
import backtrader.indicators as btind
import yfinance as yf
import datetime
import math
import numpy as np

class EnsembleStrategyWithFixedWeights(bt.Strategy):
    params = (
        ('sma_fast_p', 10), ('sma_slow_p', 30), # SMA periods
        ('rsi_p', 14), ('rsi_level', 50),       # RSI period and level
        ('macd_fast', 12), ('macd_slow', 26), ('macd_signal', 9), # MACD periods
        ('w_sma', 0.4), ('w_rsi', 0.3), ('w_macd', 0.3), # Fixed weights for each sub-model
        ('consensus_threshold', 0.1), # Minimum weighted vote sum to trigger action
        ('stop_loss_perc', 0.04),     # Stop loss percentage
        ('printlog', True),
    )

    def __init__(self):
        self.dataclose = self.datas[0].close
        # Initialize each sub-model indicator
        self.sma_fast = btind.SimpleMovingAverage(self.dataclose, period=self.p.sma_fast_p)
        self.sma_slow = btind.SimpleMovingAverage(self.dataclose, period=self.p.sma_slow_p)
        self.rsi = btind.RSI(self.dataclose, period=self.p.rsi_p)
        self.macd = btind.MACD(self.dataclose, period_me1=self.p.macd_fast, period_me2=self.p.macd_slow, period_signal=self.p.macd_signal)
        self.macd_line = self.macd.macd
        self.macd_signal_line = self.macd.signal

        # Store and normalize weights
        self.weights = {'sma': max(0, self.p.w_sma), 'rsi': max(0, self.p.w_rsi), 'macd': max(0, self.p.w_macd)}
        total_weight = sum(self.weights.values())
        if total_weight > 0 and not math.isclose(total_weight, 1.0):
            self.weights = {k: v / total_weight for k, v in self.weights.items()}
            self.log(f"Normalized Weights: {self.weights}")

        self.order = None
        self.stop_order = None

    # Helper functions to get signals from each sub-model
    def _get_sma_signal(self):
        return 1 if self.sma_fast[0] > self.sma_slow[0] else (-1 if self.sma_fast[0] < self.sma_slow[0] else 0)
    def _get_rsi_signal(self):
        return 1 if self.rsi[0] > self.p.rsi_level else (-1 if self.rsi[0] < self.p.rsi_level else 0)
    def _get_macd_signal(self):
        if not math.isnan(self.macd_line[0]) and not math.isnan(self.macd_signal_line[0]):
            return 1 if self.macd_line[0] > self.macd_signal_line[0] else (-1 if self.macd_line[0] < self.macd_signal_line[0] else 0)
        return 0

    def notify_order(self, order):
        # Manages order lifecycle, including placing and cancelling stop-loss orders
        # This function is crucial for robust risk management in live trading.
        if order.status == order.Completed:
            if order.isbuy() and self.position.size > 0: # Entry completed
                stop_price = order.executed.price * (1.0 - self.p.stop_loss_perc)
                self.stop_order = self.sell(exectype=bt.Order.Stop, price=stop_price, size=order.executed.size)
            elif order.issell() and self.position.size < 0: # Entry completed (short)
                stop_price = order.executed.price * (1.0 + self.p.stop_loss_perc)
                self.stop_order = self.buy(exectype=bt.Order.Stop, price=stop_price, size=order.executed.size)
            elif not self.position and self.stop_order: # Close completed
                self.cancel(self.stop_order)
                self.stop_order = None
        # ... (other status handling like Canceled, Rejected, Margin, Expired)
        self.order = None # Reset order state after processing

    def next(self):
        if self.order: return # Only one order at a time

        # Get signals from all sub-models
        sma_signal = self._get_sma_signal()
        rsi_signal = self._get_rsi_signal()
        macd_signal = self._get_macd_signal()

        # Calculate the weighted consensus signal
        weighted_vote_sum = (
            sma_signal * self.weights['sma'] +
            rsi_signal * self.weights['rsi'] +
            macd_signal * self.weights['macd']
        )
        
        final_signal = 0
        if weighted_vote_sum > self.p.consensus_threshold:
            final_signal = 1 # Strong enough for long
        elif weighted_vote_sum < -self.p.consensus_threshold:
            final_signal = -1 # Strong enough for short

        # Trading Logic: Enter if no position, Exit if signal flips
        if not self.position:
            if final_signal == 1:
                self.log(f'CONSENSUS LONG SIGNAL (Vote: {weighted_vote_sum:.2f}). Placing BUY Order.')
                if self.stop_order: self.cancel(self.stop_order) # Cancel old stop if any
                self.order = self.buy()
            elif final_signal == -1:
                self.log(f'CONSENSUS SHORT SIGNAL (Vote: {weighted_vote_sum:.2f}). Placing SELL Order.')
                if self.stop_order: self.cancel(self.stop_order) # Cancel old stop if any
                self.order = self.sell()
        else: # Already in a position
            current_position_signal = 1 if self.position.size > 0 else -1
            if final_signal != current_position_signal: # Exit if signal flips or goes neutral
                self.log(f'CONSENSUS EXIT SIGNAL (Vote: {weighted_vote_sum:.2f}, Final Signal: {final_signal}). Closing Position.')
                if self.stop_order: self.cancel(self.stop_order) # Cancel active stop
                self.order = self.close() # Close position

Rolling Backtest: Assessing Performance Consistency

To properly evaluate any trading strategy, especially one with fixed parameters like the weights here, a simple historical backtest might mislead due to overfitting. A rolling backtest provides a more realistic assessment of a strategy’s expected performance over time. It segments the historical data into consecutive, fixed-length windows (e.g., 3 months, 12 months), and the strategy is run independently on each window.

This approach helps answer crucial questions:

import pandas as pd
import dateutil.relativedelta as rd
import matplotlib.pyplot as plt
import seaborn as sns

def run_rolling_backtest(
    ticker="BTC-USD",
    start="2018-01-01",
    end="2025-12-31",
    window_months=3, # e.g., 3-month rolling windows
    strategy_params=None
):
    strategy_params = strategy_params or {}
    all_results = []
    start_dt = pd.to_datetime(start)
    end_dt = pd.to_datetime(end)
    current_start = start_dt

    while True:
        current_end = current_start + rd.relativedelta(months=window_months)
        if current_end > end_dt: break

        print(f"\nROLLING BACKTEST: {current_start.date()} to {current_end.date()}")
        # Data download adheres to saved instruction: auto_adjust=False and droplevel(1, axis=1)
        data = yf.download(ticker, start=current_start, end=current_end, progress=False, auto_adjust=False).droplevel(1, axis=1)
        if data.empty or len(data) < 90: # Ensure sufficient data for indicators
            print("Not enough data.")
            current_start += rd.relativedelta(months=window_months)
            continue

        cerebro = bt.Cerebro()
        cerebro.addstrategy(EnsembleStrategyWithFixedWeights, **strategy_params)
        cerebro.adddata(bt.feeds.PandasData(dataname=data))
        cerebro.broker.setcash(100000)
        cerebro.broker.setcommission(commission=0.001)
        cerebro.addsizer(bt.sizers.PercentSizer, percents=95)

        start_val = cerebro.broker.getvalue()
        cerebro.run() # Run the backtest for this window
        final_val = cerebro.broker.getvalue()
        ret = (final_val - start_val) / start_val * 100

        all_results.append({'start': current_start.date(), 'end': current_end.date(),
                            'return_pct': ret, 'final_value': final_val})
        print(f"Return: {ret:.2f}% | Final Value: {final_val:.2f}")
        current_start += rd.relativedelta(months=window_months)

    return pd.DataFrame(all_results)

def report_stats(df):
    # Calculates and prints mean, median, std dev, min/max returns, and Sharpe Ratio
    # ... (function body)

def plot_four_charts(df, rolling_sharpe_window=4):
    # Visualizes rolling backtest results with:
    # 1. Period Returns
    # 2. Cumulative Returns
    # 3. Rolling Sharpe Ratio
    # 4. Return Distribution
    # ... (function body for plotting)

if __name__ == '__main__':
    # Run the rolling backtest with default parameters
    df_results = run_rolling_backtest(window_months=3, ticker="BTC-USD",
                                      strategy_params={'w_sma': 0.4, 'w_rsi': 0.3, 'w_macd': 0.3,
                                                       'consensus_threshold': 0.1, 'stop_loss_perc': 0.04})

    print("\n=== ROLLING BACKTEST RESULTS ===")
    print(df_results)
    report_stats(df_results)
    plot_four_charts(df_results)

Pasted image 20250619212847.png ### Conclusion: A Foundation for Strategic Evolution

The Ensemble Strategy with Fixed Weights offers a pragmatic approach to trading by combining multiple, well-understood indicators. This implementation serves as a strong foundation, demonstrating how a weighted consensus can lead to more robust signals than individual indicators alone. The rolling backtest is an indispensable tool for validating this robustness, revealing the strategy’s consistency and performance characteristics across diverse market conditions.

While fixed weights provide simplicity, future enhancements could involve dynamic weight optimization, where weights are adjusted periodically based on the recent performance of each sub-model. This would transition the strategy from a static combination to a truly adaptive ensemble, pushing towards even better and more consistent results in ever-evolving markets.