Article

Building an Adaptive Moving Average Crossover Strategy with Backtrader

Moving Averages (MAs) are fundamental tools in technical analysis, used to smooth price data and identify trends. However, traditional MAs use fixed periods, making them less responsive to changing market conditions. In fast-moving, volatile markets, a short MA might be ideal, while in calm, trending markets, a longer MA could reduce false signals.

This tutorial will guide you through building an Adaptive Moving Average (AMA) Crossover Strategy using the backtrader framework in Python. This strategy dynamically adjusts the periods of its moving averages based on market volatility, aiming to improve responsiveness and reduce whipsaws. We’ll also incorporate essential risk management with a stop-loss mechanism.

Why Adaptive Moving Averages?

A common challenge with fixed-period MAs is their “one-size-fits-all” nature.

Short MAs (e.g., 10-period SMA): Highly responsive to price changes, good for capturing short-term trends, but prone to frequent false signals (whipsaws) in choppy markets.
Long MAs (e.g., 50-period SMA): Smoother, better for identifying long-term trends, but lag price action significantly, potentially delaying entry/exit and missing profitable moves.

Adaptive MAs attempt to bridge this gap by adjusting their sensitivity. When volatility is high, the MA period shortens to become more reactive. When volatility is low, the MA period lengthens to become smoother and less prone to noise. This allows the strategy to theoretically adapt its behavior to the prevailing market environment.

Strategy Concept

Our Adaptive MA Crossover strategy will operate as follows:

Volatility Measurement: We will use the Standard Deviation of daily percentage price changes to quantify market volatility.
Adaptive Period Calculation: The period of each moving average (fast and slow) will be adjusted based on the current volatility relative to its recent historical average. If current volatility is high, the period will shorten. If low, it will lengthen.
Moving Average Crossover: Just like a traditional MA crossover, we will generate buy signals when the “fast” adaptive MA crosses above the “slow” adaptive MA, and sell signals when the fast MA crosses below the slow MA.
Risk Management: A fixed percentage stop-loss will be implemented for all positions to limit potential losses.

Prerequisites

Before we begin, ensure you have the necessary libraries installed:

pip install backtrader yfinance pandas numpy matplotlib

Step-by-Step Implementation

We’ll break down the implementation into several logical components.

1. Data Acquisition and Preparation

First, we need historical price data. We’ll use yfinance to download data for Apple (AAPL) from 2020 to 2024.

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# Set matplotlib style for better visualization
%matplotlib inline
plt.rcParams['figure.figsize'] = (10, 6)

# Download data using yfinance
# The droplevel(1) is crucial for backtrader to recognize the columns correctly
print("Downloading AAPL data...")
data = yf.download('AAPL', '2020-01-01', '2024-01-01', auto_adjust=False)
data.columns = data.columns.droplevel(1)
print("Data downloaded successfully.")
print(data.head())

# Create a backtrader data feed
data_feed = bt.feeds.PandasData(dataname=data)

Explanation:

yfinance.download: Downloads historical stock data.
auto_adjust=False: Ensures that “Open”, “High”, “Low”, “Close” are raw prices, not adjusted for splits/dividends (backtrader can handle this internally if needed, but for simplicity here we keep it raw).
data.columns = data.columns.droplevel(1): yfinance often returns a multi-level column index. backtrader expects a single-level index (Open, High, Low, Close, Volume). This line fixes that.
bt.feeds.PandasData: Feeds our pandas DataFrame into backtrader.

2. The Adaptive SMA Indicator (`AdaptiveSMA`)

This is the core component that makes our moving averages adaptive. We’ll create a custom bt.Indicator class.

