إطار Backtrader
إطار اختبار خلفي متكامل قائم على الأحداث. بناء واختبار وتحسين استراتيجيات التداول مع مؤشرات ومحللات مدمجة.
الصعوبة: متوسط
الفئة: اختبار خلفي
⭐ معيار الصناعة
التثبيت
# Install backtrader
$ pip install backtrader
# Install with plotting support
$ pip install backtrader[plotting]
بنية الإطار
Cerebro Engine
Central coordinator managing strategies, data feeds, and execution
Strategy Classes
Define your trading logic with init(), next(), and notify methods
Indicators
100+ built-in technical indicators and easy custom indicator creation
Analyzers
Calculate performance metrics like Sharpe ratio, drawdown, and returns
أمثلة كود شاملة
Your First Backtrader Strategy
Simple SMA crossover strategy implementation
Python
import backtrader as bt
class SMAStrategy(bt.Strategy):
params = (
('fast_period', 50),
('slow_period', 200),
)
def __init__(self):
# Calculate indicators
self.fast_sma = bt.indicators.SMA(
self.data.close, period=self.params.fast_period
)
self.slow_sma = bt.indicators.SMA(
self.data.close, period=self.params.slow_period
)
# Crossover signal
self.crossover = bt.indicators.CrossOver(self.fast_sma, self.slow_sma)
def next(self):
if not self.position: # Not in market
if self.crossover > 0: # Golden cross
self.buy()
else: # In market
if self.crossover < 0: # Death cross
self.close()
# Create cerebro engine
cerebro = bt.Cerebro()
cerebro.addstrategy(SMAStrategy)
# Add data
data = bt.feeds.YahooFinanceData(
dataname='EURUSD=X',
fromdate=datetime(2023, 1, 1),
todate=datetime(2024, 1, 1)
)
cerebro.adddata(data)
# Set initial capital
cerebro.broker.setcash(10000.0)
# Run backtest
print(f'Starting Portfolio Value: {cerebro.broker.getvalue():.2f}')
cerebro.run()
print(f'Final Portfolio Value: {cerebro.broker.getvalue():.2f}')
# Plot results
cerebro.plot()
Using Built-in Indicators
Leverage backtrader's extensive indicator library
Python
class IndicatorStrategy(bt.Strategy):
def __init__(self):
# Moving averages
self.sma = bt.indicators.SMA(self.data.close, period=20)
self.ema = bt.indicators.EMA(self.data.close, period=20)
# Momentum indicators
self.rsi = bt.indicators.RSI(self.data.close, period=14)
self.macd = bt.indicators.MACD(self.data.close)
self.stochastic = bt.indicators.Stochastic(self.data)
# Volatility indicators
self.atr = bt.indicators.ATR(self.data, period=14)
self.bbands = bt.indicators.BollingerBands(self.data.close)
# Volume indicators
self.volume_sma = bt.indicators.SMA(self.data.volume, period=20)
def next(self):
# Example: RSI oversold + MACD bullish crossover
if self.rsi < 30 and self.macd.macd > self.macd.signal:
if not self.position:
self.buy()
# Example: RSI overbought
elif self.rsi > 70:
if self.position:
self.close()
# Print current values
self.log(f'RSI: {self.rsi[0]:.2f}, MACD: {self.macd.macd[0]:.4f}')
def log(self, txt):
dt = self.datas[0].datetime.date(0)
print(f'{dt.isoformat()} {txt}')
Advanced Order Types
Market, limit, stop, and bracket orders
Python
class OrderStrategy(bt.Strategy):
def __init__(self):
self.order = None
self.sma = bt.indicators.SMA(self.data.close, period=20)
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}')
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')
self.order = None
def next(self):
if self.order:
return # Pending order exists
if not self.position:
if self.data.close > self.sma:
# Market order
self.order = self.buy()
# Or limit order
# limit_price = self.data.close * 0.99
# self.order = self.buy(exectype=bt.Order.Limit, price=limit_price)
# Or stop order
# stop_price = self.data.close * 1.01
# self.order = self.buy(exectype=bt.Order.Stop, price=stop_price)
else:
# Bracket order (entry + SL + TP)
if self.data.close < self.sma:
# Calculate SL and TP
entry_price = self.position.price
sl_price = entry_price * 0.98 # 2% stop loss
tp_price = entry_price * 1.04 # 4% take profit
# Close with bracket
self.sell(exectype=bt.Order.Stop, price=sl_price)
self.sell(exectype=bt.Order.Limit, price=tp_price)
def log(self, txt):
dt = self.datas[0].datetime.date(0)
print(f'{dt.isoformat()} {txt}')
Position Sizing Strategies
Fixed, percentage, and Kelly criterion sizing
Python
class PositionSizingStrategy(bt.Strategy):
params = (
('risk_per_trade', 0.02), # 2% risk per trade
('atr_period', 14),
)
def __init__(self):
self.sma = bt.indicators.SMA(self.data.close, period=50)
self.atr = bt.indicators.ATR(self.data, period=self.params.atr_period)
def next(self):
if not self.position:
if self.data.close > self.sma:
# Calculate position size based on ATR
account_value = self.broker.getvalue()
risk_amount = account_value * self.params.risk_per_trade
# Stop loss at 2 ATR
sl_distance = 2 * self.atr[0]
# Position size = Risk Amount / Stop Loss Distance
position_size = risk_amount / sl_distance
# Convert to lots (assuming forex)
lot_size = position_size / 100000 # Standard lot
# Place order with calculated size
self.buy(size=lot_size)
self.log(f'Position Size: {lot_size:.2f} lots, ATR: {self.atr[0]:.5f}')
def log(self, txt):
dt = self.datas[0].datetime.date(0)
print(f'{dt.isoformat()} {txt}')
# Alternative: Fixed percentage sizing
class PercentSizer(bt.Sizer):
params = (('percents', 95),) # Use 95% of available cash
def _getsizing(self, comminfo, cash, data, isbuy):
if isbuy:
size = (cash * self.params.percents / 100) / data.close[0]
return int(size)
return self.broker.getposition(data).size
# Add sizer to cerebro
cerebro.addsizer(PercentSizer)
Performance Analysis
Use analyzers to calculate metrics
Python
# Add analyzers to cerebro
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
cerebro.addanalyzer(bt.analyzers.SQN, _name='sqn')
# Run backtest
results = cerebro.run()
strat = results[0]
# Extract analyzer results
sharpe = strat.analyzers.sharpe.get_analysis()
drawdown = strat.analyzers.drawdown.get_analysis()
returns = strat.analyzers.returns.get_analysis()
trades = strat.analyzers.trades.get_analysis()
sqn = strat.analyzers.sqn.get_analysis()
# Print results
print('\n=== Performance Metrics ===')
print(f'Sharpe Ratio: {sharpe.get("sharperatio", 0):.2f}')
print(f'Max Drawdown: {drawdown.max.drawdown:.2f}%')
print(f'Total Return: {returns.rtot * 100:.2f}%')
print(f'Annual Return: {returns.rnorm100:.2f}%')
print('\n=== Trade Statistics ===')
total_trades = trades.total.closed
won_trades = trades.won.total
lost_trades = trades.lost.total
win_rate = (won_trades / total_trades * 100) if total_trades > 0 else 0
print(f'Total Trades: {total_trades}')
print(f'Won: {won_trades}, Lost: {lost_trades}')
print(f'Win Rate: {win_rate:.2f}%')
print(f'Avg Win: {trades.won.pnl.average:.2f}')
print(f'Avg Loss: {trades.lost.pnl.average:.2f}')
print(f'SQN: {sqn.sqn:.2f}')
Parameter Optimization
Find optimal strategy parameters
Python
class OptimizableStrategy(bt.Strategy):
params = (
('fast_period', 50),
('slow_period', 200),
)
def __init__(self):
self.fast_sma = bt.indicators.SMA(self.data.close, period=self.params.fast_period)
self.slow_sma = bt.indicators.SMA(self.data.close, period=self.params.slow_period)
self.crossover = bt.indicators.CrossOver(self.fast_sma, self.slow_sma)
def next(self):
if not self.position:
if self.crossover > 0:
self.buy()
else:
if self.crossover < 0:
self.close()
# Optimize parameters
cerebro = bt.Cerebro()
# Add strategy with parameter ranges
cerebro.optstrategy(
OptimizableStrategy,
fast_period=range(20, 100, 10), # 20, 30, 40, ..., 90
slow_period=range(100, 300, 20) # 100, 120, 140, ..., 280
)
# Add data
data = bt.feeds.YahooFinanceData(dataname='EURUSD=X', fromdate=datetime(2023, 1, 1))
cerebro.adddata(data)
# Add analyzer
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
# Run optimization
print('Running optimization...')
results = cerebro.run()
# Find best parameters
best_sharpe = -999
best_params = None
for result in results:
strat = result[0]
sharpe = strat.analyzers.sharpe.get_analysis().get('sharperatio', 0)
if sharpe and sharpe > best_sharpe:
best_sharpe = sharpe
best_params = (strat.params.fast_period, strat.params.slow_period)
print(f'\nBest Parameters: Fast={best_params[0]}, Slow={best_params[1]}')
print(f'Best Sharpe Ratio: {best_sharpe:.2f}')
Walk-Forward Analysis
Avoid overfitting with out-of-sample testing
Python
from datetime import datetime, timedelta
def walk_forward_analysis(data_feed, in_sample_days=180, out_sample_days=60):
"""
Perform walk-forward analysis
in_sample_days: Days to optimize on
out_sample_days: Days to test on
"""
results = []
start_date = datetime(2023, 1, 1)
end_date = datetime(2024, 1, 1)
current_date = start_date
while current_date < end_date:
# In-sample period (optimization)
in_start = current_date
in_end = current_date + timedelta(days=in_sample_days)
# Out-of-sample period (testing)
out_start = in_end
out_end = out_start + timedelta(days=out_sample_days)
if out_end > end_date:
break
print(f'\nIn-Sample: {in_start.date()} to {in_end.date()}')
print(f'Out-Sample: {out_start.date()} to {out_end.date()}')
# Optimize on in-sample
cerebro_opt = bt.Cerebro()
cerebro_opt.optstrategy(
OptimizableStrategy,
fast_period=range(20, 100, 10),
slow_period=range(100, 300, 20)
)
# Add in-sample data
data_in = bt.feeds.YahooFinanceData(
dataname='EURUSD=X',
fromdate=in_start,
todate=in_end
)
cerebro_opt.adddata(data_in)
cerebro_opt.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
# Find best parameters
opt_results = cerebro_opt.run()
best_params = find_best_params(opt_results)
# Test on out-of-sample
cerebro_test = bt.Cerebro()
cerebro_test.addstrategy(
OptimizableStrategy,
fast_period=best_params[0],
slow_period=best_params[1]
)
data_out = bt.feeds.YahooFinanceData(
dataname='EURUSD=X',
fromdate=out_start,
todate=out_end
)
cerebro_test.adddata(data_out)
cerebro_test.addanalyzer(bt.analyzers.Returns, _name='returns')
test_results = cerebro_test.run()
returns = test_results[0].analyzers.returns.get_analysis()
results.append({
'period': f'{out_start.date()} to {out_end.date()}',
'params': best_params,
'return': returns.rtot * 100
})
# Move to next period
current_date = out_end
return results
# Run walk-forward
wf_results = walk_forward_analysis('EURUSD=X')
# Analyze results
for r in wf_results:
print(f"{r['period']}: Params={r['params']}, Return={r['return']:.2f}%")
Live Trading Integration
Connect to live broker feeds
Python
import backtrader as bt
class LiveStrategy(bt.Strategy):
def __init__(self):
self.sma = bt.indicators.SMA(self.data.close, period=20)
def notify_data(self, data, status, *args, **kwargs):
print(f'Data Status: {data._getstatusname(status)}')
if status == data.LIVE:
print('LIVE DATA - Trading with real money!')
def notify_order(self, order):
if order.status in [order.Completed]:
if order.isbuy():
print(f'BUY EXECUTED: {order.executed.price:.5f}')
else:
print(f'SELL EXECUTED: {order.executed.price:.5f}')
def next(self):
# Only trade during market hours
if not self.data.datetime.time() or \
self.data.datetime.time() < datetime.time(9, 0) or \
self.data.datetime.time() > datetime.time(17, 0):
return
if not self.position:
if self.data.close > self.sma:
self.buy()
else:
if self.data.close < self.sma:
self.close()
# Create cerebro for live trading
cerebro = bt.Cerebro()
# Add live data feed (example with IB)
# store = bt.stores.IBStore(host='127.0.0.1', port=7497, clientId=1)
# data = store.getdata(dataname='EUR.USD', timeframe=bt.TimeFrame.Minutes, compression=5)
# Or use Oanda
# store = bt.stores.OandaStore()
# data = store.getdata(dataname='EUR_USD', timeframe=bt.TimeFrame.Minutes, compression=5)
# cerebro.adddata(data)
# cerebro.addstrategy(LiveStrategy)
# cerebro.run()
print('Live trading setup complete. Uncomment broker-specific code to run.')
أفضل الممارسات
استخدم المحللات
Always add analyzers to track performance metrics automatically
اختبار المشي للأمام
Validate strategies with out-of-sample data to avoid overfitting
حجم المركز
Implement proper risk management with custom sizers
تجنب انحياز الاستباق
Never access future data in your strategy logic
العمولة والانزلاق
Always include realistic commission and slippage in backtests
جودة البيانات
Ensure clean, adjusted data for accurate backtest results
الموارد
الموارد الرسمية
المجتمع
- Backtrader Community Forum
- GitHub Examples Repository