@jas2210
I'm learning Backtrader too, and noticed the Writer class, which may help you with this. (It is described in an article written after the documentation.) Write it Down
It writes a csv file, which includes time series data. I'm using it to help indicator development and plotting, but the file also contains trades and p&l, etc.
This is the Quickstart example from the docs with the Writer added (and a function to extract time series data from the csv file).
'''
from future import absolute_import, division, print_function, unicode_literals
from datetime import datetime
import pandas as pd
import matplotlib.pyplot as plt
import backtrader as bt
class TestStrategy(bt.Strategy):
params = (('maperiod', 15),
('printlog', False),
('tradelog', {})
)
def log(self, txt, dt=None, doprint=False):
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat():s}: {txt}')
def __init__(self):
# Keep a reference to the 'close' line in the data[0] dataseries.
self.data_close = self.datas[0].close
# Track pending orders.
self.order = None
self.buyprice = None
self.buycomm = None
# SMA indicator.
self.sma = bt.indicators.SimpleMovingAverage(self.datas[0], period=self.params.maperiod)
self.sma.csv = True
# Inidcators to demostrate plotting.
# bt.indicators.ExponentialMovingAverage(self.datas[0], period=25)
# bt.indicators.WeightedMovingAverage(self.datas[0], period=25, subplot=True)
# bt.indicators.StochasticSlow(self.datas[0])
# bt.indicators.MACDHisto(self.datas[0])
# rsi = bt.indicators.RSI(self.datas[0])
# bt.indicators.SmoothedMovingAverage(rsi, period=10)
# bt.indicators.ATR(self.datas[0], plot=False)
self.n_trades = 0
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
# The order has been submitted to or accepted by the broker. Do nothing.
return
# Check if the order has been completed.
if order.status in [order.Completed]:
if order.isbuy():
self.log(f'BUY: Fill price {order.executed.price:.2f}'
+ f', Cost {order.executed.value:.2f}'
+ f', Comm {order.executed.comm:.2f}')
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
elif order.issell():
self.log(f'SELL: Fill price {order.executed.price:.2f}'
+ f', Cost {order.executed.value:.2f}'
+ f', Comm {order.executed.comm:.2f}')
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
self.bar_executed = len(self)
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log("Order Canceled/Margin/Rejected.")
self.order = None # No pending order.
def notify_trade(self, trade):
if not trade.isclosed:
return
self.log(f'Operation Profit: Gross {trade.pnl:.2f}, Net {trade.pnlcomm:.2f}',
doprint=True)
self.params.tradelog[self.n_trades] = (trade.pnl,trade.pnlcomm )
self.n_trades += 1
def next(self):
# Simply log the closing price of the series from the reference.
self.log(f'Close {self.data_close[0]:.2f}')
# If there is a pending order...
if self.order:
return # Do nothing.
# If no existing position...
if not self.position:
# Check for a buy signal.
if self.data_close[0] > self.sma[0]:
self.log(f'Buy at {self.data_close[0]:.2f}')
self.order = self.buy()
# Else there is an existing position.
else:
# Check for a sell condition.
if self.data_close[0] < self.sma[0]:
self.log(f'Sell at {self.data_close[0]:.2f}')
self.order = self.sell()
def stop(self):
self.log(f'(MA Period {self.params.maperiod:2d}) Ending value {self.broker.getvalue():.2f}',
doprint=True)
def get_timeseries_data(writer_filepath):
# From the Backtrader documentation: the time series block is preceded and
# followed by a row of '====='.
raw_df = pd.read_csv(writer_filepath, skiprows=1)
# Keep the time series rows.
first_drop_row = (raw_df.loc[raw_df['Id'].str.find("===") > -1]
.index.values[0])
data_df = raw_df.iloc[0 : first_drop_row]
# Make datetime column the index.
data_df.set_index('datetime',inplace=True)
data_df.index = data_df.index.astype('datetime64[ns]')
return data_df
if name == 'main':
# Create the back test engine.
cerebro = bt.Cerebro()
# Dictionary to store trades.
trade_log = {}
# Add the strategy.
cerebro.addstrategy(TestStrategy, maperiod=20, printlog=False,
tradelog=trade_log)
# Alternatively, optimise.
# strats = cerebro.optstrategy(TestStrategy, maperiod=range(10, 31))
# Create a data feed.
data_path = ('C:/Users/tony/Documents/SwingTree/Trading/Automation/Python/'
+ 'BackTrader Sample Data/datas/orcl-1995-2014.txt')
data = bt.feeds.YahooFinanceCSVData(dataname=data_path,
fromdate=datetime(2000, 1, 1),
todate=datetime(2000, 12, 31),
reverse=False)
# Add the data feed to Cerebro.
cerebro.adddata(data)
cerebro.broker.setcash(1000.0)
cerebro.broker.setcommission(commission=0.001)
cerebro.addsizer(bt.sizers.FixedSize, stake=10)
# Add Analyzers.
cerebro.addanalyzer(bt.analyzers.Transactions, _name='Transactions')
cerebro.addanalyzer(bt.analyzers.GrossLeverage, _name='GrossLeverage')
# Add the Writer.
cerebro.addwriter(bt.WriterFile, csv=True, out='Test.csv')
# Run the backtest.
print(f'Starting porfolio value: {cerebro.broker.getvalue():.2f}')
strat_results = cerebro.run()
# cerebro.run(maxcpus=1) # For optimisation.
print(f'Final porfolio value: {cerebro.broker.getvalue():.2f}')
# Plot the results.
data_df = get_timeseries_data('Test.csv')
plt.plot(data_df[['close', 'sma']])
'''
