← Back to Home
Rolling Backtest of an EMA Crossover Trading Strategy with MACD, ADX, and Trailing Stops

Rolling Backtest of an EMA Crossover Trading Strategy with MACD, ADX, and Trailing Stops

This article introduces a comprehensive trading strategy, EmaMacdAdxStrategy, designed for trend following in both long and short directions. It combines the core concept of Exponential Moving Average (EMA) crossovers with the power of MACD and ADX indicators for signal filtering, and crucially, incorporates dynamic trailing stop-loss orders for effective risk management. The strategy’s performance is rigorously evaluated using a rolling backtesting framework.

1. The EMA, MACD, and ADX Strategy Concept

The EmaMacdAdxStrategy aims to identify and capitalize on strong trends while filtering out choppy market conditions. It achieves this by combining three popular technical indicators:

Entry Logic:

Exit Logic:

The strategy utilizes a two-pronged approach for exits:

  1. Indicator-based Exit:
    • For an existing long position: Exit if Fast EMA < Slow EMA OR MACD Line < Signal Line (reversal of primary or momentum trend).
    • For an existing short position: Exit if Fast EMA > Slow EMA OR MACD Line > Signal Line (reversal of primary or momentum trend).
  2. Trailing Stop Exit:
    • A StopTrail order is placed immediately after an entry. This order automatically adjusts its price to a percentage below (for long) or above (for short) the highest/lowest price reached since entry. If the price moves against the position and hits this trailing level, the position is automatically closed, ensuring profit protection or loss limitation.

2. The EmaMacdAdxStrategy Implementation

Here’s the core backtrader strategy code:

import backtrader as bt
import yfinance as yf
import datetime
import matplotlib.pyplot as plt
import pandas as pd # Ensure pandas is imported for DataFrame operations

