Article

Trading Breakouts from Consolidation with Momentum Ignition Strategy - Optimization and Rolling Backtest

The Momentum Ignition Strategy is designed to identify and capitalize on strong directional moves that emerge after periods of price consolidation. The core idea is that tight price ranges often precede significant breakouts. By combining volatility analysis for consolidation detection with Rate of Change (ROC) for momentum ignition, and filtering with a long-term trend, the strategy aims to enter trades at the cusp of a new trend. All positions are managed with a dynamic ATR-based trailing stop.

Strategy Overview

The MomentumIgnitionStrategy defines trade entries and exits based on several key components:

Entry Logic: An entry is triggered when the following conditions align, indicating a potential strong directional move:

Consolidation Detection: The strategy first identifies periods of low volatility, where price is consolidating. This is determined by comparing the standard deviation of the closing price over a consolidation_period to the current close price. If this ratio falls below a consolidation_threshold, the market is considered to be consolidating.
Momentum Ignition: Once consolidation is detected, the strategy looks for a sudden surge in momentum that breaks out of the ROC’s recent range. The ROC (Rate of Change) is calculated, and its moving average and standard deviation are used to define a channel.
- Bullish Ignition: The current ROC value rises above its ROC moving average plus a multiple (roc_breakout_std) of its ROC standard deviation.
- Bearish Ignition: The current ROC value falls below its ROC moving average minus a multiple (roc_breakout_std) of its ROC standard deviation.
Trend Filter: To align trades with the broader market direction, a long-term Simple Moving Average (SMA) acts as a trend filter.
- For long entries, the current close price must be above the trend_sma.
- For short entries, the current close price must be below the trend_sma.

All three sets of conditions (consolidation, momentum ignition, and trend alignment) must be met for an order to be placed.

Exit Logic (ATR Trailing Stop): Once a position is open, a dynamic ATR-based trailing stop is employed for risk management and profit protection. This type of stop automatically adjusts as the price moves in the favor of the trade.

For long positions, the stop_price continually updates to trail the highest_price_since_entry by a fixed multiple (atr_stop_multiplier) of the Average True Range (ATR). If the price falls below this stop, the position is closed.
For short positions, the stop_price updates to trail the lowest_price_since_entry by a fixed multiple of the ATR. If the price rises above this stop, the position is closed.

Backtrader Implementation

Here’s the MomentumIgnitionStrategy class:

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import dateutil.relativedelta as rd
import warnings
from scipy import stats # Not directly used in the strategy logic provided, but in overall script
import seaborn as sns # Not directly used in the strategy logic provided, but in overall script
import multiprocessing # Not directly used in the strategy logic provided, but in overall script

warnings.filterwarnings("ignore")

