The MACD Momentum Strategy is a trend-following trading system designed to capitalize on strong directional moves in financial markets, particularly suited for volatile assets like cryptocurrencies. By integrating the Moving Average Convergence Divergence (MACD) indicator with the Average Directional Index (ADX) and volatility-based position sizing, this strategy aims to identify robust trends while managing risk effectively. This article outlines the strategy’s logic, reasoning, and implementation, highlighting the key code components and their functionality.
The strategy combines three primary components:
The strategy employs trailing stops to lock in profits or limit losses and dynamically adjusts position sizes based on recent market volatility.
The strategy enters trades when the following conditions align:
Position sizes are calculated dynamically based on recent volatility, measured by the ATR normalized by price. Higher volatility results in smaller positions to limit risk, while lower volatility allows larger positions. The position size scales between a minimum (30%) and maximum (95%) of available capital.
Trailing stops are set using the ATR multiplied by a factor (e.g., 2x ATR) to exit positions when the price reverses significantly, balancing trend-following with profit protection.
Below are the main components of the
MACDMomentumStrategy class, focusing on the parameters and
the next function, which implements the trading logic.
import backtrader as bt
import numpy as np
class MACDMomentumStrategy(bt.Strategy):
params = (
('macd_fast', 12), # MACD fast period
('macd_slow', 26), # MACD slow period
('macd_signal', 9), # MACD signal period
('adx_period', 14), # ADX period
('adx_threshold', 25), # ADX threshold for strong trend
('atr_period', 14), # ATR period
('atr_multiplier', 2.0), # ATR multiplier for trailing stops
('vol_lookback', 20), # Volatility lookback for position sizing
('max_position_pct', 0.95), # Maximum position size
('min_position_pct', 0.30), # Minimum position size
)
def __init__(self):
self.dataclose = self.datas[0].close
# MACD indicator
self.macd = bt.indicators.MACD(
self.dataclose,
period_me1=self.params.macd_fast,
period_me2=self.params.macd_slow,
period_signal=self.params.macd_signal
)
# ADX for trend strength
self.adx = bt.indicators.ADX(period=self.params.adx_period)
self.plusdi = bt.indicators.PlusDI(period=self.params.adx_period)
self.minusdi = bt.indicators.MinusDI(period=self.params.adx_period)
# ATR for volatility and stops
self.atr = bt.indicators.ATR(period=self.params.atr_period)
# Track orders
self.order = None
self.trail_order = None
# Volatility tracking for position sizing
self.volatility_history = []
def next(self):
# Skip if order is pending
if self.order:
return
# Update volatility history
if len(self.atr) > 0 and self.dataclose[0] > 0:
normalized_atr = self.atr[0] / self.dataclose[0]
self.volatility_history.append(normalized_atr)
if len(self.volatility_history) > self.params.vol_lookback:
self.volatility_history = self.volatility_history[-self.params.vol_lookback:]
# Handle trailing stops for existing positions
if self.position:
if not self.trail_order:
if self.position.size > 0:
self.trail_order = self.sell(
exectype=bt.Order.StopTrail,
trailamount=self.atr[0] * self.params.atr_multiplier)
elif self.position.size < 0:
self.trail_order = self.buy(
exectype=bt.Order.StopTrail,
trailamount=self.atr[0] * self.params.atr_multiplier)
return
# Ensure sufficient data
required_bars = max(self.params.macd_slow, self.params.adx_period, self.params.atr_period)
if len(self) < required_bars:
return
# MACD crossover signals
macd_line = self.macd.macd[0]
macd_signal = self.macd.signal[0]
macd_prev = self.macd.macd[-1]
signal_prev = self.macd.signal[-1]
macd_bullish_cross = (macd_line > macd_signal and macd_prev <= signal_prev)
macd_bearish_cross = (macd_line < macd_signal and macd_prev >= signal_prev)
# ADX trend strength filter
adx_strong = self.adx[0] > self.params.adx_threshold
trend_bullish = self.plusdi[0] > self.minusdi[0]
trend_bearish = self.minusdi[0] > self.plusdi[0]
# Position sizing based on volatility
position_size_pct = self.calculate_volatility_position_size()
current_price = self.dataclose[0]
# LONG ENTRY: MACD bullish cross + strong ADX + bullish trend
if macd_bullish_cross and adx_strong and trend_bullish and not self.position:
self.cancel_trail()
cash = self.broker.getcash()
target_value = cash * position_size_pct
shares = target_value / current_price
self.order = self.buy(size=shares)
# SHORT ENTRY: MACD bearish cross + strong ADX + bearish trend
elif macd_bearish_cross and adx_strong and trend_bearish and not self.position:
self.cancel_trail()
cash = self.broker.getcash()
target_value = cash * position_size_pct
shares = target_value / current_price
self.order = self.sell(size=shares)
# Alternative entry: MACD above/below signal with very strong ADX
elif not self.position and self.adx[0] > (self.params.adx_threshold + 10):
if macd_line > macd_signal and trend_bullish and macd_line > 0:
cash = self.broker.getcash()
target_value = cash * (position_size_pct * 0.7)
shares = target_value / current_price
self.order = self.buy(size=shares)
elif macd_line < macd_signal and trend_bearish and macd_line < 0:
cash = self.broker.getcash()
target_value = cash * (position_size_pct * 0.7)
shares = target_value / current_price
self.order = self.sell(size=shares)params tuple defines
configurable settings, including MACD periods (macd_fast,
macd_slow, macd_signal), ADX period and
threshold, ATR period, and position sizing limits. These allow
customization for different markets or timeframes.__init__ method
initializes the MACD, ADX, +DI, -DI, and ATR indicators, along with
variables for order tracking and volatility history.next method is the
core decision-making loop, executed for each data point. It:
Rolling backtests are a good way to analyze if your strategy is consistent over time and for different assets.
def run_rolling_backtest(
ticker="ETH-USD",
start="2020-01-01",
end="2025-01-01",
window_months=12,
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 = yf.download(ticker, start=current_start, end=current_end, progress=False)
if data.empty or len(data) < 90:
print("Not enough data.")
current_start += rd.relativedelta(months=window_months)
continue
data = data.droplevel(1, 1) if isinstance(data.columns, pd.MultiIndex) else data
feed = bt.feeds.PandasData(dataname=data)
cerebro = bt.Cerebro()
cerebro.addstrategy(strategy, **strategy_params)
cerebro.adddata(feed)
cerebro.broker.setcash(100000)
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}")
current_start += rd.relativedelta(months=window_months)
return pd.DataFrame(all_results) ## Conclusion
The MACD Momentum Strategy is a robust trend-following system that combines MACD crossovers, ADX trend strength, and volatility-based risk management to capture significant market moves. Its dynamic position sizing and trailing stops make it suitable for volatile markets like cryptocurrencies. The provided code offers a flexible framework for traders to backtest and optimize this strategy across various assets and timeframes.