Backtrader Base Strategy (adapted from Esteban Thiliez for multiinstruments and some simplifications)

base_strategy.py

from abc import abstractmethod
import backtrader as bt


class BaseStrategy(bt.Strategy):
    params = dict(
        verbose=True,
        longs_enabled=True,
        shorts_enabled=True
    )

    def __init__(self):
        self.orders = []
        self._init_indicators()

    def _init_indicators(self):
        pass

    def next(self):
        for i, d in enumerate(self.datas):
            if not self.getposition(d).size:
                self._not_yet_in_market(i, d)
            else:
                self._in_market(i, d)

    def _not_yet_in_market(self, i, d):
        if self._open_long_condition(i, d) and self.params.longs_enabled:
            stop_price = self._get_long_stop_loss_price(d)
            take_profit_price = self._get_long_take_profit_price(d)
            self._go_long(d, stop_price, take_profit_price)

        if self._open_short_condition(i, d) and self.params.shorts_enabled:
            stop_price = self._get_short_stop_loss_price(d)
            take_profit_price = self._get_short_take_profit_price(d)
            self._go_short(d, stop_price, take_profit_price)

    def _in_market(self, i, d):
        if self._close_long_condition(i, d) and self.getposition(d).size > 0:
            return self.close(d)
        if self._close_short_condition(i, d) and self.getposition(d).size < 0:
            return self.close(d)

    def _go_long(self, d, stop_price, take_profit_price):
        orders = self._get_long_orders_from_stop_and_take_profit(d, stop_price, take_profit_price)
        self.orders_ref = [order.ref for order in orders if order]
        self.orders.append(self.orders_ref)

    def _go_short(self, d, stop_price, take_profit_price):
        orders = self._get_short_orders_from_stop_and_take_profit(d, stop_price, take_profit_price)
        self.orders_ref = [order.ref for order in orders if order]
        self.orders.append(self.orders_ref)

    def _get_long_orders_from_stop_and_take_profit(self, d, stop_price, take_profit_price):
        actual_price = d.close[0]
        if stop_price != actual_price and take_profit_price != actual_price:
            orders = self.buy_bracket(data=d, price=actual_price, stopprice=stop_price, limitprice=take_profit_price)
        elif stop_price != actual_price and take_profit_price == actual_price:
            orders = [self.buy(data=d), self.sell(data=d, exectype=bt.Order.Stop, price=stop_price)]
        elif stop_price == actual_price and take_profit_price != actual_price:
            orders = [self.buy(data=d), self.sell(data=d, exectype=bt.Order.Limit, price=take_profit_price)]
        else:
            orders = [self.buy(data=d)]
        return orders

    def _get_short_orders_from_stop_and_take_profit(self, d, stop_price, take_profit_price):
        actual_price = d.close[0]
        if stop_price != actual_price and take_profit_price != actual_price:
            orders = self.sell_bracket(data=d, price=actual_price, stopprice=stop_price, limitprice=take_profit_price)
        elif stop_price != actual_price and take_profit_price == actual_price:
            orders = [self.sell(data=d), self.buy(data=d, exectype=bt.Order.Stop, price=stop_price)]
        elif stop_price == actual_price and take_profit_price != actual_price:
            orders = [self.sell(data=d), self.buy(data=d, exectype=bt.Order.Limit, price=take_profit_price)]
        else:
            orders = [self.sell(data=d)]
        return orders

    @abstractmethod
    def _open_short_condition(self, i, d) -> bool:
        pass

    @abstractmethod
    def _open_long_condition(self, i, d) -> bool:
        pass

    @abstractmethod
    def _close_short_condition(self, i, d) -> bool:
        pass

    @abstractmethod
    def _close_long_condition(self, i, d) -> bool:
        pass

    def _get_long_stop_loss_price(self, d) -> float:
        return d.close[0]

    def _get_long_take_profit_price(self, d) -> float:
        return d.close[0]

    def _get_short_stop_loss_price(self, d) -> float:
        return d.close[0]

    def _get_short_take_profit_price(self, d) -> float:
        return d.close[0]

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

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

        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log('Order Canceled/Margin/Rejected')

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