class MomentumIgnitionStrategy(bt.Strategy):
    params = (
        ('consolidation_period', 30),
        ('consolidation_threshold', 0.05),
        ('roc_period', 7),
        ('roc_ma_period', 30),
        ('roc_breakout_std', 1.0),
        ('trend_period', 30),
        ('atr_period', 7),
        ('atr_stop_multiplier', 3.0),
        ('order_percentage', 0.95),
        ('printlog', False),
    )

    def __init__(self):
        self.order = None
        self.price_stddev = bt.indicators.StandardDeviation(self.data.close, period=self.p.consolidation_period)
        self.roc = bt.indicators.RateOfChange(self.data.close, period=self.p.roc_period)
        self.roc_ma = bt.indicators.SimpleMovingAverage(self.roc, period=self.p.roc_ma_period)
        self.roc_stddev = bt.indicators.StandardDeviation(self.roc, period=self.p.roc_ma_period)
        self.trend_sma = bt.indicators.SimpleMovingAverage(self.data.close, period=self.p.trend_period)
        self.atr = bt.indicators.AverageTrueRange(self.data, period=self.p.atr_period)
        
        self.stop_price = None
        self.highest_price_since_entry = None
        self.lowest_price_since_entry = None
        
        self.trade_count = 0
        self.long_entries = 0
        self.short_entries = 0
        self.atr_stop_exits = 0
        self.profitable_trades = 0

    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]: 
            return
        if order.status in [order.Completed]:
            if order.isbuy():
                self.trade_count += 1
                self.long_entries += 1
                if self.position: # Position is now open
                    self.highest_price_since_entry = self.data.high[0]
                    self.stop_price = self.highest_price_since_entry - (self.atr[0] * self.p.atr_stop_multiplier)
            elif order.issell():
                self.trade_count += 1
                self.short_entries += 1
                if self.position: # Position is now open
                    self.lowest_price_since_entry = self.data.low[0]
                    self.stop_price = self.lowest_price_since_entry + (self.atr[0] * self.p.atr_stop_multiplier)
            if not self.position: # Position was closed
                self.stop_price = None
                self.highest_price_since_entry = None
                self.lowest_price_since_entry = None
        self.order = None

    def notify_trade(self, trade):
        if not trade.isclosed: 
            return
        if trade.pnl > 0:
            self.profitable_trades += 1

    def next(self):
        if self.order: 
            return

        if not self.position:
            # Entry Logic
            is_consolidating = (self.price_stddev[0] / self.data.close[0]) < self.p.consolidation_threshold
            is_macro_uptrend = self.data.close[0] > self.trend_sma[0]
            is_macro_downtrend = self.data.close[0] < self.trend_sma[0]

            if is_consolidating:
                roc_upper_band = self.roc_ma[0] + (self.roc_stddev[0] * self.p.roc_breakout_std)
                roc_lower_band = self.roc_ma[0] - (self.roc_stddev[0] * self.p.roc_breakout_std)
                
                is_mom_breakout_up = self.roc[0] > roc_upper_band
                is_mom_breakout_down = self.roc[0] < roc_lower_band
                
                if is_macro_uptrend and is_mom_breakout_up:
                    cash = self.broker.get_cash()
                    size = (cash * self.p.order_percentage) / self.data.close[0]
                    self.order = self.buy(size=size)
                elif is_macro_downtrend and is_mom_breakout_down:
                    cash = self.broker.get_cash()
                    size = (cash * self.p.order_percentage) / self.data.close[0]
                    self.order = self.sell(size=size)

        elif self.position:
            # Trailing Stop Exit Logic
            if self.position.size > 0: # Long position
                self.highest_price_since_entry = max(self.highest_price_since_entry, self.data.high[0])
                new_stop = self.highest_price_since_entry - (self.atr[0] * self.p.atr_stop_multiplier)
                self.stop_price = max(self.stop_price, new_stop) # Only move stop up
                if self.data.close[0] < self.stop_price: 
                    self.order = self.close()
                    self.atr_stop_exits += 1
            elif self.position.size < 0: # Short position
                self.lowest_price_since_entry = min(self.lowest_price_since_entry, self.data.low[0])
                new_stop = self.lowest_price_since_entry + (self.atr[0] * self.p.atr_stop_multiplier)
                self.stop_price = min(self.stop_price, new_stop) # Only move stop down
                if self.data.close[0] > self.stop_price: 
                    self.order = self.close()
                    self.atr_stop_exits += 1

Parameters (`params`):

consolidation_period: Period for calculating price standard deviation to detect consolidation.
consolidation_threshold: The maximum normalized standard deviation (stddev / close price) for a period to be considered consolidating.
roc_period: Period for the Rate of Change indicator.
roc_ma_period: Period for the moving average of ROC, used to define its channel.
roc_breakout_std: Multiplier for the standard deviation of ROC to define its breakout bands.
trend_period: Period for the Simple Moving Average (SMA) used as the long-term trend filter.
atr_period: Period for the Average True Range (ATR) indicator.
atr_stop_multiplier: Multiplier for ATR to set the distance of the trailing stop.
order_percentage: The percentage of available cash to use for each trade.
printlog: A boolean flag to enable or disable logging.

Initialization (`init`):

self.order: Tracks any pending orders to prevent multiple orders on the same bar.
Indicators:
- self.price_stddev: Standard Deviation of closing price, for consolidation detection.
- self.roc: Rate of Change of closing price.
- self.roc_ma: Simple Moving Average of the ROC.
- self.roc_stddev: Standard Deviation of the ROC, used to define ROC bands.
- self.trend_sma: Simple Moving Average of closing price for macro trend filtering.
- self.atr: Average True Range for trailing stop calculation.
State Variables for Trailing Stop: self.stop_price, self.highest_price_since_entry, self.lowest_price_since_entry are initialized.
Custom Counters: trade_count, long_entries, short_entries, atr_stop_exits, profitable_trades are initialized to track strategy-specific performance metrics.

Order and Trade Notifications (`notify_order`, `notify_trade`):

notify_order: This method is called by backtrader when an order’s status changes.
- Upon a completed buy or sell order that opens a position, it increments trade_count, long_entries/short_entries, and initializes the stop_price and highest_price_since_entry/lowest_price_since_entry for the ATR trailing stop.
- If an order closes a position, all trailing stop related state variables are reset.
- self.order is reset after completion or failure.
notify_trade: This method is called when a trade is fully closed. It increments self.profitable_trades if the trade generated a profit.

Main Logic (`next`):

The next method contains the core trading logic, executed on each new bar of data:

Pending Order Check: if self.order: return ensures no new orders are placed while one is pending.
Entry Logic (if not self.position):
- Consolidation Check: is_consolidating is true if the normalized price standard deviation is below consolidation_threshold.
- Macro Trend Check: is_macro_uptrend (close > trend_sma) or is_macro_downtrend (close < trend_sma).
- Momentum Breakout Check: If consolidating, it defines roc_upper_band and roc_lower_band based on roc_ma and roc_stddev. is_mom_breakout_up is true if roc[0] crosses above roc_upper_band; is_mom_breakout_down is true if roc[0] crosses below roc_lower_band.
- Order Placement: If is_macro_uptrend AND is_mom_breakout_up, a buy order is placed. If is_macro_downtrend AND is_mom_breakout_down, a sell order is placed. Position size is determined by order_percentage.
Exit Logic (elif self.position - ATR Trailing Stop):
- Long Position: self.highest_price_since_entry is updated. A new_stop is calculated using atr_stop_multiplier. self.stop_price is updated to always move in the profitable direction (max for long). If data.close[0] falls below self.stop_price, the position is closed, and self.atr_stop_exits is incremented.
- Short Position: self.lowest_price_since_entry is updated. A new_stop is calculated. self.stop_price is updated to always move in the profitable direction (min for short). If data.close[0] rises above self.stop_price, the position is closed, and self.atr_stop_exits is incremented.

Comprehensive Analysis Framework

The following code provides a complete framework for analyzing the MomentumIgnitionStrategy, including optimization and a rolling backtest.

