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
"ignore")
warnings.filterwarnings(
# ------------------------------------------------------------------
# 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 or self.datas[0].datetime.date(0)
dt 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
= bt.indicators.SMA if self.params.ma_type == 'SMA' else bt.indicators.EMA
ma_indicator
# 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)
= (self.data_close[0] > self.filter_ma[0] and
uptrend_confirmed 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
= self.data_low[-1] <= self.key_ma[-1]
touched_ma_prev_bar
# 3. Check for Rejection/Entry Trigger: Price closes back ABOVE the key MA on the current bar
= self.data_close[0] > self.key_ma[0]
closed_above_ma_curr_bar
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}')
= self.broker.get_cash()
cash = (cash * self.params.order_percentage) / self.data_close[0]
size 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...")
= yf.download('BTC-USD', start='2020-01-01', end='2025-01-01', auto_adjust=False, progress=False)
df if isinstance(df.columns, pd.MultiIndex):
= df.droplevel(1, axis=1)
df
print(f"Data shape: {df.shape}")
print(f"Date range: {df.index[0]} to {df.index[-1]}")
# Set up optimization
= bt.Cerebro()
cerebro = bt.feeds.PandasData(dataname=df)
data
cerebro.adddata(data)
= 10000.0
start_cash
cerebro.broker.setcash(start_cash)=0.001)
cerebro.broker.setcommission(commission
# Add analyzers
='sharpe',
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name=bt.TimeFrame.Days, annualize=True, riskfreerate=0.0)
timeframe='returns')
cerebro.addanalyzer(bt.analyzers.Returns, _name='trades')
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name
print("Testing parameter combinations...")
# Start with smaller parameter ranges for testing
cerebro.optstrategy(
MaBounceStrategy,=[7, 14, 20], # 3 values - Bounce MA period
key_ma_period=[50, 100, 200], # 3 values - Trend filter MA
filter_ma_period=['SMA'], # 1 value - Start with SMA only
ma_type=[0.02, 0.03, 0.05] # 3 values - Stop loss %
stop_loss_pct
)# Total: 3 × 3 × 1 × 3 = 27 combinations
= cerebro.run()
stratruns print(f"Optimization complete! Tested {len(stratruns)} combinations.")
# Collect and analyze results with detailed diagnostics
= []
results = 0
valid_count = 0
no_trades_count = 0
invalid_sharpe_count
for i, run in enumerate(stratruns):
= run[0]
strategy = strategy.analyzers.sharpe.get_analysis()
sharpe_analysis = strategy.analyzers.returns.get_analysis()
returns_analysis = strategy.analyzers.trades.get_analysis()
trades_analysis
= returns_analysis.get('rtot', 0.0)
rtot = start_cash * (1 + rtot)
final_value = sharpe_analysis.get('sharperatio', None)
sharpe_ratio = trades_analysis.get('total', {}).get('total', 0)
total_trades
# Diagnostic information
= total_trades > 0
has_trades = sharpe_ratio is not None and not np.isnan(sharpe_ratio)
has_valid_sharpe
if not has_trades:
+= 1
no_trades_count if not has_valid_sharpe:
+= 1
invalid_sharpe_count = -999.0
sharpe_ratio
# Safe attribute access
= getattr(strategy, 'trade_count', 0)
trade_count = getattr(strategy, 'bounce_entries', 0)
bounce_entries = getattr(strategy, 'stop_loss_exits', 0)
stop_loss_exits = getattr(strategy, 'profitable_trades', 0)
profitable_trades
= {
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:
+= 1
valid_count
# 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
= f"{result['sharpe_ratio']:.3f}" if result['sharpe_ratio'] != -999.0 else "Invalid"
sharpe_display 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
= [r for r in results if r['has_trades'] and r['has_valid_sharpe']]
valid_results
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}")
= bt.Cerebro()
cerebro_test =df))
cerebro_test.adddata(bt.feeds.PandasData(dataname**test_params)
cerebro_test.addstrategy(MaBounceStrategy, 10000)
cerebro_test.broker.setcash(=0.001)
cerebro_test.broker.setcommission(commission='trades')
cerebro_test.addanalyzer(bt.analyzers.TradeAnalyzer, _name
print("Running test strategy...")
= cerebro_test.run()
test_result = test_result[0]
test_strategy
= test_strategy.analyzers.trades.get_analysis()
trades_test = trades_test.get('total', {}).get('total', 0)
total_trades_test
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
= df['Close'].rolling(14).mean()
sma_14 = df['Close'].rolling(50).mean()
sma_50
# Check how often we're in uptrend
= (df['Close'] > sma_50) & (sma_14 > sma_50)
uptrend_conditions = uptrend_conditions.sum()
uptrend_days = len(df)
total_days
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
= sorted(valid_results, key=lambda x: x['sharpe_ratio'], reverse=True)
results_sorted
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]):
= (result['profitable_trades'] / max(1, result['trade_count'])) * 100
win_rate 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}%")
= results_sorted[0]
best_params 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']}")
= (best_params['profitable_trades'] / max(1, best_params['trade_count'])) * 100
win_rate print(f"Win Rate: {win_rate:.1f}%")
return best_params
The 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 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()}")
# Fetch data using yfinance
= yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)
data
if data.empty or len(data) < 90:
print("Not enough data for this period.")
+= rd.relativedelta(months=window_months)
current_start continue
if isinstance(data.columns, pd.MultiIndex):
= data.droplevel(1, 1)
data
# Calculate Buy & Hold return for the period
= data['Close'].iloc[0]
start_price = data['Close'].iloc[-1]
end_price = (end_price - start_price) / start_price * 100
benchmark_ret
= bt.feeds.PandasData(dataname=data)
feed = bt.Cerebro()
cerebro
**strategy_params)
cerebro.addstrategy(MaBounceStrategy,
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()
result = cerebro.broker.getvalue()
final_val = (final_val - start_val) / start_val * 100
strategy_ret
# Get strategy statistics
= result[0]
strategy_instance = getattr(strategy_instance, 'trade_count', 0)
trades = getattr(strategy_instance, 'bounce_entries', 0)
bounces = getattr(strategy_instance, 'stop_loss_exits', 0)
stops = getattr(strategy_instance, 'profitable_trades', 0)
profitable
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}")
+= rd.relativedelta(months=window_months)
current_start
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.