A backtrader multidata example with golden cross strategy and stop trailer (based on strategy entries and exits)

golden_cross_stop_trailer.py

import backtrader as bt

from trading_backtesting.stop_trailer import StopTrailer


class GoldenCrossStrategy(bt.Strategy):
    SHORT, NONE, LONG = -1, 0, 1

    params = dict(
        fast_length=50,
        slow_length=200,
        stop_loss=False,
        trail_percent=None,
        atr_period=14,  # measure volatility over x days
        ema_period=10,  # smooth out period for atr volatility
        stop_factor=None,  # actual stop distance for smoothed atr
        samebar=True,  # close and re-open on samebar
        verbose=True
    )

    def in_market(self, d) -> bool:
        return self.getposition(d).size != 0

    def is_long_position(self, d) -> bool:
        return self.getposition(d).size > 0

    def is_short_position(self, d) -> bool:
        return self.getposition(d).size < 0

    def __init__(self):
        self.entering = dict()
        self.orders = dict()

        self.crossovers = dict()
        self.stop_trailers = dict()
        self.exit_longs = dict()
        self.exit_shorts = dict()

        for d in self.datas:
            ma_fast = bt.ind.SMA(d, period=self.params.fast_length)
            ma_slow = bt.ind.SMA(d, period=self.params.slow_length)
            self.crossovers[d] = bt.ind.CrossOver(ma_fast, ma_slow)

            self.stop_trailers[d] = StopTrailer(d, atrperiod=self.p.atr_period, emaperiod=self.p.ema_period,
                        stopfactor=self.p.stop_factor)

            self.exit_longs[d] = bt.ind.CrossDown(d, self.stop_trailers[d].stop_long, plotname='Exit Long')
            self.exit_shorts[d] = bt.ind.CrossUp(d, self.stop_trailers[d].stop_short, plotname='Exit Short')

    def start(self):
        for d in self.datas:
            self.entering[d] = 0

    def next(self):
        # Some ideas from: https://www.backtrader.com/blog/2019-08-22-practical-backtesting-replication/practical-replication/
        for i, d in enumerate(self.datas):
            closing = None
            """
            self.log(
                '%s, close: %.2f' %
                (d.p.name,
                 d.close[0]))
            """

            if self.is_long_position(d):
                self.log('Long Stop Price: {:.2f}', self.stop_trailers[d].stop_long[0])
                if self.exit_longs[d][0]:
                    closing = self.close(data=d)
            elif self.is_short_position(d):
                self.log('Short Stop Price: {:.2f}', self.stop_trailers[d].stop_short[0])
                if self.exit_shorts[d][0]:
                    closing = self.close(data=d)


            self.entering[d] = self.NONE

            if not self.in_market(d) or (closing and self.p.samebar):
                # Not in the market or closing pos and reenter in samebar
                if self.crossovers[d] > 0:
                    order = self.buy(data=d)
                    self.orders[d] = [order]
                    self.entering[d] = self.LONG if order else self.NONE
                elif self.crossovers[d] < 0:
                    order = self.sell(data=d)
                    self.orders[d] = [order]
                    self.entering[d] = self.SHORT if order else self.NONE

    def submit_stop_loss(self, order):
        d = order.p.data

        if self.p.stop_loss:
            if self.is_long_position(d) > 0:
                stop_price = order.executed.price * (1.0 - self.p.stop_loss)
                self.sell(exectype=bt.Order.Stop, price=stop_price)

            if self.is_long_position(d) < 0:
                stop_price = order.executed.price * (1.0 + self.p.stop_loss)
                self.sell(exectype=bt.Order.Stop, price=stop_price)

        if self.p.trail_percent:
            if self.is_long_position(d) > 0:
                self.sell(exectype=bt.Order.StopTrail, trailpercent=self.p.trail_percent)
            if self.is_long_position(d) < 0:
                self.buy(exectype=bt.Order.StopTrail, trailpercent=self.p.trail_percent)

    def notify_order(self, order):
        symbol = order.p.data.p.name
        size = order.size

        if order.status in [order.Submitted, order.Accepted]:
            # Buy/Sell order submitted/accepted to/by broker - Nothing to do
            return

        # Check if an order has been completed
        # Attention: broker could reject order if not enough cash
        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(
                    '%s, BUY EXECUTED, Size: %.2f, Price: %.2f, Comm %.2f' %
                    (symbol,
                     size,
                     order.executed.price,
                     order.executed.comm))

                self.buyprice = order.executed.price
                self.buycomm = order.executed.comm
            else:  # Sell
                self.log(
                    '%s, SELL EXECUTED, Size: %.2f, Price: %.2f, Comm %.2f' %
                    (symbol,
                     size,
                     order.executed.price,
                     order.executed.comm))

            self.submit_stop_loss(order)

        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log('Order Canceled/Margin/Rejected')
            dorders = self.orders[order.data]
            idx = dorders.index(order)
            dorders[idx] = None

    def notify_trade(self, trade):
        if not trade.isclosed:
            return

        symbol = trade.data.p.name
        self.log('%s, OPERATION PROFIT, GROSS %.2f, NET %.2f' %
                 (symbol, trade.pnl, trade.pnlcomm))

    def log(self, txt, *args):
        if self.p.verbose:
            out = [self.datetime.date().isoformat(), txt.format(*args)]
            print(','.join(out))