The MA Bounce Strategy is a trend-following system
designed to identify and trade pullbacks to a key moving average within
an established trend. It aims to capture continuation moves after a
temporary dip in price. This article details the implementation of this
strategy in backtrader, along with a robust framework for
parameter optimization and a comprehensive rolling backtest analysis,
including statistical reporting and graphical visualizations.
This section presents the core MaBounceStrategy class,
which defines the trading logic.
import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import dateutil.relativedelta as rd
import warnings
warnings.filterwarnings("ignore")
# ------------------------------------------------------------------
# 1. MA Bounce Strategy (from your code with minor fixes)
# ------------------------------------------------------------------
class MaBounceStrategy(bt.Strategy):
params = (
('key_ma_period', 7), # MA for bounce (e.g., 50 SMA)
('filter_ma_period', 30), # Longer MA for trend filter (e.g., 200 SMA)
('ma_type', 'SMA'), # Type of MA ('SMA' or 'EMA')
('order_percentage', 0.95),
('stop_loss_pct', 0.02), # Example: 2% stop loss below entry price
('printlog', False),
)
def log(self, txt, dt=None):
if self.params.printlog:
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()}: {txt}')
def __init__(self):
self.data_close = self.datas[0].close
self.data_low = self.datas[0].low
self.data_high = self.datas[0].high
# Select MA type based on params
ma_indicator = bt.indicators.SMA if self.params.ma_type == 'SMA' else bt.indicators.EMA
# Initialize the key MA and filter MA
self.key_ma = ma_indicator(self.data_close, period=self.params.key_ma_period)
self.filter_ma = ma_indicator(self.data_close, period=self.params.filter_ma_period)
# Order tracking and stop price
self.order = None
self.stop_price = None
# Track strategy statistics
self.trade_count = 0
self.bounce_entries = 0
self.stop_loss_exits = 0
self.profitable_trades = 0
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.log(f'BUY EXECUTED: Price {order.executed.price:.2f}')
# Set stop loss price after buy order executes
self.stop_price = order.executed.price * (1.0 - self.params.stop_loss_pct)
self.trade_count += 1
self.bounce_entries += 1
elif order.issell():
self.log(f'SELL EXECUTED: Price {order.executed.price:.2f}')
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
# Reset order tracking after completion/failure
self.order = None
def notify_trade(self, trade):
if not trade.isclosed:
return
if trade.pnl > 0:
self.profitable_trades += 1
self.log(f'TRADE CLOSED: PnL {trade.pnl:.2f}')
# Reset stop price when trade is closed
self.stop_price = None
def next(self):
# Check if indicators have enough data
if len(self.data_close) < self.params.filter_ma_period:
return
# Check for open orders
if self.order:
return
# --- Check Stop Loss ---
if self.position and self.stop_price is not None:
if self.data_close[0] < self.stop_price:
self.log(f'STOP LOSS: Price {self.data_close[0]:.2f} < Stop {self.stop_price:.2f}')
self.order = self.close() # Close position
self.stop_loss_exits += 1
return # Exit check for this bar
# --- Entry Logic ---
if not self.position:
# 1. Confirm Uptrend State (Price > Filter MA, Key MA > Filter MA)
uptrend_confirmed = (self.data_close[0] > self.filter_ma[0] and
self.key_ma[0] > self.filter_ma[0])
if uptrend_confirmed:
# 2. Check for Pullback: Low price touched or went below the key MA in the previous bar
touched_ma_prev_bar = self.data_low[-1] <= self.key_ma[-1]
# 3. Check for Rejection/Entry Trigger: Price closes back ABOVE the key MA on the current bar
closed_above_ma_curr_bar = self.data_close[0] > self.key_ma[0]
if touched_ma_prev_bar and closed_above_ma_curr_bar:
self.log(f'MA BOUNCE SIGNAL: Price bounced from {self.key_ma[0]:.2f}')
cash = self.broker.get_cash()
size = (cash * self.params.order_percentage) / self.data_close[0]
self.order = self.buy(size=size)params):key_ma_period: The period for the primary moving
average that price is expected to “bounce” from.filter_ma_period: A longer moving average used to
confirm the overall trend direction.ma_type: Specifies whether to use Simple Moving Average
(‘SMA’) or Exponential Moving Average (‘EMA’).order_percentage: The percentage of available cash to
use for each trade.stop_loss_pct: A fixed percentage below the entry price
for the stop loss.printlog: A boolean flag to enable or disable logging
of strategy events.__init__):self.data_close, self.data_low,
self.data_high: References to the close, low, and high
price lines of the data feed.ma_indicator: Dynamically selects between
bt.indicators.SMA and bt.indicators.EMA based
on self.params.ma_type.self.key_ma: The primary moving average for bounce
detection.self.filter_ma: The longer moving average for trend
filtering.self.order: Tracks any pending orders.self.stop_price: Stores the calculated stop loss price
for the current position.self.trade_count, self.bounce_entries,
self.stop_loss_exits, self.profitable_trades:
Custom counters to track strategy statistics.notify_order,
notify_trade):notify_order: This method is called by
backtrader when an order’s status changes.
buy order, it logs the execution, sets
the stop_price based on the entry price and
stop_loss_pct, and increments trade counters.sell executions and any order
rejections/cancellations.self.order is reset after an order is completed or
fails.notify_trade: This method is called when a trade (a
complete buy and sell cycle) is closed. It logs the profit/loss of the
trade and increments self.profitable_trades if the trade
was profitable. It also resets self.stop_price.next):The next method contains the core trading logic,
executed on each new bar of data:
self.position) and a stop_price is set, it
checks if the current close price has fallen below the
stop_price. If so, a close order is placed,
and self.stop_loss_exits is incremented. The method then
returns.if not self.position):
filter_ma AND the key_ma is also above the
filter_ma.key_ma of the previous
bar.key_ma of the current bar.buy order is placed, with the size determined by
order_percentage of the available cash.This function automates the process of finding the most effective combination of strategy parameters by running multiple backtests and evaluating their performance using metrics like Sharpe Ratio and trade statistics.
# ------------------------------------------------------------------
# 2. Optimization Function
# ------------------------------------------------------------------
def optimize_ma_bounce_parameters():
"""Run optimization to find best parameters with diagnostics"""
print("="*60)
print("MA BOUNCE STRATEGY OPTIMIZATION")
print("="*60)
# Fetch data for optimization
print("Fetching data for optimization...")
df = yf.download('BTC-USD', start='2020-01-01', end='2025-01-01', auto_adjust=False, progress=False)
if isinstance(df.columns, pd.MultiIndex):
df = df.droplevel(1, axis=1)
print(f"Data shape: {df.shape}")
print(f"Date range: {df.index[0]} to {df.index[-1]}")
# Set up optimization
cerebro = bt.Cerebro()
data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)
start_cash = 10000.0
cerebro.broker.setcash(start_cash)
cerebro.broker.setcommission(commission=0.001)
# Add analyzers
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe',
timeframe=bt.TimeFrame.Days, annualize=True, riskfreerate=0.0)
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
print("Testing parameter combinations...")
# Start with smaller parameter ranges for testing
cerebro.optstrategy(
MaBounceStrategy,
key_ma_period=[7, 14, 20], # 3 values - Bounce MA period
filter_ma_period=[50, 100, 200], # 3 values - Trend filter MA
ma_type=['SMA'], # 1 value - Start with SMA only
stop_loss_pct=[0.02, 0.03, 0.05] # 3 values - Stop loss %
)
# Total: 3 × 3 × 1 × 3 = 27 combinations
stratruns = cerebro.run()
print(f"Optimization complete! Tested {len(stratruns)} combinations.")
# Collect and analyze results with detailed diagnostics
results = []
valid_count = 0
no_trades_count = 0
invalid_sharpe_count = 0
for i, run in enumerate(stratruns):
strategy = run[0]
sharpe_analysis = strategy.analyzers.sharpe.get_analysis()
returns_analysis = strategy.analyzers.returns.get_analysis()
trades_analysis = strategy.analyzers.trades.get_analysis()
rtot = returns_analysis.get('rtot', 0.0)
final_value = start_cash * (1 + rtot)
sharpe_ratio = sharpe_analysis.get('sharperatio', None)
total_trades = trades_analysis.get('total', {}).get('total', 0)
# Diagnostic information
has_trades = total_trades > 0
has_valid_sharpe = sharpe_ratio is not None and not np.isnan(sharpe_ratio)
if not has_trades:
no_trades_count += 1
if not has_valid_sharpe:
invalid_sharpe_count += 1
sharpe_ratio = -999.0
# Safe attribute access
trade_count = getattr(strategy, 'trade_count', 0)
bounce_entries = getattr(strategy, 'bounce_entries', 0)
stop_loss_exits = getattr(strategy, 'stop_loss_exits', 0)
profitable_trades = getattr(strategy, 'profitable_trades', 0)
result = {
'combination_id': i + 1,
'sharpe_ratio': sharpe_ratio,
'final_value': final_value,
'return_pct': rtot * 100,
'total_analyzer_trades': total_trades,
'key_ma_period': strategy.p.key_ma_period,
'filter_ma_period': strategy.p.filter_ma_period,
'ma_type': strategy.p.ma_type,
'stop_loss_pct': strategy.p.stop_loss_pct,
'trade_count': trade_count,
'bounce_entries': bounce_entries,
'stop_loss_exits': stop_loss_exits,
'profitable_trades': profitable_trades,
'has_trades': has_trades,
'has_valid_sharpe': has_valid_sharpe,
}
results.append(result)
if has_trades and has_valid_sharpe:
valid_count += 1
# Print diagnostics
print(f"\n{'='*60}")
print("OPTIMIZATION DIAGNOSTICS")
print(f"{'='*60}")
print(f"Total combinations tested: {len(results)}")
print(f"Combinations with trades: {len(results) - no_trades_count}")
print(f"Combinations with no trades: {no_trades_count}")
print(f"Combinations with invalid Sharpe: {invalid_sharpe_count}")
print(f"Valid combinations: {valid_count}")
# Show some examples of each category
print(f"\n--- SAMPLE RESULTS ---")
for result in results[:5]: # Show first 5 combinations
sharpe_display = f"{result['sharpe_ratio']:.3f}" if result['sharpe_ratio'] != -999.0 else "Invalid"
print(f"Combination {result['combination_id']}: "
f"Key MA({result['key_ma_period']}) Filter MA({result['filter_ma_period']}) "
f"Stop({result['stop_loss_pct']:.1%}) -> "
f"Trades: {result['total_analyzer_trades']}, "
f"Return: {result['return_pct']:.1f}%, "
f"Sharpe: {sharpe_display}")
# Try to find any valid results
valid_results = [r for r in results if r['has_trades'] and r['has_valid_sharpe']]
if not valid_results:
print(f"\n{'='*60}")
print("NO VALID RESULTS FOUND - RUNNING SINGLE STRATEGY TEST")
print(f"{'='*60}")
# Test a single strategy with logging enabled to see what's happening
test_params = {
'key_ma_period': 14,
'filter_ma_period': 50,
'ma_type': 'SMA',
'stop_loss_pct': 0.03,
'printlog': True # Enable logging
}
print(f"Testing single strategy with parameters: {test_params}")
cerebro_test = bt.Cerebro()
cerebro_test.adddata(bt.feeds.PandasData(dataname=df))
cerebro_test.addstrategy(MaBounceStrategy, **test_params)
cerebro_test.broker.setcash(10000)
cerebro_test.broker.setcommission(commission=0.001)
cerebro_test.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
print("Running test strategy...")
test_result = cerebro_test.run()
test_strategy = test_result[0]
trades_test = test_strategy.analyzers.trades.get_analysis()
total_trades_test = trades_test.get('total', {}).get('total', 0)
print(f"Test strategy results:")
print(f" Total trades: {total_trades_test}")
print(f" Strategy trade count: {getattr(test_strategy, 'trade_count', 0)}")
print(f" Bounce entries: {getattr(test_strategy, 'bounce_entries', 0)}")
print(f" Final value: ${cerebro_test.broker.getvalue():.2f}")
if total_trades_test == 0:
print(f"\n*** STRATEGY IS NOT GENERATING ANY TRADES ***")
print(f"Possible issues:")
print(f"1. MA periods too long - not enough data for setup")
print(f"2. Uptrend conditions too strict")
print(f"3. Bounce conditions not being met")
print(f"4. Data period doesn't contain suitable market conditions")
# Let's check the data characteristics
print(f"\nDATA ANALYSIS:")
print(f"Price range: ${df['Close'].min():.2f} - ${df['Close'].max():.2f}")
print(f"Price change: {((df['Close'].iloc[-1] / df['Close'].iloc[0]) - 1) * 100:.1f}%")
# Calculate some MAs to see if conditions are being met
sma_14 = df['Close'].rolling(14).mean()
sma_50 = df['Close'].rolling(50).mean()
# Check how often we're in uptrend
uptrend_conditions = (df['Close'] > sma_50) & (sma_14 > sma_50)
uptrend_days = uptrend_conditions.sum()
total_days = len(df)
print(f"Days in uptrend (Price > SMA50 & SMA14 > SMA50): {uptrend_days}/{total_days} ({uptrend_days/total_days*100:.1f}%)")
if uptrend_days < 100:
print("*** Very few uptrend days - this may explain lack of trades ***")
return None
# If we have valid results, continue with normal processing
results_sorted = sorted(valid_results, key=lambda x: x['sharpe_ratio'], reverse=True)
print(f"\n{'='*120}")
print("TOP 10 PARAMETER COMBINATIONS BY SHARPE RATIO")
print(f"{'='*120}")
print("Rank | Sharpe | Return% | Value | Key MA | Filter MA | Type | Stop% | Trades | Bounces | Win%")
print("-" * 120)
for i, result in enumerate(results_sorted[:10]):
win_rate = (result['profitable_trades'] / max(1, result['trade_count'])) * 100
print(f"{i+1:4d} | {result['sharpe_ratio']:5.2f} | {result['return_pct']:6.1f}% | "
f"${result['final_value']:8,.0f} | {result['key_ma_period']:6d} | {result['filter_ma_period']:9d} | "
f"{result['ma_type']:4s} | {result['stop_loss_pct']:4.1%} | {result['total_analyzer_trades']:6d} | "
f"{result['bounce_entries']:7d} | {win_rate:4.1f}%")
best_params = results_sorted[0]
print(f"\n{'='*60}")
print("BEST PARAMETERS FOUND:")
print(f"{'='*60}")
print(f"Key MA Period: {best_params['key_ma_period']}")
print(f"Filter MA Period: {best_params['filter_ma_period']}")
print(f"MA Type: {best_params['ma_type']}")
print(f"Stop Loss: {best_params['stop_loss_pct']:.1%}")
print(f"Sharpe Ratio: {best_params['sharpe_ratio']:.3f}")
print(f"Total Return: {best_params['return_pct']:.1f}%")
print(f"Total Trades: {best_params['total_analyzer_trades']}")
print(f"Bounce Entries: {best_params['bounce_entries']}")
win_rate = (best_params['profitable_trades'] / max(1, best_params['trade_count'])) * 100
print(f"Win Rate: {win_rate:.1f}%")
return best_paramsThe optimize_ma_bounce_parameters function performs a
parameter optimization for the MaBounceStrategy.
bt.Cerebro instance, adds the data,
initial cash, and commission. It also adds SharpeRatio,
Returns, and TradeAnalyzer to collect detailed
performance metrics.cerebro.optstrategy is used to run the strategy with
various combinations of key_ma_period,
filter_ma_period, ma_type, and
stop_loss_pct.A rolling backtest evaluates a strategy’s performance over sequential, overlapping or non-overlapping time windows. This provides a more robust assessment of its consistency across different market conditions than a single historical backtest.
# ------------------------------------------------------------------
# 3. Rolling Backtest Function
# ------------------------------------------------------------------
def run_rolling_backtest(ticker, start, end, window_months, strategy_params=None):
"""Rolling backtest function for MA Bounce strategy"""
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()}")
# Fetch data using yfinance
data = yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)
if data.empty or len(data) < 90:
print("Not enough data for this period.")
current_start += rd.relativedelta(months=window_months)
continue
if isinstance(data.columns, pd.MultiIndex):
data = data.droplevel(1, 1)
# Calculate Buy & Hold return for the period
start_price = data['Close'].iloc[0]
end_price = data['Close'].iloc[-1]
benchmark_ret = (end_price - start_price) / start_price * 100
feed = bt.feeds.PandasData(dataname=data)
cerebro = bt.Cerebro()
cerebro.addstrategy(MaBounceStrategy, **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()
result = cerebro.run()
final_val = cerebro.broker.getvalue()
strategy_ret = (final_val - start_val) / start_val * 100
# Get strategy statistics
strategy_instance = result[0]
trades = getattr(strategy_instance, 'trade_count', 0)
bounces = getattr(strategy_instance, 'bounce_entries', 0)
stops = getattr(strategy_instance, 'stop_loss_exits', 0)
profitable = getattr(strategy_instance, 'profitable_trades', 0)
all_results.append({
'start': current_start.date(),
'end': current_end.date(),
'return_pct': strategy_ret,
'benchmark_pct': benchmark_ret,
'final_value': final_val,
'trades': trades,
'bounce_entries': bounces,
'stop_exits': stops,
'profitable_trades': profitable,
})
print(f"Strategy Return: {strategy_ret:.2f}% | Buy & Hold Return: {benchmark_ret:.2f}% | Trades: {trades}")
current_start += rd.relativedelta(months=window_months)
return pd.DataFrame(all_results)The run_rolling_backtest function takes a
ticker, an overall start and end
date, and a window_months parameter to define the size of
each rolling window.
yfinance
(with auto_adjust=False and droplevel for
column handling), sets up a bt.Cerebro instance, adds the
MaBounceStrategy with the specified
strategy_params, and runs the backtest.The main execution block performs the following steps:
optimize_ma_bounce_parameters() to find the best performing
parameters for the strategy over a specific historical period (2020-2025
for BTC-USD). This step includes detailed diagnostics for the
optimization process.run_rolling_backtest() function to evaluate the strategy
with the optimized parameters over a longer, comprehensive
period (2018-2025 for BTC-USD), using 12-month rolling windows.report_rolling_stats_with_plots() to display key
performance metrics and generate the three specified plots (cumulative
returns, return distribution, and period-by-period comparison).
============================================================
BEST PARAMETERS FOUND:
============================================================
Key MA Period: 14
Filter MA Period: 50
MA Type: SMA
Stop Loss: 3.0%
Sharpe Ratio: 0.962
Total Return: 253.0%
Total Trades: 1
Bounce Entries: 0
Win Rate: 0.0%
============================================================
ROLLING BACKTEST STATISTICS
============================================================
Total Periods: 7
Strategy Wins: 2 (28.6%)
Strategy Losses: 5 (71.4%)
STRATEGY PERFORMANCE:
Mean Return: 61.80%
Std Deviation: 110.07%
Best Period: 293.22%
Worst Period: -28.61%
BUY & HOLD PERFORMANCE:
Mean Return: 82.21%
Std Deviation: 129.78%
Best Period: 302.79%
Worst Period: -72.60%
CUMULATIVE PERFORMANCE:
Strategy Total: 870.4%
Buy & Hold Total: 507.8%
Outperformance: +362.6 percentage points
TRADING STATISTICS:
Total Trades: 17
Bounce Entries: 17
Stop Loss Exits: 12
Profitable Trades: 0
Overall Win Rate: 0.0%
Stop Loss Rate: 70.6%
RISK-ADJUSTED METRICS:
Strategy Sharpe Ratio: 0.562
Buy & Hold Sharpe Ratio: 0.633
============================================================
PERFORMANCE COMPARISON SUMMARY
============================================================
Strategy Average Return: 61.80%
Buy & Hold Average Return: 82.21%
Average Outperformance: -20.41%
Strategy Volatility: 110.07%
Buy & Hold Volatility: 129.78%
Volatility Difference: -19.71%
This structured approach allows for a thorough analysis of the MA Bounce Strategy, from identifying optimal parameters to understanding its performance consistency across different market conditions.