class AdaptiveSMA(bt.Indicator):
    # Define the output lines for the indicator
    lines = ('sma',)

    # Define tunable parameters for the indicator
    params = (
        ('base_period', 30),  # Base period for the SMA in normal volatility
        ('vol_period', 7),    # Period for calculating recent average volatility
        ('vol_factor', 1.),   # Factor to control the responsiveness to volatility changes
        ('min_period', 7),    # Minimum allowed period for the SMA
        ('max_period', 90),   # Maximum allowed period for the SMA
    )
    
    def __init__(self):
        # Calculate volatility using the standard deviation of percentage price changes
        self.volatility = bt.indicators.StandardDeviation(
            bt.indicators.PctChange(self.data),  # PctChange calculates daily returns
            period=self.params.vol_period
        )
        # Initialize a list to store historical volatility values
        self.vol_history = []
        
    def next(self):
        # Store the current volatility value if it's not NaN
        if not np.isnan(self.volatility[0]):
            self.vol_history.append(self.volatility[0])
        
        # Determine the adaptive period
        if len(self.vol_history) >= self.params.vol_period:
            current_vol = self.volatility[0]
            avg_vol = np.mean(self.vol_history[-self.params.vol_period:])
            
            # Avoid division by zero and handle NaN current_vol
            if avg_vol > 0 and not np.isnan(current_vol):
                vol_ratio = current_vol / avg_vol
                # Calculate the new period: higher vol_ratio -> shorter period
                period = int(self.params.base_period / (1 + (vol_ratio - 1) * self.params.vol_factor))
                # Clamp the period within defined min and max bounds
                period = max(self.params.min_period, min(self.params.max_period, period))
            else:
                # If avg_vol is zero or current_vol is NaN, revert to base period
                period = self.params.base_period
        else:
            # Not enough data for volatility calculation, use base period
            period = self.params.base_period
        
        # Manually calculate the Simple Moving Average for the adaptive period
        # We need enough data points for the calculated period
        if len(self.data) >= period:
            # Sum the 'period' most recent data points and divide by 'period'
            sma_value = sum(self.data[-(i+1)] for i in range(period)) / period
            self.lines.sma[0] = sma_value # Set the current SMA value
        else:
            # If not enough data for the current adaptive period, set to current close
            self.lines.sma[0] = self.data[0]

Explanation of AdaptiveSMA:

lines = ('sma',): Declares that this indicator will output a single line of data, which we name ‘sma’.
params: Defines configurable parameters for the indicator:
- base_period: The default period of the SMA when volatility is “normal”.
- vol_period: The lookback period for calculating average volatility.
- vol_factor: Controls how aggressively the period adapts. A higher factor means more responsiveness to volatility changes.
- min_period, max_period: Ensure the adaptive period stays within reasonable bounds.
__init__(self):
- self.volatility: An instance of bt.indicators.StandardDeviation on bt.indicators.PctChange(self.data). This calculates the standard deviation of daily returns over vol_period, serving as our volatility measure.
- self.vol_history: A list to store the volatility values for calculating the average volatility.
next(self): This method is called for each new bar of data.
- It appends the current volatility to vol_history.
- It calculates current_vol and avg_vol (mean of recent vol_history).
- vol_ratio = current_vol / avg_vol: This ratio tells us if the market is more or less volatile than its recent average.
- period = int(self.params.base_period / (1 + (vol_ratio - 1) * self.params.vol_factor)): This is the core adaptive logic.
  - If vol_ratio is > 1 (higher volatility), (vol_ratio - 1) is positive, making 1 + (...) greater than 1, thus shortening the base_period.
  - If vol_ratio is < 1 (lower volatility), (vol_ratio - 1) is negative, making 1 + (...) less than 1, thus lengthening the base_period.
- period = max(self.params.min_period, min(self.params.max_period, period)): Clamps the calculated period between min_period and max_period.
- The SMA is then calculated manually using sum(self.data[-(i+1)] for i in range(period)) / period. This is done because backtrader’s built-in SMA indicator doesn’t allow its period to be dynamically changed after initialization.
- self.lines.sma[0] = sma_value: Assigns the calculated SMA to the indicator’s output line for the current bar.

3. The Adaptive MA Crossover Strategy (`AdaptiveMAStrategy`)

This class defines our trading logic, utilizing the AdaptiveSMA indicator.