class EmaMacdAdxStrategy(bt.Strategy):
    """
    EMA Crossover Strategy with MACD and ADX Filters and Trailing Stop-Loss.
    - Long Entry: Fast EMA > Slow EMA AND MACD Line > Signal Line AND ADX > Threshold
    - Short Entry: Fast EMA < Slow EMA AND MACD Line < Signal Line AND ADX > Threshold
    - Exit:
        - Opposite EMA cross OR MACD cross reversal (Indicator Exit)
        - OR Price hits trailing stop level (Stop Exit)
    """
    params = (
        ('ema_fast_period', 7),
        ('ema_slow_period', 30),
        ('macd_fast_period', 12),
        ('macd_slow_period', 26),
        ('macd_signal_period', 9),
        ('adx_period', 14),
        ('adx_threshold', 25.0),
        ('trail_percent', 0.02), # Trailing stop percentage (e.g., 0.02 for 2%)
        ('printlog', True),      # Enable/Disable logging
    )

    def __init__(self):
        # Keep references to the closing prices
        self.dataclose = self.datas[0].close

        # Keep track of pending orders and buy price/commission
        self.order = None
        self.stop_order = None # To track the trailing stop order
        self.buyprice = None   # Track buy price for debugging/logging, not directly used for trailing stop
        self.buycomm = None    # Track commission for debugging/logging

        # --- Indicator Definitions ---
        # EMAs for crossover signals
        self.ema_fast = bt.indicators.ExponentialMovingAverage(
            self.datas[0], period=self.params.ema_fast_period)
        self.ema_slow = bt.indicators.ExponentialMovingAverage(
            self.datas[0], period=self.params.ema_slow_period)
        
        # MACD for momentum confirmation (using MACDHisto which provides MACD line and Signal line)
        self.macd = bt.indicators.MACDHisto(
            self.datas[0],
            period_me1=self.params.macd_fast_period,
            period_me2=self.params.macd_slow_period,
            period_signal=self.params.macd_signal_period)
        
        # ADX for trend strength confirmation
        self.adx = bt.indicators.AverageDirectionalMovementIndex(
            self.datas[0], period=self.params.adx_period)

        self.log(f"Strategy Parameters: Fast EMA={self.p.ema_fast_period}, Slow EMA={self.p.ema_slow_period}, "
                 f"ADX Period={self.p.adx_period}, ADX Threshold={self.p.adx_threshold}, "
                 f"Trailing Stop={self.p.trail_percent*100:.2f}%")

    def log(self, txt, dt=None, doprint=False):
        ''' Logging function for this strategy'''
        if self.params.printlog or doprint:
            dt = dt or self.datas[0].datetime.date(0)
            print(f'{dt.isoformat()} - {txt}')

    def notify_order(self, order):
        ''' Handles order notifications and trailing stop placement '''
        # Ignore submitted/accepted orders, wait for completion or failure
        if order.status in [order.Submitted, order.Accepted]:
            return

        # --- Handle Completed Order ---
        if order.status == order.Completed:
            if order.isbuy(): # --- Buy Order Completed ---
                self.log(
                    f'BUY EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}'
                )
                self.buyprice = order.executed.price # Store buy price
                self.buycomm = order.executed.comm   # Store commission

                # If it's an entry buy (i.e., we just opened a long position)
                if self.position.size > 0 and self.params.trail_percent:
                    self.log(f'>>> Placing SELL STOP TRAIL Order at {self.params.trail_percent * 100:.2f}%')
                    # Place a trailing stop order. 'size=self.position.size' ensures it closes the whole position.
                    self.stop_order = self.sell(exectype=bt.Order.StopTrail,
                                                 trailpercent=self.params.trail_percent,
                                                 size=self.position.size) # Specify size for stop-loss
                    
            elif order.issell(): # --- Sell Order Completed ---
                # Check if this sell order was the execution of our trailing stop (by checking its reference)
                is_stop_trail_execution = (self.stop_order is not None and 
                                           order.ref == self.stop_order.ref)

                if is_stop_trail_execution:
                    self.log(f'>>> SELL STOP TRAIL EXECUTED, Price: {order.executed.price:.2f}')
                    self.stop_order = None # Reset stop order tracker as it's been executed
                elif self.position.size < 0 and self.buyprice is None: # This implies a new short entry (no prior buyprice)
                    self.log(
                        f'SELL SHORT EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}'
                    )
                    # Place Trailing Stop Buy Order for short position
                    if self.params.trail_percent:
                        self.log(f'>>> Placing BUY STOP TRAIL Order at {self.params.trail_percent * 100:.2f}%')
                        self.stop_order = self.buy(exectype=bt.Order.StopTrail,
                                                    trailpercent=self.params.trail_percent,
                                                    size=abs(self.position.size)) # Specify size for stop-loss
                else: # This is a normal closing sell order (exiting a long position, not a trailing stop)
                    self.log(
                        f'SELL CLOSE EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}'
                    )
                    self.buyprice = None # Reset buy price for next trade cycle
                    self.buycomm = None

            self.order = None # Clear general order tracker after any order completes

        # --- Handle Non-Completed Order (Canceled, Margin, Rejected) ---
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log(f'Order Canceled/Margin/Rejected: Status {order.getstatusname()}, Ref: {order.ref}')
            
            # If the cancelled/rejected order was our tracked entry order
            if self.order and order.ref == self.order.ref:
                self.log('Entry Order Canceled/Rejected - Resetting tracker')
                self.order = None
            
            # If the cancelled/rejected order was our tracked stop order
            elif self.stop_order and order.ref == self.stop_order.ref:
                self.log('WARNING: Trailing Stop Order Canceled/Rejected - Resetting tracker')
                self.stop_order = None # Stop order failed, so we are now without trailing stop. Handle with care.


    def notify_trade(self, trade):
        ''' Handles trade notifications (when a position is fully closed) '''
        if not trade.isclosed:
            return # Ignore trades that are not yet closed

        self.log(f'OPERATION PROFIT, GROSS: {trade.pnl:.2f}, NET: {trade.pnlcomm:.2f}')
        
        # Reset any relevant state after a trade closes completely
        # (Self.stop_order and self.buyprice/buycomm are also reset in notify_order if relevant)
        if self.stop_order and not self.stop_order.alive(): # If stop order existed but is not alive (implies it was executed)
            self.stop_order = None # This handles case where stop_order completed itself, not via general 'close()'

    def next(self):
        ''' Core strategy logic executed on each bar '''
        # First, ensure no pending entry/primary exit orders. Trailing stops run asynchronously.
        if self.order:
            return

        # Ensure indicators have warmed up sufficiently
        min_warmup_period = max(self.p.ema_slow_period, self.p.macd_slow_period, self.p.adx_period) + 1
        if len(self.data) < min_warmup_period:
            return

        # --- Decision Logic: Enter or Exit Position ---

        # If we are currently flat (no position)
        if not self.position:
            # --- Potential LONG Entry Conditions ---
            long_condition_1 = self.ema_fast[0] > self.ema_slow[0]       # EMA crossover
            long_condition_2 = self.macd.macd[0] > self.macd.signal[0]   # MACD bullish
            long_condition_3 = self.adx.adx[0] > self.params.adx_threshold # ADX strong trend

            if long_condition_1 and long_condition_2 and long_condition_3:
                self.log(f'BUY CREATE - EMA/MACD/ADX confirmed LONG. Price: {self.dataclose[0]:.2f}')
                self.order = self.buy() # Place buy order. Trailing stop will be placed in notify_order.

            # --- Potential SHORT Entry Conditions ---
            else: # Only consider short if long conditions are not met
                short_condition_1 = self.ema_fast[0] < self.ema_slow[0]       # EMA crossover
                short_condition_2 = self.macd.macd[0] < self.macd.signal[0]   # MACD bearish
                short_condition_3 = self.adx.adx[0] > self.params.adx_threshold # ADX strong trend

                if short_condition_1 and short_condition_2 and short_condition_3:
                    self.log(f'SELL CREATE (Short) - EMA/MACD/ADX confirmed SHORT. Price: {self.dataclose[0]:.2f}')
                    self.order = self.sell() # Place sell (short) order. Trailing stop will be placed in notify_order.

        # If we are already in the market, check for indicator-based exits
        else:
            # --- Indicator-based EXIT for LONG position ---
            if self.position.size > 0: # Currently long
                long_exit_condition_1 = self.ema_fast[0] < self.ema_slow[0]     # EMA crossover reversal
                long_exit_condition_2 = self.macd.macd[0] < self.macd.signal[0] # MACD bearish reversal

                # Exit if either EMA or MACD indicates a trend reversal
                if long_exit_condition_1 or long_exit_condition_2:
                    self.log(f'INDICATOR EXIT - CLOSE LONG. Price: {self.dataclose[0]:.2f}')
                    # self.close() will automatically cancel any associated stop trail order for this position
                    self.order = self.close()

            # --- Indicator-based EXIT for SHORT position ---
            elif self.position.size < 0: # Currently short
                short_exit_condition_1 = self.ema_fast[0] > self.ema_slow[0]     # EMA crossover reversal
                short_exit_condition_2 = self.macd.macd[0] > self.macd.signal[0] # MACD bullish reversal

                # Exit if either EMA or MACD indicates a trend reversal
                if short_exit_condition_1 or short_exit_condition_2:
                    self.log(f'INDICATOR EXIT - CLOSE SHORT. Price: {self.dataclose[0]:.2f}')
                    # self.close() will automatically cancel any associated stop trail order for this position
                    self.order = self.close()

    def stop(self):
        self.log(f'Final Portfolio Value: {self.broker.getvalue():.2f}')

