Simply access some values like trade.pnl
-
Hello everyone,
I am following the official guide (first pages).I would like to store trade.pnl values in a list for example. And the for example I want to get the sum of all pnl of the positions.
I cannot do it inside this function of the class because every time trade.pnl is different? Where I have to puy an empty list and append this values? In the class "class Name_of_strategy(bt.Strategy):" or outside of the class?
def notify_trade(self, trade): if not trade.isclosed: return self.log('OPERATION PROFIT, GROSS %.2f, NET %.2f' % (trade.pnl, trade.pnlcomm))I am lacking the real function of cerebro.run I think...
Thank you :)
-
@jas2210 It could be placed inside a strategy class of cause. However it is even better to use analyzers for such type of statistics gathering.
More docs here: https://www.backtrader.com/docu/analyzers/analyzers/
-
@vladisld Yes I suspected it.
I just to know how to move from the values results in the iteretion of the cerebro engine to for example a dataframe in pandas or an array in numpy.
Basically from lines object to an array if I am not wrong.Thanks for your time.
-
@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 DownIt 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_literalsfrom datetime import datetime
import pandas as pd
import matplotlib.pyplot as plt
import backtrader as btclass 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_dfif 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']])'''
-
@jas2210 Definitely use analyzers. You can collect a wide array of data into dictionaries/lists/tuples, etc.
Then at the end of your run, you can extract the analyzer, and then turn it into a dataframe.
Here's an example of an analyzer that collects closed trade information similar to what you are after:
class TradeClosed(bt.analyzers.Analyzer): """ Analyzer returning closed trade information. """ def start(self): super(TradeClosed, self).start() def create_analysis(self): self.rets = {} self.vals = tuple() def notify_trade(self, trade): """Receives trade notifications before each next cycle""" if trade.isclosed: self.vals = ( self.strategy.datetime.datetime(), trade.data._name, round(trade.pnl, 2), round(trade.pnlcomm, 2), trade.commission, (trade.dtclose - trade.dtopen), ) self.rets[trade.ref] = self.vals def get_analysis(self): return self.retsYou would implement when setting up cerebro using:
self.cerebro.addanalyzer(TradeClosed, _name="trade_closed")You would then use the following line to get your data after the backtest runs:
trade_dict = results[0].analyzers.getbyname("trade_closed").get_analysis()Finally, turn this into a dataframe:
columns_df = [ "Date Closed", "Ticker", "PnL", "PnL Comm", "Commission", "Days Open", ] df = pd.DataFrame(trade_dict) df = df.T df.columns = columns_df -
@run-out
Thanks. That's much simpler.