class AdaptiveMAStrategy(bt.Strategy):
    # Define strategy parameters
    params = (
        ('fast_base', 7),       # Base period for the fast adaptive MA
        ('slow_base', 30),      # Base period for the slow adaptive MA
        ('vol_period', 7),      # Volatility period (shared for both MAs)
        ('vol_factor', 1.),     # Volatility factor (shared for both MAs)
        ('stop_loss_pct', 0.01), # Percentage for the stop-loss (e.g., 0.01 = 1%)
    )
    
    def __init__(self):
        self.close = self.data.close # Keep a reference to the close price line
        
        # Create instances of our custom AdaptiveSMA indicator for fast and slow MAs
        self.fast_ma = AdaptiveSMA(
            self.data,
            base_period=self.params.fast_base,
            vol_period=self.params.vol_period,
            vol_factor=self.params.vol_factor,
            min_period=5,
            max_period=25
        )
        
        self.slow_ma = AdaptiveSMA(
            self.data,
            base_period=self.params.slow_base,
            vol_period=self.params.vol_period,
            vol_factor=self.params.vol_factor,
            min_period=15,
            max_period=60
        )
        
        # Create a CrossOver indicator to detect when fast_ma crosses slow_ma
        self.crossover = bt.indicators.CrossOver(self.fast_ma, self.slow_ma)
        
        # Initialize variables to track orders and stop-loss orders
        self.order = None       # To hold an active order
        self.stop_order = None  # To hold an active stop-loss order

    def notify_order(self, order):
        # This method is called when the status of an order changes
        if order.status in [order.Completed]:
            if order.isbuy() and self.position.size > 0:
                # If a buy order is completed and we have a long position
                # Set a stop-loss order below the entry price
                stop_price = order.executed.price * (1 - self.params.stop_loss_pct)
                self.stop_order = self.sell(exectype=bt.Order.Stop, price=stop_price)
                self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, Stop Loss set at: {stop_price:.2f}')
            elif order.issell() and self.position.size < 0:
                # If a sell order (for shorting) is completed and we have a short position
                # Set a stop-loss order above the entry price
                stop_price = order.executed.price * (1 + self.params.stop_loss_pct)
                self.stop_order = self.buy(exectype=bt.Order.Stop, price=stop_price)
                self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, Stop Loss set at: {stop_price:.2f}')
        
        # If an order is completed, canceled, or rejected, clear the order reference
        if order.status in [order.Completed, order.Canceled, order.Rejected]:
            self.order = None
            # If the completed order was the stop-loss order, clear its reference too
            if order == self.stop_order:
                self.stop_order = None

    def log(self, txt, dt=None):
        ''' Logging function for the strategy '''
        dt = dt or self.datas[0].datetime.date(0)
        print(f'{dt.isoformat()}, {txt}')

    def next(self):
        # Skip if there's an active order (wait for it to complete)
        if self.order is not None:
            return
        
        # Check for crossover signals
        if self.crossover > 0:  # Fast MA crosses above Slow MA (Bullish signal)
            if self.position.size < 0:  # If currently short, close short position first
                self.log(f'CLOSING SHORT POSITION, Price: {self.close[0]:.2f}')
                if self.stop_order is not None: # Cancel existing stop-loss if any
                    self.cancel(self.stop_order)
                self.order = self.close() # Close the short position
            elif not self.position:  # If not in a position, open a long position
                self.log(f'OPENING LONG POSITION, Price: {self.close[0]:.2f}')
                self.order = self.buy() # Execute a buy order
                
        elif self.crossover < 0:  # Fast MA crosses below Slow MA (Bearish signal)
            if self.position.size > 0:  # If currently long, close long position first
                self.log(f'CLOSING LONG POSITION, Price: {self.close[0]:.2f}')
                if self.stop_order is not None: # Cancel existing stop-loss if any
                    self.cancel(self.stop_order)
                self.order = self.close() # Close the long position
            elif not self.position:  # If not in a position, open a short position
                self.log(f'OPENING SHORT POSITION, Price: {self.close[0]:.2f}')
                self.order = self.sell() # Execute a sell order

Explanation of AdaptiveMAStrategy:

params: Defines strategy-specific parameters like fast_base, slow_base, vol_period, vol_factor (passed to AdaptiveSMA), and stop_loss_pct.
__init__(self):
- Initializes fast_ma and slow_ma as instances of our AdaptiveSMA indicator, passing self.data and the respective parameters.
- self.crossover = bt.indicators.CrossOver(self.fast_ma, self.slow_ma): This built-in backtrader indicator detects when one line crosses another. A positive value means fast_ma crossed above slow_ma, a negative value means it crossed below.
- self.order and self.stop_order: Used to keep track of active orders and prevent sending new orders while one is pending.
notify_order(self, order):
- This is a crucial callback method in backtrader that gets triggered whenever an order’s status changes.
- If an order is Completed (bought or sold), it checks the isbuy() or issell() status.
- For a completed buy (long position), it calculates a stop_price at (1 - stop_loss_pct) below the executed price and places a sell(exectype=bt.Order.Stop, price=stop_price) order.
- For a completed sell (short position), it calculates a stop_price at (1 + stop_loss_pct) above the executed price and places a buy(exectype=bt.Order.Stop, price=stop_price) order.
- It clears self.order and self.stop_order references once orders are completed, canceled, or rejected, allowing new orders to be placed.
log(self, txt, dt=None): A simple helper function to print logs with dates.
next(self): This is the main trading logic loop, executed for each new bar.
- if self.order is not None: return: Prevents placing new orders if one is already active.
- Crossover Logic:
  - if self.crossover > 0: Fast MA crosses above slow MA (bullish).
    - If already short (self.position.size < 0), it closes the short position first (self.close()) and cancels any active stop-loss.
    - If not in any position (not self.position), it opens a long position (self.buy()).
  - elif self.crossover < 0: Fast MA crosses below slow MA (bearish).
    - If already long (self.position.size > 0), it closes the long position first (self.close()) and cancels any active stop-loss.
    - If not in any position (not self.position), it opens a short position (self.sell()).

4. Backtesting Setup and Execution

Finally, we set up the backtrader Cerebro engine, add our strategy and data, configure the broker, and run the backtest. We’ll also add some analyzers for performance insights.

# Create a Cerebro entity
cerebro = bt.Cerebro()

# Add the strategy
cerebro.addstrategy(AdaptiveMAStrategy)

# Add the data feed
cerebro.adddata(data_feed)

# Set the sizer: invest 95% of available cash
cerebro.addsizer(bt.sizers.PercentSizer, percents=95)

# Set starting cash
cerebro.broker.setcash(100000.0)

# Set commission to 0.1%
cerebro.broker.setcommission(commission=0.001) # 0.1% commission

# Add analyzers for performance evaluation
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='tradeanalyzer')
cerebro.addanalyzer(bt.analyzers.SQN, _name='sqn') # System Quality Number

# Print starting portfolio value
print(f'Starting Portfolio Value: ${cerebro.broker.getvalue():,.2f}')

# Run the backtest
print("Running backtest...")
results = cerebro.run()
print("Backtest finished.")

# Print final portfolio value
final_value = cerebro.broker.getvalue()
print(f'Final Portfolio Value: ${final_value:,.2f}')

# Get and print analysis results
strat = results[0]

print("\n--- Strategy Performance Metrics ---")
# Returns Analyzer
returns_analysis = strat.analyzers.returns.get_analysis()
total_return = returns_analysis.get('rtot', 'N/A') * 100
annual_return = returns_analysis.get('rnorm100', 'N/A')
print(f"Total Return: {total_return:.2f}%")
print(f"Annualized Return: {annual_return:.2f}%")

# Sharpe Ratio
sharpe_ratio = strat.analyzers.sharpe.get_analysis()
print(f"Sharpe Ratio: {sharpe_ratio.get('sharperatio', 'N/A'):.2f}")

# DrawDown Analyzer
drawdown_analysis = strat.analyzers.drawdown.get_analysis()
max_drawdown = drawdown_analysis.get('maxdrawdown', 'N/A')
print(f"Max Drawdown: {max_drawdown:.2f}%")
print(f"Longest Drawdown Duration: {drawdown_analysis.get('maxdrawdownperiod', 'N/A')} bars")

