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,
=self.params.macd_fast,
period_me1=self.params.macd_slow,
period_me2=self.params.macd_signal
period_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:
= self.atr[0] / self.dataclose[0]
normalized_atr 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(
=bt.Order.StopTrail,
exectype=self.atr[0] * self.params.atr_multiplier)
trailamountelif self.position.size < 0:
self.trail_order = self.buy(
=bt.Order.StopTrail,
exectype=self.atr[0] * self.params.atr_multiplier)
trailamountreturn
# Ensure sufficient data
= max(self.params.macd_slow, self.params.adx_period, self.params.atr_period)
required_bars if len(self) < required_bars:
return
# MACD crossover signals
= self.macd.macd[0]
macd_line = self.macd.signal[0]
macd_signal = self.macd.macd[-1]
macd_prev = self.macd.signal[-1]
signal_prev = (macd_line > macd_signal and macd_prev <= signal_prev)
macd_bullish_cross = (macd_line < macd_signal and macd_prev >= signal_prev)
macd_bearish_cross
# ADX trend strength filter
= self.adx[0] > self.params.adx_threshold
adx_strong = self.plusdi[0] > self.minusdi[0]
trend_bullish = self.minusdi[0] > self.plusdi[0]
trend_bearish
# Position sizing based on volatility
= self.calculate_volatility_position_size()
position_size_pct = self.dataclose[0]
current_price
# 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()
= self.broker.getcash()
cash = cash * position_size_pct
target_value = target_value / current_price
shares 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()
= self.broker.getcash()
cash = cash * position_size_pct
target_value = target_value / current_price
shares 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:
= self.broker.getcash()
cash = cash * (position_size_pct * 0.7)
target_value = target_value / current_price
shares self.order = self.buy(size=shares)
elif macd_line < macd_signal and trend_bearish and macd_line < 0:
= self.broker.getcash()
cash = cash * (position_size_pct * 0.7)
target_value = target_value / current_price
shares 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(
="ETH-USD",
ticker="2020-01-01",
start="2025-01-01",
end=12,
window_months=None
strategy_params
):= strategy_params or {}
strategy_params = []
all_results = pd.to_datetime(start)
start_dt = pd.to_datetime(end)
end_dt = start_dt
current_start
while True:
= current_start + rd.relativedelta(months=window_months)
current_end if current_end > end_dt:
break
print(f"\nROLLING BACKTEST: {current_start.date()} to {current_end.date()}")
= yf.download(ticker, start=current_start, end=current_end, progress=False)
data if data.empty or len(data) < 90:
print("Not enough data.")
+= rd.relativedelta(months=window_months)
current_start continue
= data.droplevel(1, 1) if isinstance(data.columns, pd.MultiIndex) else data
data
= bt.feeds.PandasData(dataname=data)
feed = bt.Cerebro()
cerebro **strategy_params)
cerebro.addstrategy(strategy,
cerebro.adddata(feed)100000)
cerebro.broker.setcash(=0.001)
cerebro.broker.setcommission(commission=95)
cerebro.addsizer(bt.sizers.PercentSizer, percents
= cerebro.broker.getvalue()
start_val
cerebro.run()= cerebro.broker.getvalue()
final_val = (final_val - start_val) / start_val * 100
ret
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}")
+= rd.relativedelta(months=window_months)
current_start
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.