Explanation of EmaMacdAdxStrategy:

3. Backtesting and Analysis

The provided script includes a robust rolling backtesting framework to thoroughly evaluate the strategy’s performance.

# ... (imports and strategy definition as above) ...

# Define the strategy for the rolling backtest
strategy = EmaMacdAdxStrategy

def run_rolling_backtest(
    ticker="ADA-USD",
    start="2018-01-01",
    # Updated end date to the current date for a more live test
    end="2025-06-22", 
    window_months=3,
    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)
        # Adjust end of current window if it exceeds overall end date
        if current_end > end_dt:
            current_end = end_dt
            if current_start >= current_end: # No valid period left
                break

        print(f"\nROLLING BACKTEST: {current_start.date()} to {current_end.date()}")

        # Data download using yfinance, respecting the user's preference for auto_adjust=False and droplevel
        data = yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)
        
        # Apply droplevel if data is a MultiIndex, as per user's preference
        if isinstance(data.columns, pd.MultiIndex):
            data = data.droplevel(1, axis=1)

        # Check for sufficient data after droplevel for strategy warm-up
        # Calculate min bars needed based on strategy parameters
        ema_slow_period = strategy_params.get('ema_slow_period', EmaMacdAdxStrategy.params.ema_slow_period)
        macd_slow_period = strategy_params.get('macd_slow_period', EmaMacdAdxStrategy.params.macd_slow_period)
        adx_period = strategy_params.get('adx_period', EmaMacdAdxStrategy.params.adx_period)
        min_bars_needed = max(ema_slow_period, macd_slow_period, adx_period) + 1 # +1 for current bar's data

        if data.empty or len(data) < min_bars_needed:
            print(f"Not enough data for period {current_start.date()} to {current_end.date()} (requires at least {min_bars_needed} bars). Skipping.")
            if current_end == end_dt: # If current window already reached overall end_dt
                break
            current_start = current_end # Advance to the end of the current (insufficient) period
            continue

        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        cerebro.addstrategy(strategy, **strategy_params)
        cerebro.adddata(feed)
        cerebro.broker.setcash(100000) # Initial cash for each window
        cerebro.broker.setcommission(commission=0.001)
        cerebro.addsizer(bt.sizers.PercentSizer, percents=95)

        start_val = cerebro.broker.getvalue()
        cerebro.run()
        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}")
        
        # Advance to the next window. If current_end already reached overall end_dt, then break.
        if current_end == end_dt:
            break
        current_start = current_end # For non-overlapping windows, next start is current end

    return pd.DataFrame(all_results)