def run_momentum_analysis():
    # OPTIMIZATION
    print("="*60)
    print("MOMENTUM IGNITION STRATEGY OPTIMIZATION")
    print("="*60)

    df = yf.download('BTC-USD', start='2020-01-01', end='2025-01-01', auto_adjust=False, progress=False)
    if isinstance(df.columns, pd.MultiIndex):
        df = df.droplevel(1, axis=1)

    print(f"Fetched data: {df.shape}")

    cerebro = bt.Cerebro()
    data = bt.feeds.PandasData(dataname=df)
    cerebro.adddata(data)
    cerebro.broker.setcash(10000.0)
    cerebro.broker.setcommission(commission=0.001)
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', timeframe=bt.TimeFrame.Days, annualize=True, riskfreerate=0.0)
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')

    print("Testing parameter combinations...")
    cerebro.optstrategy(
        MomentumIgnitionStrategy,
        consolidation_period=[20, 30, 40],
        consolidation_threshold=[0.03, 0.05, 0.07],
        roc_period=[5, 7, 10],
        roc_breakout_std=[1.0, 1.5, 2.0],
        atr_stop_multiplier=[2.0, 3.0, 4.0]
    )

    stratruns = cerebro.run(maxcpus=1)  # Force single-threaded to avoid multiprocessing issues
    print(f"Optimization complete! Tested {len(stratruns)} combinations.")

    # Find best parameters
    best_result = None
    best_sharpe = -999

    for run in stratruns:
        strategy = run[0]
        sharpe_analysis = strategy.analyzers.sharpe.get_analysis()
        returns_analysis = strategy.analyzers.returns.get_analysis()
        
        sharpe_ratio = sharpe_analysis.get('sharperatio', None)
        if sharpe_ratio is not None and not np.isnan(sharpe_ratio) and sharpe_ratio > best_sharpe:
            best_sharpe = sharpe_ratio
            best_result = {
                'consolidation_period': strategy.p.consolidation_period,
                'consolidation_threshold': strategy.p.consolidation_threshold,
                'roc_period': strategy.p.roc_period,
                'roc_breakout_std': strategy.p.roc_breakout_std,
                'atr_stop_multiplier': strategy.p.atr_stop_multiplier,
                'sharpe': sharpe_ratio,
                'return': returns_analysis.get('rtot', 0) * 100
            }

    if best_result:
        print(f"\nBEST PARAMETERS FOUND:")
        print(f"Consolidation Period: {best_result['consolidation_period']}")
        print(f"Consolidation Threshold: {best_result['consolidation_threshold']}")
        print(f"ROC Period: {best_result['roc_period']}")
        print(f"ROC Breakout Std: {best_result['roc_breakout_std']}")
        print(f"ATR Stop Multiplier: {best_result['atr_stop_multiplier']}")
        print(f"Sharpe Ratio: {best_result['sharpe']:.3f}")
        print(f"Total Return: {best_result['return']:.1f}%")
    else:
        print("No valid results found, using defaults")
        best_result = {
            'consolidation_period': 30,
            'consolidation_threshold': 0.05,
            'roc_period': 7,
            'roc_breakout_std': 1.0,
            'atr_stop_multiplier': 3.0
        }

    # ROLLING BACKTEST
    print(f"\n{'='*60}")
    print("RUNNING YEARLY ROLLING BACKTESTS")
    print(f"{'='*60}")

    strategy_params = {
        'consolidation_period': best_result['consolidation_period'],
        'consolidation_threshold': best_result['consolidation_threshold'], 
        'roc_period': best_result['roc_period'],
        'roc_breakout_std': best_result['roc_breakout_std'],
        'atr_stop_multiplier': best_result['atr_stop_multiplier'],
        'order_percentage': 0.95,
        'printlog': False
    }

    results = []
    start_dt = pd.to_datetime("2018-01-01")
    end_dt = pd.to_datetime("2025-01-01")
    current_start = start_dt

    while True:
        current_end = current_start + rd.relativedelta(months=12)
        if current_end > end_dt:
            break
            
        print(f"Period: {current_start.date()} to {current_end.date()}")
        
        data = yf.download("BTC-USD", start=current_start, end=current_end, auto_adjust=False, progress=False)
        if data.empty or len(data) < 90:
            current_start += rd.relativedelta(months=12)
            continue
            
        if isinstance(data.columns, pd.MultiIndex):
            data = data.droplevel(1, 1)
        
        start_price = data['Close'].iloc[0]
        end_price = data['Close'].iloc[-1]
        benchmark_return = (end_price - start_price) / start_price
        
        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        cerebro.addstrategy(MomentumIgnitionStrategy, **strategy_params)
        cerebro.adddata(feed)
        cerebro.broker.setcash(100000)
        cerebro.broker.setcommission(commission=0.001)
        
        start_val = cerebro.broker.getvalue()
        result = cerebro.run()
        final_val = cerebro.broker.getvalue()
        strategy_return = (final_val - start_val) / start_val
        
        strategy_instance = result[0]
        trades = getattr(strategy_instance, 'trade_count', 0)
        long_entries = getattr(strategy_instance, 'long_entries', 0)
        short_entries = getattr(strategy_instance, 'short_entries', 0)
        profitable = getattr(strategy_instance, 'profitable_trades', 0)
        
        results.append({
            'year': current_start.year,
            'strategy_return': strategy_return,
            'benchmark_return': benchmark_return,
            'trades': trades,
            'long_entries': long_entries,
            'short_entries': short_entries,
            'profitable_trades': profitable
        })
        
        print(f"Strategy: {strategy_return:.1%} | Buy&Hold: {benchmark_return:.1%} | Trades: {trades}")
        current_start += rd.relativedelta(months=12)

    # RESULTS & PLOTTING
    results_df = pd.DataFrame(results)

    if not results_df.empty:
        print(f"\n{'='*80}")
        print("YEARLY RESULTS")
        print(f"{'='*80}")
        print("Year | Strategy | Buy&Hold | Outperf | Trades | Long | Short | WinRate")
        print("-" * 80)
        
        for _, row in results_df.iterrows():
            strat_ret = row['strategy_return'] * 100
            bench_ret = row['benchmark_return'] * 100
            outperf = strat_ret - bench_ret
            win_rate = (row['profitable_trades'] / max(1, row['trades'])) * 100
            
            print(f"{int(row['year'])} | {strat_ret:7.1f}% | {bench_ret:7.1f}% | {outperf:+6.1f}% | "
                  f"{int(row['trades']):6d} | {int(row['long_entries']):4d} | {int(row['short_entries']):5d} | {win_rate:6.1f}%")
        
        # PLOTS
        fig, axes = plt.subplots(2, 2, figsize=(16, 10))
        fig.suptitle('Momentum Ignition Strategy Analysis', fontsize=16, fontweight='bold')
        
        # Cumulative performance
        strategy_cumulative = (1 + results_df['strategy_return']).cumprod()
        benchmark_cumulative = (1 + results_df['benchmark_return']).cumprod()
        
        ax1 = axes[0, 0]
        ax1.plot(results_df['year'], strategy_cumulative, 'o-', linewidth=3, color='purple', label='Strategy')
        ax1.plot(results_df['year'], benchmark_cumulative, 's-', linewidth=3, color='orange', label='Buy & Hold')
        ax1.set_title('Cumulative Performance')
        ax1.set_ylabel('Cumulative Return')
        ax1.legend()
        ax1.grid(True, alpha=0.3)
        
        # Annual returns
        ax2 = axes[0, 1]
        x_pos = range(len(results_df))
        width = 0.35
        ax2.bar([x - width/2 for x in x_pos], results_df['strategy_return'] * 100, width, 
                label='Strategy', color='purple', alpha=0.8)
        ax2.bar([x + width/2 for x in x_pos], results_df['benchmark_return'] * 100, width,
                label='Buy & Hold', color='orange', alpha=0.8)
        ax2.set_title('Annual Returns')
        ax2.set_ylabel('Return (%)')
        ax2.set_xticks(x_pos)
        ax2.set_xticklabels([int(year) for year in results_df['year']])
        ax2.legend()
        ax2.grid(True, alpha=0.3)
        
        # Long vs Short
        ax3 = axes[1, 0]
        ax3.bar(x_pos, results_df['long_entries'], label='Long', color='green', alpha=0.8)
        ax3.bar(x_pos, results_df['short_entries'], bottom=results_df['long_entries'], 
                label='Short', color='red', alpha=0.8)
        ax3.set_title('Long vs Short Entries')
        ax3.set_ylabel('Number of Entries')
        ax3.set_xticks(x_pos)
        ax3.set_xticklabels([int(year) for year in results_df['year']])
        ax3.legend()
        ax3.grid(True, alpha=0.3)
        
        # Summary stats
        ax4 = axes[1, 1]
        ax4.axis('off')
        
        total_strategy_return = strategy_cumulative.iloc[-1] - 1
        total_benchmark_return = benchmark_cumulative.iloc[-1] - 1
        total_trades = results_df['trades'].sum()
        total_profitable = results_df['profitable_trades'].sum()
        
        summary_text = f"""
SUMMARY STATISTICS

Total Strategy Return: {total_strategy_return:.1%}
Total Buy & Hold Return: {total_benchmark_return:.1%}
Outperformance: {total_strategy_return - total_benchmark_return:+.1%}

Total Trades: {int(total_trades)}
Overall Win Rate: {total_profitable/max(1,total_trades):.1%}
Long Entries: {int(results_df['long_entries'].sum())}
Short Entries: {int(results_df['short_entries'].sum())}

Strategy wins in {(results_df['strategy_return'] > results_df['benchmark_return']).sum()}/{len(results_df)} years
        """
        
        ax4.text(0.1, 0.9, summary_text, transform=ax4.transAxes, va='top', ha='left',
                  fontsize=12, fontfamily='monospace',
                  bbox=dict(boxstyle="round,pad=0.5", facecolor="lightgray", alpha=0.8))
        
        plt.tight_layout()
        plt.show()
        
        # Final verdict
        print(f"\n{'='*60}")
        if total_strategy_return > total_benchmark_return:
            print(f"STRATEGY WINS by {total_strategy_return - total_benchmark_return:+.1%}!")
        else:
            print(f"Strategy underperformed by {total_strategy_return - total_benchmark_return:.1%}")
        print(f"{'='*60}")

    else:
        print("No results generated!")

    print("ANALYSIS COMPLETE!")

