Position analyzer with entry/exit details
-
Sometimes during backtesting, we may want to know the timings of entry/exit along with other position details so we can verify it with real data.
This is even more imperative when the strategy is being used to generate signals on T - 1 day and then later confirmed in the market how it would have performed on day T.I have written a little utility(Analyzer) which outputs all these details in a CSV file. Hope this is useful.
import time import queue import pandas as pd import backtrader as bt from collections import defaultdict class PositionAnalyzer(bt.Analyzer): params = ( ("continuous", False), ("fname", "position_{0}.csv".format(time.strftime("%Y%m%d-%H%M%S"))) ) def __init__(self): super(PositionAnalyzer, self).__init__() self.order_map = defaultdict(queue.Queue) self.df = pd.DataFrame(columns=["instrument", "entry_dt", "exit_dt", "size", "value", "direction", "entry_price", "exit_price", "pnl", "pnl_comm"]) def notify_order(self, order): if not order.status == order.Completed: return self.order_map[order.data].put(order) def notify_trade(self, trade): if not trade.isclosed: return entry_order = self.order_map[trade.data].get() exit_order = self.order_map[trade.data].get() instrument = trade.data._dataname entry_dt = bt.num2date(entry_order.executed.dt) exit_dt = bt.num2date(exit_order.executed.dt) size = entry_order.executed.size value = entry_order.executed.size * entry_order.executed.price direction = "BUY" if entry_order.isbuy() else "SELL" entry_price = entry_order.executed.price exit_price = exit_order.executed.price pnl = round(trade.pnl, 2) pnl_comm = round(trade.pnlcomm, 2) self.df.loc[len(self.df)] = [instrument, entry_dt, exit_dt, size, value, direction, entry_price, exit_price, pnl, pnl_comm] if self.p.continuous: self.df.to_csv(self.p.fname, index=False) def stop(self): if not self.p.continuous: self.df.to_csv(self.p.fname, index=False) def get_analysis(self): return self.dfExample of a generated output:
instrument,entry_dt,exit_dt,size,value,direction,entry_price,exit_price,pnl,pnl_comm ADANIPORTS,2018-05-22 05:30:00,2018-05-22 05:30:00,-1,-381.5,SELL,381.5,381.35,0.15,0.15 ADANIPORTS,2018-05-23 05:30:00,2018-05-23 05:30:00,-1,-377.75,SELL,377.75,374.05,3.7,3.7 ADANIPORTS,2018-05-24 05:30:00,2018-05-24 05:30:00,-1,-373.75,SELL,373.75,371.5,2.25,2.25 ADANIPORTS,2018-06-07 05:30:00,2018-06-07 05:30:00,1,378.0,BUY,378.0,381.0,3.0,3.0 ADANIPORTS,2018-06-14 05:30:00,2018-06-14 05:30:00,-1,-381.65,SELL,381.65,374.6,7.05,7.05 ADANIPORTS,2018-06-15 05:30:00,2018-06-15 05:30:00,-1,-371.3,SELL,371.3,371.65,-0.35,-0.35 ADANIPORTS,2018-06-21 05:30:00,2018-06-21 05:30:00,1,369.6,BUY,369.6,366.5,-3.1,-3.1 -
Thank you! Good analyzer.
Will it work in case of multiple position size changes?
Like buy 10, buy 20, sell 15, sell 15 = 1 trade with size of 30. -
@ab_trader
Thanks for the comments. This is a good use case and I will be thinking of how to add this. -
@kausality I've used trade history data in my trade list analyzer to get sizes and prices for trades with multiple orders. Here is the link -
https://community.backtrader.com/topic/1274/closed-trade-list-including-mfe-mae-analyzer -
Awesome work there!
I don't need to add anything.I am currently doing some live bot like trading and therefore I though adding a continuous dump feature would be helpful.
This little modification also allows one to dump the position details in a CSV fileclass trade_list(bt.Analyzer): params = ( ("continuous", False), ("fname", "position_{0}.csv".format(time.strftime("%Y%m%d-%H%M%S"))), ("dump", True) ) def __init__(self): self.trades = [] self.cumprofit = 0.0 def get_analysis(self): return self.trades def _write_to_file(self): if not self.p.dump: return df = pd.DataFrame(self.trades) df.to_csv(self.p.fname, index=False) def notify_trade(self, trade): if trade.isclosed: brokervalue = self.strategy.broker.getvalue() dir = "short" if trade.history[0].event.size > 0: dir = "long" pricein = trade.history[len(trade.history)-1].status.price priceout = trade.history[len(trade.history)-1].event.price datein = bt.num2date(trade.history[0].status.dt) dateout = bt.num2date(trade.history[len(trade.history)-1].status.dt) if trade.data._timeframe >= bt.TimeFrame.Days: datein = datein.date() dateout = dateout.date() pcntchange = 100 * priceout / pricein - 100 pnl = trade.history[len(trade.history)-1].status.pnlcomm pnlpcnt = 100 * pnl / brokervalue barlen = trade.history[len(trade.history)-1].status.barlen pbar = pnl / barlen self.cumprofit += pnl size = value = 0.0 for record in trade.history: if abs(size) < abs(record.status.size): size = record.status.size value = record.status.value highest_in_trade = max(trade.data.high.get(ago=0, size=barlen+1)) lowest_in_trade = min(trade.data.low.get(ago=0, size=barlen+1)) hp = 100 * (highest_in_trade - pricein) / pricein lp = 100 * (lowest_in_trade - pricein) / pricein if dir == "long": mfe = hp mae = lp else: mfe = -lp mae = -hp self.trades.append({"ref": trade.ref, "ticker": trade.data._name, "dir": dir, "datein": datein, "pricein": pricein, "dateout": dateout, "priceout": priceout, "chng%": round(pcntchange, 2), "pnl": pnl, "pnl%": round(pnlpcnt, 2), "size": size, "value": value, "cumpnl": self.cumprofit, "nbars": barlen, "pnl/bar": round(pbar, 2), "mfe%": round(mfe, 2), "mae%": round(mae, 2)}) if self.p.continuous: self._write_to_file() def stop(self): if not self.p.continuous: self._write_to_file()