Explanation of run_rolling_backtest:

4. Reporting and Visualization

The included functions report_stats and plot_four_charts are standard and effective for summarizing and visualizing the rolling backtest results.

# ... (rest of the report_stats and plot_four_charts functions) ...

def report_stats(df):
    returns = df['return_pct']
    stats = {
        'Mean Return %': np.mean(returns),
        'Median Return %': np.median(returns),
        'Std Dev %': np.std(returns),
        'Min Return %': np.min(returns),
        'Max Return %': np.max(returns),
        'Sharpe Ratio': np.mean(returns) / np.std(returns) if np.std(returns) > 0 else np.nan
    }
    print("\n=== ROLLING BACKTEST STATISTICS ===")
    for k, v in stats.items():
        print(f"{k}: {v:.2f}")
    return stats

def plot_four_charts(df, rolling_sharpe_window=4):
    """
    Generates four analytical plots for rolling backtest results.
    """
    fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(12, 8)) # Adjusted figsize for clarity
    
    periods = list(range(len(df)))
    returns = df['return_pct']
    
    # 1. Period Returns (Top Left)
    colors = ['green' if r >= 0 else 'red' for r in returns]
    ax1.bar(periods, returns, color=colors, alpha=0.7)
    ax1.set_title('Period Returns', fontsize=14, fontweight='bold')
    ax1.set_xlabel('Period')
    ax1.set_ylabel('Return %')
    ax1.axhline(y=0, color='black', linestyle='-', alpha=0.3)
    ax1.grid(True, alpha=0.3)
    
    # 2. Cumulative Returns (Top Right)
    cumulative_returns = (1 + returns / 100).cumprod() * 100 - 100
    ax2.plot(periods, cumulative_returns, marker='o', linewidth=2, markersize=4, color='blue')
    ax2.set_title('Cumulative Returns', fontsize=14, fontweight='bold')
    ax2.set_xlabel('Period')
    ax2.set_ylabel('Cumulative Return %')
    ax2.grid(True, alpha=0.3)
    
    # 3. Rolling Sharpe Ratio (Bottom Left)
    rolling_sharpe = returns.rolling(window=rolling_sharpe_window).apply(
        lambda x: x.mean() / x.std() if x.std() > 0 else np.nan, raw=False
    )
    valid_mask = ~rolling_sharpe.isna()
    valid_periods = [i for i, valid in enumerate(valid_mask) if valid]
    valid_sharpe = rolling_sharpe[valid_mask]
    
    ax3.plot(valid_periods, valid_sharpe, marker='o', linewidth=2, markersize=4, color='orange')
    ax3.axhline(y=0, color='red', linestyle='--', alpha=0.5)
    ax3.set_title(f'Rolling Sharpe Ratio ({rolling_sharpe_window}-period)', fontsize=14, fontweight='bold')
    ax3.set_xlabel('Period')
    ax3.set_ylabel('Sharpe Ratio')
    ax3.grid(True, alpha=0.3)
    
    # 4. Return Distribution (Bottom Right)
    bins = min(15, max(5, len(returns)//2))
    ax4.hist(returns, bins=bins, alpha=0.7, color='steelblue', edgecolor='black')
    mean_return = returns.mean()
    ax4.axvline(mean_return, color='red', linestyle='--', linewidth=2, 
                label=f'Mean: {mean_return:.2f}%')
    ax4.set_title('Return Distribution', fontsize=14, fontweight='bold')
    ax4.set_xlabel('Return %')
    ax4.set_ylabel('Frequency')
    ax4.legend()
    ax4.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()

if __name__ == '__main__':
    # Use current date for the end of the backtest for a more "live" simulation
    current_date = datetime.datetime.now().date() 
    
    df = run_rolling_backtest(
        ticker="ADA-USD",
        start="2018-01-01",
        end=current_date, # Use the current date
        window_months=3,
      
    )

    print("\n=== ROLLING BACKTEST RESULTS ===")
    print(df)

    stats = report_stats(df)
    plot_four_charts(df)
Pasted image 20250622010146.png

5. Conclusion

The EmaMacdAdxStrategy offers a robust framework for systematic trend following by combining the strengths of EMA crossovers for trend direction, MACD for momentum confirmation, and ADX for trend strength validation. The integration of StopTrail orders provides a dynamic and effective mechanism for managing risk and protecting profits as trades evolve. The rolling backtesting approach is crucial for demonstrating the strategy’s consistency and resilience across various market cycles, offering a more reliable assessment of its performance than a single historical backtest. Further optimization of indicator periods and thresholds, potentially through walk-forward analysis, could enhance the strategy’s adaptability and profitability in live trading environments.