if __name__ == '__main__':
    multiprocessing.freeze_support() # Required for Windows
    run_momentum_analysis()

Parameters (`params`):

consolidation_period: Period for calculating price standard deviation to detect consolidation.
consolidation_threshold: The maximum normalized standard deviation (stddev / close price) for a period to be considered consolidating.
roc_period: Period for the Rate of Change (ROC) indicator.
roc_ma_period: Period for the moving average of ROC, used to define its channel.
roc_breakout_std: Multiplier for the standard deviation of ROC to define its breakout bands.
trend_period: Period for the Simple Moving Average (SMA) used as the long-term trend filter.
atr_period: Period for the Average True Range (ATR) indicator.
atr_stop_multiplier: Multiplier for ATR to set the distance of the trailing stop.
order_percentage: The percentage of available cash to use for each trade.
printlog: A boolean flag to enable or disable logging.

Initialization (`init`):

self.order: Tracks any pending orders to prevent multiple orders on the same bar.
Indicators:
- self.price_stddev: Standard Deviation of closing price, for consolidation detection.
- self.roc: Rate of Change of closing price.
- self.roc_ma: Simple Moving Average of the ROC.
- self.roc_stddev: Standard Deviation of the ROC, used to define ROC bands.
- self.trend_sma: Simple Moving Average of closing price for macro trend filtering.
- self.atr: Average True Range for trailing stop calculation.
State Variables for Trailing Stop: self.stop_price, self.highest_price_since_entry, self.lowest_price_since_entry are initialized.
Custom Counters: trade_count, long_entries, short_entries, atr_stop_exits, profitable_trades are initialized to track strategy-specific performance metrics.

Order and Trade Notifications (`notify_order`, `notify_trade`):

notify_order: This method is called by backtrader when an order’s status changes.
- Upon a completed buy or sell order that opens a position, it increments trade_count, long_entries/short_entries, and initializes the stop_price and highest_price_since_entry/lowest_price_since_entry for the ATR trailing stop.
- If an order closes a position, all trailing stop related state variables are reset.
- self.order is reset after completion or failure.
notify_trade: This method is called when a trade is fully closed. It increments self.profitable_trades if the trade generated a profit.

Main Logic (`next`):

The next method contains the core trading logic, executed on each new bar of data:

Pending Order Check: if self.order: return ensures no new orders are placed while one is pending.
Entry Logic (if not self.position):
- Consolidation Check: is_consolidating is true if the normalized price standard deviation is below consolidation_threshold.
- Macro Trend Check: is_macro_uptrend (close > trend_sma) or is_macro_downtrend (close < trend_sma).
- Momentum Breakout Check: If consolidating, it defines roc_upper_band and roc_lower_band based on roc_ma and roc_stddev. is_mom_breakout_up is true if roc[0] crosses above roc_upper_band; is_mom_breakout_down is true if roc[0] crosses below roc_lower_band.
- Order Placement: If is_macro_uptrend AND is_mom_breakout_up, a buy order is placed. If is_macro_downtrend AND is_mom_breakout_down, a sell order is placed. Position size is determined by order_percentage.
Exit Logic (elif self.position - ATR Trailing Stop):
- Long Position: self.highest_price_since_entry is updated. A new_stop is calculated using atr_stop_multiplier. self.stop_price is updated to always move in the profitable direction (max for long). If data.close[0] falls below self.stop_price, the position is closed, and self.atr_stop_exits is incremented.
- Short Position: self.lowest_price_since_entry is updated. A new_stop is calculated. self.stop_price is updated to always move in the profitable direction (min for short). If data.close[0] rises above self.stop_price, the position is closed, and self.atr_stop_exits is incremented.

Comprehensive Analysis Framework

This section presents the full code for running optimization, a rolling backtest, and generating detailed results with plots.