# Trade Analyzer
trade_analysis = strat.analyzers.tradeanalyzer.get_analysis()
total_trades = trade_analysis.get('total', {}).get('total', 0)
won_trades = trade_analysis.get('won', {}).get('total', 0)
lost_trades = trade_analysis.get('lost', {}).get('total', 0)
win_rate = (won_trades / total_trades) * 100 if total_trades > 0 else 0
print(f"Total Trades: {total_trades}")
print(f"Winning Trades: {won_trades} ({win_rate:.2f}%)")
print(f"Losing Trades: {lost_trades} ({100-win_rate:.2f}%)")
print(f"Avg Win: {trade_analysis.get('won',{}).get('pnl',{}).get('average', 'N/A'):.2f}")
print(f"Avg Loss: {trade_analysis.get('lost',{}).get('pnl',{}).get('average', 'N/A'):.2f}")

# SQN (System Quality Number)
sqn_analysis = strat.analyzers.sqn.get_analysis()
print(f"System Quality Number (SQN): {sqn_analysis.get('sqn', 'N/A'):.2f}")


# Plot the results
print("\nPlotting results...")
cerebro.plot(iplot=False, style='candlestick',
             barup=dict(fill=False, lw=1.0, ls='-', color='green'),
             bardown=dict(fill=False, lw=1.0, ls='-', color='red'),
             plotreturn=True # Plot equity curve
            )
print("Plot generated.")

Explanation of Backtesting Setup:

bt.Cerebro(): The main engine for running backtests.
cerebro.addstrategy(AdaptiveMAStrategy): Adds our defined strategy to the engine.
cerebro.adddata(data_feed): Feeds the prepared historical data to the engine.
cerebro.addsizer(bt.sizers.PercentSizer, percents=95): Tells the broker to invest 95% of available cash on each trade.
cerebro.broker.setcash(100000.0): Sets the initial capital for the backtest.
cerebro.broker.setcommission(commission=0.001): Sets a commission of 0.1% per transaction. This is crucial for realistic backtesting.
cerebro.addanalyzer(...): Adds various backtrader analyzers to compute key performance metrics (Sharpe Ratio, Drawdown, Returns, Trade Statistics, SQN).
cerebro.run(): Executes the backtest.
Printing Results: After the backtest, we access the results from the analyzers and print them to the console, providing a summary of the strategy’s performance.
cerebro.plot(iplot=False): Generates a visual plot of the backtest, showing price action, indicators, and trade entries/exits. iplot=False ensures it generates a static plot (useful if not in a Jupyter environment with interactive plotting). style='candlestick' gives candlestick charts. plotreturn=True adds the equity curve.

Further Enhancements and Considerations

While this is a robust adaptive MA strategy, here are ideas for further improvement:

Parameter Optimization: The base_period, vol_period, vol_factor, min_period, max_period, and stop_loss_pct are fixed. Use backtrader’s optimization capabilities (cerebro.optstrategy) to find the best parameter sets for different assets or market conditions.
Alternative Volatility Measures: Experiment with other volatility indicators like Average True Range (ATR) or Keltner Channels to drive the adaptive periods.
Adaptive Stop-Loss: Instead of a fixed percentage, make the stop-loss adaptive (e.g., based on ATR or a multiple of the adaptive MA’s distance from price).
Take-Profit Mechanism: Add a take-profit target to lock in gains and prevent giving back profits.
Market Regime Filters: Implement filters to only trade when certain market conditions are met (e.g., only trade in trending markets, avoid sideways markets). This could involve higher timeframe MAs or other trend indicators.
Slippage Simulation: For more realistic backtesting, especially with frequent trades, add slippage to your broker settings: cerebro.broker.set_slippage_fixed(0.01) (for 1 cent slippage per share).
Data Resampling: Test the strategy on different timeframes (e.g., weekly, hourly data) to see its performance characteristics.
Position Sizing: Explore more advanced position sizing techniques beyond a fixed percentage, such as fixed risk per trade or Kelly Criterion.

Conclusion

You’ve successfully built and backtested an Adaptive Moving Average Crossover strategy in backtrader. By dynamically adjusting to market volatility, this strategy aims to be more resilient and effective than traditional fixed-period MA strategies. This comprehensive tutorial provides a strong foundation for further experimentation and development of more sophisticated trading systems. Remember, successful algorithmic trading requires continuous iteration, rigorous testing, and a deep understanding of market dynamics.