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.
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:
fast_period
and a slow_period
) are used to
generate the primary trend direction signal.
Entry Logic:
adx_threshold
(Trend strength
confirmation)adx_threshold
(Trend strength
confirmation)Exit Logic:
The strategy utilizes a two-pronged approach for exits:
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.EmaMacdAdxStrategy
ImplementationHere’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],
=self.params.macd_fast_period,
period_me1=self.params.macd_slow_period,
period_me2=self.params.macd_signal_period)
period_signal
# 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 or self.datas[0].datetime.date(0)
dt 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,
=self.params.trail_percent,
trailpercent=self.position.size) # Specify size for stop-loss
size
elif order.issell(): # --- Sell Order Completed ---
# Check if this sell order was the execution of our trailing stop (by checking its reference)
= (self.stop_order is not None and
is_stop_trail_execution == self.stop_order.ref)
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,
=self.params.trail_percent,
trailpercent=abs(self.position.size)) # Specify size for stop-loss
sizeelse: # 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
= max(self.p.ema_slow_period, self.p.macd_slow_period, self.p.adx_period) + 1
min_warmup_period 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 ---
= self.ema_fast[0] > self.ema_slow[0] # EMA crossover
long_condition_1 = self.macd.macd[0] > self.macd.signal[0] # MACD bullish
long_condition_2 = self.adx.adx[0] > self.params.adx_threshold # ADX strong trend
long_condition_3
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
= self.ema_fast[0] < self.ema_slow[0] # EMA crossover
short_condition_1 = self.macd.macd[0] < self.macd.signal[0] # MACD bearish
short_condition_2 = self.adx.adx[0] > self.params.adx_threshold # ADX strong trend
short_condition_3
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
= self.ema_fast[0] < self.ema_slow[0] # EMA crossover reversal
long_exit_condition_1 = self.macd.macd[0] < self.macd.signal[0] # MACD bearish reversal
long_exit_condition_2
# 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
= self.ema_fast[0] > self.ema_slow[0] # EMA crossover reversal
short_exit_condition_1 = self.macd.macd[0] > self.macd.signal[0] # MACD bullish reversal
short_exit_condition_2
# 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
:
params
: Defines various configurable
parameters for the EMA periods, MACD periods, ADX period and threshold,
and the trail_percent
for the trailing stop-loss.__init__(self)
:
self.dataclose
(the closing
price series).self.order
and self.stop_order
to track pending orders. self.buyprice
and
self.buycomm
are used for logging purposes.bt.indicators.ExponentialMovingAverage
(fast and slow),
bt.indicators.MACDHisto
(for MACD Line, Signal Line, and
Histogram), and
bt.indicators.AverageDirectionalMovementIndex
(for
ADX).log(self, txt, dt=None, doprint=False)
:
A utility function for consistent logging of strategy actions.notify_order(self, order)
: This is a
crucial method that backtrader
calls whenever an order’s
status changes.
Completed
buy or sell
(short) entry orders.bt.Order.StopTrail
order with the specified trail_percent
. This is a
trailing stop that moves to protect profits.Canceled
, Margin
, or
Rejected
orders, clearing the relevant order references to
avoid erroneous state.notify_trade(self, trade)
: This method
is called when a trade is fully closed (e.g., position goes from open to
zero). It logs the gross and net profit/loss and performs any necessary
state resets after a trade cycle completes.next(self)
: This method contains the
core trading logic and is executed on each new bar (e.g., daily).
self.order
to prevent placing multiple entry/primary exit
orders if one is already in progress.if not self.position
):
self.buy()
order
is placed.self.sell()
(short) order is placed.else:
, meaning
self.position
is active):
StopTrail
order.self.close()
method is called
to exit the long position.self.close()
method is called
to exit the short position.self.close()
automatically cancels any existing
StopTrail
order associated with that position.stop(self)
: A method called at the
very end of the backtest to log the final portfolio value.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
= EmaMacdAdxStrategy
strategy
def run_rolling_backtest(
="ADA-USD",
ticker="2018-01-01",
start# Updated end date to the current date for a more live test
="2025-06-22",
end=3,
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 # Adjust end of current window if it exceeds overall end date
if current_end > end_dt:
= end_dt
current_end 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
= yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)
data
# Apply droplevel if data is a MultiIndex, as per user's preference
if isinstance(data.columns, pd.MultiIndex):
= data.droplevel(1, axis=1)
data
# Check for sufficient data after droplevel for strategy warm-up
# Calculate min bars needed based on strategy parameters
= strategy_params.get('ema_slow_period', EmaMacdAdxStrategy.params.ema_slow_period)
ema_slow_period = strategy_params.get('macd_slow_period', EmaMacdAdxStrategy.params.macd_slow_period)
macd_slow_period = strategy_params.get('adx_period', EmaMacdAdxStrategy.params.adx_period)
adx_period = max(ema_slow_period, macd_slow_period, adx_period) + 1 # +1 for current bar's data
min_bars_needed
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_end # Advance to the end of the current (insufficient) period
current_start continue
= bt.feeds.PandasData(dataname=data)
feed = bt.Cerebro()
cerebro **strategy_params)
cerebro.addstrategy(strategy,
cerebro.adddata(feed)100000) # Initial cash for each window
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}")
# Advance to the next window. If current_end already reached overall end_dt, then break.
if current_end == end_dt:
break
= current_end # For non-overlapping windows, next start is current end
current_start
return pd.DataFrame(all_results)
Explanation of
run_rolling_backtest
:
ticker
, overall
start
and end
dates,
window_months
for the length of each sub-period, and
strategy_params
to pass custom parameters to the
EmaMacdAdxStrategy
.while
loop
iterates through the defined overall time range, creating sequential,
non-overlapping monthly windows. It handles the last window gracefully
to not exceed the overall end
date.yf.download
. It rigorously adheres to
your saved preference of auto_adjust=False
and
droplevel(axis=1, level=1)
if a MultiIndex is present,
ensuring data consistency.min_bars_needed
based on the longest period of the
strategy’s indicators (EMA slow, MACD slow, ADX) and checks if the
downloaded data has enough bars for these indicators to warm up and
provide valid values within the current window. Periods with
insufficient data are skipped.bt.Cerebro
instance is created. This ensures each backtest
run is independent. The EmaMacdAdxStrategy
is added, along
with the data feed, initial cash, commission, and a sizer (to allocate
95% of capital).cerebro.run()
executes the backtest for the current segment. The start/end dates,
percentage return, and final portfolio value for that window are
recorded.current_start
is
updated to current_end
for the next iteration, ensuring
non-overlapping backtest periods.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):
= df['return_pct']
returns = {
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.
"""
= plt.subplots(2, 2, figsize=(12, 8)) # Adjusted figsize for clarity
fig, ((ax1, ax2), (ax3, ax4))
= list(range(len(df)))
periods = df['return_pct']
returns
# 1. Period Returns (Top Left)
= ['green' if r >= 0 else 'red' for r in returns]
colors =colors, alpha=0.7)
ax1.bar(periods, returns, color'Period Returns', fontsize=14, fontweight='bold')
ax1.set_title('Period')
ax1.set_xlabel('Return %')
ax1.set_ylabel(=0, color='black', linestyle='-', alpha=0.3)
ax1.axhline(yTrue, alpha=0.3)
ax1.grid(
# 2. Cumulative Returns (Top Right)
= (1 + returns / 100).cumprod() * 100 - 100
cumulative_returns ='o', linewidth=2, markersize=4, color='blue')
ax2.plot(periods, cumulative_returns, marker'Cumulative Returns', fontsize=14, fontweight='bold')
ax2.set_title('Period')
ax2.set_xlabel('Cumulative Return %')
ax2.set_ylabel(True, alpha=0.3)
ax2.grid(
# 3. Rolling Sharpe Ratio (Bottom Left)
= returns.rolling(window=rolling_sharpe_window).apply(
rolling_sharpe lambda x: x.mean() / x.std() if x.std() > 0 else np.nan, raw=False
)= ~rolling_sharpe.isna()
valid_mask = [i for i, valid in enumerate(valid_mask) if valid]
valid_periods = rolling_sharpe[valid_mask]
valid_sharpe
='o', linewidth=2, markersize=4, color='orange')
ax3.plot(valid_periods, valid_sharpe, marker=0, color='red', linestyle='--', alpha=0.5)
ax3.axhline(yf'Rolling Sharpe Ratio ({rolling_sharpe_window}-period)', fontsize=14, fontweight='bold')
ax3.set_title('Period')
ax3.set_xlabel('Sharpe Ratio')
ax3.set_ylabel(True, alpha=0.3)
ax3.grid(
# 4. Return Distribution (Bottom Right)
= min(15, max(5, len(returns)//2))
bins =bins, alpha=0.7, color='steelblue', edgecolor='black')
ax4.hist(returns, bins= returns.mean()
mean_return ='red', linestyle='--', linewidth=2,
ax4.axvline(mean_return, color=f'Mean: {mean_return:.2f}%')
label'Return Distribution', fontsize=14, fontweight='bold')
ax4.set_title('Return %')
ax4.set_xlabel('Frequency')
ax4.set_ylabel(
ax4.legend()True, alpha=0.3)
ax4.grid(
plt.tight_layout()
plt.show()
if __name__ == '__main__':
# Use current date for the end of the backtest for a more "live" simulation
= datetime.datetime.now().date()
current_date
= run_rolling_backtest(
df ="ADA-USD",
ticker="2018-01-01",
start=current_date, # Use the current date
end=3,
window_months
)
print("\n=== ROLLING BACKTEST RESULTS ===")
print(df)
= report_stats(df)
stats plot_four_charts(df)
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.