def run_momentum_analysis():
    # OPTIMIZATION
    print("="*60)
    print("MOMENTUM IGNITION STRATEGY OPTIMIZATION")
    print("="*60)

    df = yf.download('BTC-USD', start='2020-01-01', end='2025-01-01', auto_adjust=False, progress=False)
    if isinstance(df.columns, pd.MultiIndex):
        df = df.droplevel(1, axis=1)

    print(f"Fetched data: {df.shape}")

    cerebro = bt.Cerebro()
    data = bt.feeds.PandasData(dataname=df)
    cerebro.adddata(data)
    cerebro.broker.setcash(10000.0)
    cerebro.broker.setcommission(commission=0.001)
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', timeframe=bt.TimeFrame.Days, annualize=True, riskfreerate=0.0)
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')

    print("Testing parameter combinations...")
    cerebro.optstrategy(
        MomentumIgnitionStrategy,
        consolidation_period=[20, 30, 40],
        consolidation_threshold=[0.03, 0.05, 0.07],
        roc_period=[5, 7, 10],
        roc_breakout_std=[1.0, 1.5, 2.0],
        atr_stop_multiplier=[2.0, 3.0, 4.0]
    )

    stratruns = cerebro.run(maxcpus=1)  # Force single-threaded to avoid multiprocessing issues
    print(f"Optimization complete! Tested {len(stratruns)} combinations.")

    # Find best parameters
    best_result = None
    best_sharpe = -999

    for run in stratruns:
        strategy = run[0]
        sharpe_analysis = strategy.analyzers.sharpe.get_analysis()
        returns_analysis = strategy.analyzers.returns.get_analysis()
        
        sharpe_ratio = sharpe_analysis.get('sharperatio', None)
        if sharpe_ratio is not None and not np.isnan(sharpe_ratio) and sharpe_ratio > best_sharpe:
            best_sharpe = sharpe_ratio
            best_result = {
                'consolidation_period': strategy.p.consolidation_period,
                'consolidation_threshold': strategy.p.consolidation_threshold,
                'roc_period': strategy.p.roc_period,
                'roc_breakout_std': strategy.p.roc_breakout_std,
                'atr_stop_multiplier': strategy.p.atr_stop_multiplier,
                'sharpe': sharpe_ratio,
                'return': returns_analysis.get('rtot', 0) * 100
            }

    if best_result:
        print(f"\nBEST PARAMETERS FOUND:")
        print(f"Consolidation Period: {best_result['consolidation_period']}")
        print(f"Consolidation Threshold: {best_result['consolidation_threshold']}")
        print(f"ROC Period: {best_result['roc_period']}")
        print(f"ROC Breakout Std: {best_result['roc_breakout_std']}")
        print(f"ATR Stop Multiplier: {best_result['atr_stop_multiplier']}")
        print(f"Sharpe Ratio: {best_result['sharpe']:.3f}")
        print(f"Total Return: {best_result['return']:.1f}%")
    else:
        print("No valid results found, using defaults")
        best_result = {
            'consolidation_period': 30,
            'consolidation_threshold': 0.05,
            'roc_period': 7,
            'roc_breakout_std': 1.0,
            'atr_stop_multiplier': 3.0
        }

    # ROLLING BACKTEST
    print(f"\n{'='*60}")
    print("RUNNING YEARLY ROLLING BACKTESTS")
    print(f"{'='*60}")

    strategy_params = {
        'consolidation_period': best_result['consolidation_period'],
        'consolidation_threshold': best_result['consolidation_threshold'], 
        'roc_period': best_result['roc_period'],
        'roc_breakout_std': best_result['roc_breakout_std'],
        'atr_stop_multiplier': best_result['atr_stop_multiplier'],
        'order_percentage': 0.95,
        'printlog': False
    }

    results = []
    start_dt = pd.to_datetime("2018-01-01")
    end_dt = pd.to_datetime("2025-01-01")
    current_start = start_dt

    while True:
        current_end = current_start + rd.relativedelta(months=12)
        if current_end > end_dt:
            break
            
        print(f"Period: {current_start.date()} to {current_end.date()}")
        
        data = yf.download("BTC-USD", start=current_start, end=current_end, auto_adjust=False, progress=False)
        if data.empty or len(data) < 90:
            current_start += rd.relativedelta(months=12)
            continue
            
        if isinstance(data.columns, pd.MultiIndex):
            data = data.droplevel(1, 1)
        
        start_price = data['Close'].iloc[0]
        end_price = data['Close'].iloc[-1]
        benchmark_return = (end_price - start_price) / start_price
        
        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        cerebro.addstrategy(MomentumIgnitionStrategy, **strategy_params)
        cerebro.adddata(feed)
        cerebro.broker.setcash(100000)
        cerebro.broker.setcommission(commission=0.001)
        
        start_val = cerebro.broker.getvalue()
        result = cerebro.run()
        final_val = cerebro.broker.getvalue()
        strategy_return = (final_val - start_val) / start_val
        
        strategy_instance = result[0]
        trades = getattr(strategy_instance, 'trade_count', 0)
        long_entries = getattr(strategy_instance, 'long_entries', 0)
        short_entries = getattr(strategy_instance, 'short_entries', 0)
        profitable = getattr(strategy_instance, 'profitable_trades', 0)
        
        results.append({
            'year': current_start.year,
            'strategy_return': strategy_return,
            'benchmark_return': benchmark_return,
            'trades': trades,
            'long_entries': long_entries,
            'short_entries': short_entries,
            'profitable_trades': profitable
        })
        
        print(f"Strategy: {strategy_return:.1%} | Buy&Hold: {benchmark_return:.1%} | Trades: {trades}")
        current_start += rd.relativedelta(months=12)

    # RESULTS & PLOTTING
    results_df = pd.DataFrame(results)

    if not results_df.empty:
        print(f"\n{'='*80}")
        print("YEARLY RESULTS")
        print(f"{'='*80}")
        print("Year | Strategy | Buy&Hold | Outperf | Trades | Long | Short | WinRate")
        print("-" * 80)
        
        for _, row in results_df.iterrows():
            strat_ret = row['strategy_return'] * 100
            bench_ret = row['benchmark_return'] * 100
            outperf = strat_ret - bench_ret
            win_rate = (row['profitable_trades'] / max(1, row['trades'])) * 100
            
            print(f"{int(row['year'])} | {strat_ret:7.1f}% | {bench_ret:7.1f}% | {outperf:+6.1f}% | "
                  f"{int(row['trades']):6d} | {int(row['long_entries']):4d} | {int(row['short_entries']):5d} | {win_rate:6.1f}%")
        
        # PLOTS
        fig, axes = plt.subplots(2, 2, figsize=(16, 10))
        fig.suptitle('Momentum Ignition Strategy Analysis', fontsize=16, fontweight='bold')
        
        # Cumulative performance
        strategy_cumulative = (1 + results_df['strategy_return']).cumprod()
        benchmark_cumulative = (1 + results_df['benchmark_return']).cumprod()
        
        ax1 = axes[0, 0]
        ax1.plot(results_df['year'], strategy_cumulative, 'o-', linewidth=3, color='purple', label='Strategy')
        ax1.plot(results_df['year'], benchmark_cumulative, 's-', linewidth=3, color='orange', label='Buy & Hold')
        ax1.set_title('Cumulative Performance')
        ax1.set_ylabel('Cumulative Return')
        ax1.legend()
        ax1.grid(True, alpha=0.3)
        
        # Annual returns
        ax2 = axes[0, 1]
        x_pos = range(len(results_df))
        width = 0.35
        ax2.bar([x - width/2 for x in x_pos], results_df['strategy_return'] * 100, width, 
                label='Strategy', color='purple', alpha=0.8)
        ax2.bar([x + width/2 for x in x_pos], results_df['benchmark_return'] * 100, width,
                label='Buy & Hold', color='orange', alpha=0.8)
        ax2.set_title('Annual Returns')
        ax2.set_ylabel('Return (%)')
        ax2.set_xticks(x_pos)
        ax2.set_xticklabels([int(year) for year in results_df['year']])
        ax2.legend()
        ax2.grid(True, alpha=0.3)
        
        # Long vs Short
        ax3 = axes[1, 0]
        ax3.bar(x_pos, results_df['long_entries'], label='Long', color='green', alpha=0.8)
        ax3.bar(x_pos, results_df['short_entries'], bottom=results_df['long_entries'], 
                label='Short', color='red', alpha=0.8)
        ax3.set_title('Long vs Short Entries')
        ax3.set_ylabel('Number of Entries')
        ax3.set_xticks(x_pos)
        ax3.set_xticklabels([int(year) for year in results_df['year']])
        ax3.legend()
        ax3.grid(True, alpha=0.3)
        
        # Summary stats
        ax4 = axes[1, 1]
        ax4.axis('off')
        
        total_strategy_return = strategy_cumulative.iloc[-1] - 1
        total_benchmark_return = benchmark_cumulative.iloc[-1] - 1
        total_trades = results_df['trades'].sum()
        total_profitable = results_df['profitable_trades'].sum()
        
        summary_text = f"""
SUMMARY STATISTICS

Total Strategy Return: {total_strategy_return:.1%}
Total Buy & Hold Return: {total_benchmark_return:.1%}
Outperformance: {total_strategy_return - total_benchmark_return:+.1%}

Total Trades: {int(total_trades)}
Overall Win Rate: {total_profitable/max(1,total_trades):.1%}
Long Entries: {int(results_df['long_entries'].sum())}
Short Entries: {int(results_df['short_entries'].sum())}

Strategy wins in {(results_df['strategy_return'] > results_df['benchmark_return']).sum()}/{len(results_df)} years
        """
        
        ax4.text(0.1, 0.9, summary_text, transform=ax4.transAxes, va='top', ha='left',
                  fontsize=12, fontfamily='monospace',
                  bbox=dict(boxstyle="round,pad=0.5", facecolor="lightgray", alpha=0.8))
        
        plt.tight_layout()
        plt.show()
        
        # Final verdict
        print(f"\n{'='*60}")
        if total_strategy_return > total_benchmark_return:
            print(f"STRATEGY WINS by {total_strategy_return - total_benchmark_return:+.1%}!")
        else:
            print(f"Strategy underperformed by {total_strategy_return - total_benchmark_return:.1%}")
        print(f"{'='*60}")

    else:
        print("No results generated!")

    print("ANALYSIS COMPLETE!")

