Backtrader is an exceptional Python framework for backtesting trading strategies, offering powerful building blocks. While its built-in indicators and functions cover many scenarios, true power comes from implementing your own custom logic – both in indicators and strategy rules, including vital risk management techniques.
This article presents a complete, runnable Backtrader example. We will:
yfinance
.This example showcases how to combine custom analytics with practical strategy execution and risk control, based on concepts detailed in “Backtrader Essentials: Building Successful Strategies with Python”.
First, we need our adaptive indicator. The VAMA adjusts its
sensitivity based on recent market volatility (measured by Standard
Deviation). It aims to be faster in volatile markets and smoother in
calm ones. Here is the complete code for the
VolatilityAdjustedMovingAverage
indicator class:
Python
import backtrader as bt
import math
# Ensure yfinance and matplotlib are installed: pip install yfinance matplotlib
import yfinance as yf
import matplotlib.pyplot as plt
class VolatilityAdjustedMovingAverage(bt.Indicator):
"""
Volatility Adjusted Moving Average (VAMA / VIDYA)
Adapts EMA smoothing based on the ratio of current to average volatility.
"""
lines = ('vama',)
params = (
('period', 20), # Base EMA period
('vol_period', 9), # Volatility calculation period (StdDev)
('min_alpha_ratio', 0.1), # Floor for volatility ratio adjustment
('max_alpha_ratio', 2.0), # Ceiling for volatility ratio adjustment
)
def __init__(self):
if self.p.vol_period <= 1:
raise ValueError("vol_period must be greater than 1")
if self.p.period <= 1:
raise ValueError("period must be greater than 1")
self.base_alpha = 2.0 / (self.p.period + 1.0)
self.vol = bt.indicators.StandardDeviation(self.data.close, period=self.p.vol_period)
self.avg_vol = bt.indicators.SimpleMovingAverage(self.vol, period=self.p.period)
# Set minimum period based on nested indicators' requirements
self.addminperiod(self.p.vol_period + self.p.period -1)
def next(self):
current_close = self.data.close[0]
current_vol = self.vol[0]
current_avg_vol = self.avg_vol[0]
if current_avg_vol is not None and current_avg_vol != 0 and not math.isnan(current_avg_vol):
vol_ratio = current_vol / current_avg_vol
else:
vol_ratio = 1.0 # Neutral ratio
# Clamp ratio and calculate adjusted alpha
vol_ratio = max(self.p.min_alpha_ratio, min(vol_ratio, self.p.max_alpha_ratio))
adjusted_alpha = self.base_alpha * vol_ratio
adjusted_alpha = max(1e-9, min(adjusted_alpha, 1.0)) # Ensure valid alpha
# Iterative VAMA calculation using adjusted alpha
if len(self) > 1 and not math.isnan(self.lines.vama[-1]):
prev_vama = self.lines.vama[-1]
self.lines.vama[0] = (adjusted_alpha * current_close) + ((1 - adjusted_alpha) * prev_vama)
else:
self.lines.vama[0] = current_close # Initialize
Key Points:
StandardDeviation
and
SimpleMovingAverage
indicators.addminperiod
ensures calculations wait for sufficient
data.next
method performs the bar-by-bar adaptive EMA
calculation, crucially depending on the previous VAMA value
(self.lines.vama[-1]
).Now, we build a strategy (VamaStrategy
) that utilizes
our custom indicator. This strategy enters long when the faster VAMA
crosses above a slower standard SMA and includes a trailing stop loss
for risk management.
Python
class VamaStrategy(bt.Strategy):
params = (
('vama_period', 20), # Passed to VAMA indicator
('vama_vol_period', 9), # Passed to VAMA indicator
('sma_period', 50), # Period for the slow SMA
('trail_perc', None), # Trailing stop percentage (e.g., 0.05). None to disable.
)
def __init__(self):
# Instantiate the custom VAMA indicator
self.vama = VolatilityAdjustedMovingAverage(
period=self.p.vama_period,
vol_period=self.p.vama_vol_period
)
# Instantiate the standard SMA
self.sma = bt.indicators.SimpleMovingAverage(
self.data.close, period=self.p.sma_period
)
# Instantiate the Crossover detector
self.crossover = bt.indicators.CrossOver(self.vama, self.sma)
# Order tracking
self.order = None
def next(self):
# Prevent placing new orders if one is already pending
if self.order:
return
# Entry Logic: No position currently open
if not self.position:
# Buy signal: VAMA crosses above SMA
if self.crossover > 0:
print(f'{self.data.datetime.date(0)}: BUY SIGNAL - VAMA {self.vama[0]:.2f} crossed above SMA {self.sma[0]:.2f}')
# Place the market buy order
self.order = self.buy()
# --- Add Trailing Stop Loss ---
# Check if trailing stop is enabled in parameters
if self.p.trail_perc is not None and self.p.trail_perc > 0.0:
# Place a sell stop order that trails the price
self.sell(exectype=bt.Order.StopTrail,
trailpercent=self.p.trail_perc)
print(f'{self.data.datetime.date(0)}: Trailing Stop set at {self.p.trail_perc*100:.2f}%')
# --- ---
# Exit Logic: Position currently open
else:
# Sell signal: VAMA crosses below SMA
if self.crossover < 0:
print(f'{self.data.datetime.date(0)}: SELL SIGNAL (Crossover) - VAMA {self.vama[0]:.2f} crossed below SMA {self.sma[0]:.2f}')
# Close the position; this should also cancel the associated trailing stop
self.order = self.close()
def notify_order(self, order):
# Handle order status notifications
if order.status in [order.Submitted, order.Accepted]:
# Order is pending, nothing to do for basic logic
return
if order.status in [order.Completed]:
if order.isbuy():
print(f'{self.data.datetime.date(0)}: BUY EXECUTED, Price: {order.executed.price:.2f}, Size: {order.executed.size}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}')
elif order.issell():
print(f'{self.data.datetime.date(0)}: SELL EXECUTED, Price: {order.executed.price:.2f}, Size: {order.executed.size}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}')
# Identify if the sell was due to the trailing stop
if order.exectype == bt.Order.StopTrail:
print(f'{self.data.datetime.date(0)}: --- Trailing Stop Loss Triggered ---')
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
print(f'{self.data.datetime.date(0)}: Order Canceled/Margin/Rejected - Status: {order.getstatusname()}')
# Reset order tracker allows checking signals again after order resolution
self.order = None
Strategy Highlights:
VAMA
, a
standard SMA
, and CrossOver
.self.crossover > 0
).exectype=bt.Order.StopTrail
) if
trail_perc
parameter is set. This order automatically
trails the price upwards.self.close()
) if the VAMA crosses back below the SMA
(self.crossover < 0
). This signal takes precedence if
the trailing stop hasn’t been hit yet. self.close()
should
cancel the pending trailing stop.Finally, we need the main script block
(if __name__ == '__main__':
) to set up the Backtrader
engine (Cerebro), load data, configure the test, run it, and plot the
results.
Python
# --- Cerebro Setup ---
if __name__ == '__main__':
# Create a Cerebro entity
cerebro = bt.Cerebro()
# --- Add data feed ---
print("Downloading Data...")
# Define parameters for data download
ticker = 'BTC-USD'
start_date = '2021-01-01'
end_date = '2021-12-31' # Adjust as needed
# Download data using yfinance
data_df = yf.download(ticker, start=start_date, end=end_date, progress=False)
# Ensure the multi-level column index from yfinance is removed if present (older versions)
if isinstance(data_df.columns, pd.MultiIndex):
data_df.columns = data_df.columns.droplevel(1)
if data_df.empty:
raise ValueError(f"Data download for {ticker} failed or returned empty DataFrame.")
print(f"Data Downloaded for {ticker}. Adding to Backtrader...")
# Wrap the pandas DataFrame using Backtrader's feed
data_feed = bt.feeds.PandasData(dataname=data_df)
cerebro.adddata(data_feed)
print("Data Added.")
# --- Add strategy ---
# Define strategy parameters
trail_stop_percentage = 0.10 # Example: 10% Trailing Stop
cerebro.addstrategy(VamaStrategy,
vama_period=30, # Example VAMA period
vama_vol_period=7, # Example VAMA volatility period
sma_period=90, # Example slow SMA period
trail_perc=trail_stop_percentage) # Pass the trailing stop parameter
print(f"Strategy Added with {trail_stop_percentage*100}% Trailing Stop.")
# --- Configure Broker ---
initial_cash = 10000.0
commission_rate = 0.001 # 0.1%
cerebro.broker.setcash(initial_cash)
cerebro.broker.setcommission(commission=commission_rate)
print(f"Broker configured: Cash={initial_cash}, Commission={commission_rate*100}%")
# --- Add Sizer ---
position_size_perc = 95 # Use 95% of equity per trade
cerebro.addsizer(bt.sizers.PercentSizer, percents=position_size_perc)
print(f"Sizer Added: {position_size_perc}% PercentSizer")
# --- Run Backtest ---
print(f'Starting Portfolio Value: {cerebro.broker.getvalue():.2f}')
print("Running Backtest...")
results = cerebro.run() # Run the backtest
print("Backtest Finished.")
print(f'Ending Portfolio Value: {cerebro.broker.getvalue():.2f}')
# --- Plot Results ---
print("Generating Plot...")
# Use iplot=False for non-interactive environments
cerebro.plot(style='candlestick', barup='green', bardown='red', iplot=False)
print("Plotting Complete.")
Setup Walkthrough:
yfinance
to download
historical data for a specified ticker and date range, then wraps the
resulting Pandas DataFrame with bt.feeds.PandasData
.VamaStrategy
,
crucially passing the desired parameters including
trail_perc
.PercentSizer
.cerebro.run()
and visualizes the results with
cerebro.plot()
.This complete example demonstrates the process of creating a custom adaptive indicator (VAMA), incorporating it into a trading strategy with specific entry/exit rules (VAMA/SMA crossover), and adding essential risk management through a trailing stop loss. By using Backtrader’s flexible framework, you can build and test sophisticated strategies tailored to your specific hypotheses about market behavior. Remember that parameter tuning and further testing across different market conditions are crucial steps before considering any strategy for live trading.