For code/output blocks: Use ``` (aka backtick or grave accent) in a single line before and after the block. See: http://commonmark.org/help/

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.df
    

    Example 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 file

    class 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()