Optimization (`run_momentum_analysis` - Optimization Section):

This part of run_momentum_analysis is responsible for finding the best combination of parameters for the MomentumIgnitionStrategy.

It downloads BTC-USD data for a defined optimization period (2020-2025).
A bt.Cerebro instance is set up, and the downloaded data is added. Initial cash, commission, and performance analyzers (SharpeRatio, Returns) are configured.
cerebro.optstrategy is used to run the MomentumIgnitionStrategy with a grid of parameter values (e.g., consolidation_period, consolidation_threshold, roc_period, etc.).
It then iterates through the results of all these backtests to find the set of parameters that yielded the highest Sharpe Ratio.
The best parameters and their associated performance metrics (Sharpe Ratio, total return) are printed. If no valid results are found, default parameters are used.

Rolling Backtest (`run_momentum_analysis` - Rolling Backtest Section):

After optimization, this section runs a rolling backtest with the identified best parameters to assess the strategy’s performance consistency over a broader and more representative historical period.

It initializes a list to store results for each rolling window.
It iterates through yearly periods from 2018-01-01 to 2025-01-01.
For each yearly window:
- It downloads BTC-USD data using yfinance (with auto_adjust=False and droplevel for column handling).
- It calculates the simple buy-and-hold return for the period as a benchmark.
- A new bt.Cerebro instance is set up for the current window, and the MomentumIgnitionStrategy is added with the optimized parameters.
- The backtest is run, and the strategy’s return and various custom trade statistics (total trades, long/short entries, profitable trades) are collected.
- Results for the current period are printed and appended to the results list.
The function then proceeds to the “RESULTS & PLOTTING” section.

Results and Plotting

This comprehensive framework allows for thorough testing and evaluation of the Momentum Ignition Strategy, from finding optimal parameters to understanding its performance and characteristics across various market conditions.

Strategy Overview

Backtrader Implementation

Parameters (params):

Initialization (__init__):

Order and Trade Notifications (notify_order, notify_trade):

Main Logic (next):

Comprehensive Analysis Framework

Parameters (params):

Initialization (__init__):

Order and Trade Notifications (notify_order, notify_trade):

Main Logic (next):

Comprehensive Analysis Framework

Optimization (run_momentum_analysis - Optimization Section):

Rolling Backtest (run_momentum_analysis - Rolling Backtest Section):

Results and Plotting

Parameters (`params`):

Initialization (`init`):

Order and Trade Notifications (`notify_order`, `notify_trade`):

Main Logic (`next`):

Parameters (`params`):

Initialization (`init`):

Order and Trade Notifications (`notify_order`, `notify_trade`):

Main Logic (`next`):

Optimization (`run_momentum_analysis` - Optimization Section):

Rolling Backtest (`run_momentum_analysis` - Rolling Backtest Section):