Welcome!

By registering with us, you'll be able to discuss, share and private message with other members of our community.

SignUp Now!
  • Guest, before posting your code please take these rules into consideration:
    • It is required to use our BBCode feature to display your code. While within the editor click < / > or >_ and place your code within the BB Code prompt. This helps others with finding a solution by making it easier to read and easier to copy.
    • You can also use markdown to share your code. When using markdown your code will be automatically converted to BBCode. For help with markdown check out the markdown guide.
    • Don't share a wall of code. All we want is the problem area, the code related to your issue.


    To learn more about how to use our BBCode feature, please click here.

    Thank you, Code Forum.

Python Help for backtrader library

Arkom

New Coder
Hello everyone, im getting started with the backtrader library because i'm interested in testing a few strategies in day-trading.

Here is my problem:

Im currently using a strategy using fractals, normally this is a candlestick pattern with delay and the fractal appears only once the fifth candlestick is closed.
Here is an image to help you understand better :
5-bar-fractals.png
The problem is that backtrader doesn't wait for the fifth candle to close, it enters the position right at the opening of the fourth candle and knows in advance that this will be a fractal before the fifth candle closes. So it gives very good results for my backtesting but this is cheating and i don't know how to specify backtrader to enters only 2 candlesticks away from the signal.

Here are the important parts of my code (If full code needed, i have posted it below) :

Python:
[CODE]

self.buy_strat = bt.And(self.rsi > self.p.RSI_buy_limit, self.fractal_bullish > self.ema_middle,

                                self.fractal_bullish < self.ema_short)

        self.sell_strat = bt.And(self.rsi < self.p.RSI_sell_limit, self.fractal_bearish < self.ema_middle,

                                 self.fractal_bearish > self.ema_short)

                              

    def next(self):


        cash = self.broker.getvalue()

        long_sl_price = self.fractal_bullish

        long_tp_price = ((self.price - long_sl_price) * self.p.trades_ratio) + self.price

        qty_long = (self.p.risk * cash) / (self.price[0] - long_sl_price)



        short_sl_price = self.fractal_bearish

        short_tp_price = ((self.price - short_sl_price) * self.p.trades_ratio) + self.price

        qty_short = (self.p.risk * cash) / (self.price[0] - short_sl_price)


        if self.order:

            return



        # Check if we are in the market

        if not self.position:


            if self.buy_strat:



                self.log('BUY CREATE AT, %.2f' % self.dataclose[0])



                self.order = self.buy_bracket(limitprice=long_tp_price, stopprice=long_sl_price,

                                              exectype=bt.Order.Market, size=qty_long)


            if self.sell_strat:



                self.log('SELL CREATE AT, %.2f' % self.dataclose[0])



                self.order = self.sell_bracket(limitprice=short_tp_price, stopprice=short_sl_price,

                                              exectype=bt.Order.Market, size=qty_short)

        else:

            if len(self) >= (self.bar_executed + 24):

                self.log('CLOSING CURRENT ORDER AT: %.2f BEEN THERE FOR TOO LONG NOW. ' % self.dataclose[0])

                self.order = self.close()


Here is the full code:

Python:
from __future__ import (absolute_import, division, print_function,
                        unicode_literals)
import os.path
import sys
import matplotlib.pyplot as plt
from mplfinance.original_flavor import candlestick_ohlc
import mplfinance as mpf
from math import *
import numpy as np
import backtrader as bt
import pandas as pd
import json
import requests
import csv
import datetime as dt
from typing import *
import yfinance as yf
import quantstats
import time



start_time = time.time()


def get_binance_bars(symbol, interval, startTime, endTime):

    url = 'https://api.binance.com/api/v3/klines'

    startTime = str(int(startTime.timestamp() * 1000))
    endTime = str(int(endTime.timestamp() * 1000))
    limit = '1000'

    req_params = {'symbol' : symbol, 'interval' : interval, 'startTime' : startTime, 'endTime' : endTime, 'limit' : limit}

    df = pd.DataFrame(json.loads(requests.get(url, params=req_params).text))

    if (len(df.index) == 0):
        return None

    df = df.iloc[:, 0:6]
    df.columns = ['datetime', 'open', 'high', 'low', 'close', 'volume']

    df.open = df.open.astype('float')
    df.high = df.high.astype('float')
    df.low = df.low.astype('float')
    df.close = df.close.astype('float')
    df.volume = df.volume.astype('float')

    df['adj_close'] = df['close']

    df.index = [dt.datetime.fromtimestamp(x / 1000.0) for x in df.datetime]

    return df


df_list = []
last_datetime = dt.datetime(2021, 2, 1)
recent_datetime = dt.datetime(2021, 6, 1)
while True:
    new_df = get_binance_bars('BTCUSDT', '15m', last_datetime, recent_datetime)
    if new_df is None:
        break
    df_list.append(new_df)
    last_datetime = max(new_df.index) + dt.timedelta(0, 1)

df = pd.concat(df_list)
df.shape



class Fractal_bearish(bt.ind.PeriodN):
    '''
    References:
        [Ref 1] http://www.investopedia.com/articles/trading/06/fractals.asp
    '''
    lines = ('fractal_bearish', )

    plotinfo = dict(subplot=False, plotlinelabels=False, plot=True)

    plotlines = dict(
        fractal_bearish=dict(marker='v', markersize=4.0, color='orange',
                             fillstyle='full', ls=''),
    )
    params = (
        ('period', 5),
        ('bardist', 0.0008),  # distance to max/min in absolute perc
        ('shift_to_potential_fractal', 2),
    )

    def next(self):
        # A bearish turning point occurs when there is a pattern with the
        # highest high in the middle and two lower highs on each side. [Ref 1]

        last_five_highs = self.data.high.get(size=self.p.period)
        max_val = max(last_five_highs)
        max_idx = last_five_highs.index(max_val)

        if max_idx == self.p.shift_to_potential_fractal:
            self.lines.fractal_bearish[-2] = max_val * (1 + self.p.bardist)


class Fractal_bullish(bt.ind.PeriodN):
    '''
    References:
        [Ref 1] http://www.investopedia.com/articles/trading/06/fractals.asp
    '''
    lines = ('fractal_bullish', )

    plotinfo = dict(subplot=False, plotlinelabels=False, plot=True)

    plotlines = dict(

        fractal_bullish=dict(marker='^', markersize=4.0, color='blue',
                             fillstyle='full', ls='')
    )
    params = (
        ('period', 5),
        ('bardist', 0.0008),  # distance to max/min in absolute perc
        ('shift_to_potential_fractal', 2),
    )

    def next(self):
        # A bullish turning point occurs when there is a pattern with the
        # lowest low in the middle and two higher lowers on each side. [Ref 1]
        last_five_lows = self.data.low.get(size=self.p.period)
        min_val = min(last_five_lows)
        min_idx = last_five_lows.index(min_val)

        if min_idx == self.p.shift_to_potential_fractal:
            self.lines.fractal_bullish[-2] = min_val * (1 - self.p.bardist)


class TestStrategy(bt.Strategy):
    params = (
        ('risk', 0.022),
        ('stop', 0.0055),
        ('shortMiddle', 1.0094),
        ('middleLong', 1.0032),
        ('ema_short', 12),
        ('ema_middle', 50),
        ('ema_long', 200),
        ('printlog', False),
        ('trades_ratio', 2),
        ('RSI_buy_limit', 30),
        ('RSI_sell_limit', 70)
    )

    def log(self, txt, dt=None, doprint=False):
        ''' Logging function for this strategy'''
        if self.params.printlog or doprint:
            dt = dt or self.datas[0].datetime.date(0)
            print('%s, %s' % (dt.isoformat(), txt))

    def __init__(self):
        # Keep a reference to the "close" line in the data[0] dataseries
        self.dataclose = self.datas[0].close

        # To keep track of pending orders and buy price/commission
        self.order = None
        self.buyprice = None
        self.buycomm = None

        self.price = self.datas[0]

        if self.p.risk > 1 or self.p.risk < 0:
            raise ValueError('The risk parameter must be between 0 and 1. ')

        # Indicators for the plotting show

        bt.indicators.MACD(self.datas[0], plot=False)
        self.fractal_bearish = Fractal_bearish(self.datas[0], period=5, plot=True)
        self.fractal_bullish = Fractal_bullish(self.datas[0], period=5, plot=True)
        self.rsi = bt.indicators.RSI(self.datas[0], period=14, plot=True)
        self.ema_short = bt.indicators.ExponentialMovingAverage(self.datas[0], period=self.p.ema_short)
        self.ema_middle = bt.indicators.ExponentialMovingAverage(self.datas[0], period=self.p.ema_middle)
        self.ema_long = bt.indicators.ExponentialMovingAverage(self.datas[0], period=self.p.ema_long)

        # Strategy parts
        self.ema_low = bt.indicators.ExponentialMovingAverage(self.datas[0], period=12, plot=False)


        self.ema_high = bt.indicators.ExponentialMovingAverage(self.datas[0], period=26, plot=False)


        self.macd_line = self.ema_low - self.ema_high

        self.macd_signal = bt.indicators.ExponentialMovingAverage(self.macd_line, period=9, plot=False)
        self.crossover_macd = bt.indicators.CrossOver(self.macd_line, self.macd_signal)
        self.crossover = bt.indicators.CrossOver(self.ema_short, self.price)

        self.macd_crossover = bt.indicators.CrossOver(self.macd_line, self.macd_signal)

        self.ema_crossover = bt.indicators.CrossOver(self.ema_short, self.ema_middle)


        self.buy_strat = bt.And(self.rsi > self.p.RSI_buy_limit, self.fractal_bullish > self.ema_middle,
                                self.fractal_bullish < self.ema_short)
        self.sell_strat = bt.And(self.rsi < self.p.RSI_sell_limit, self.fractal_bearish < self.ema_middle,
                                 self.fractal_bearish > self.ema_short)

        self.buy_signal = bt.And(self.crossover == 1, (self.ema_short / self.ema_middle) >= self.p.shortMiddle, (self.ema_middle / self.ema_long) >= self.p.middleLong)

        self.sell_signal = bt.And(self.crossover == -1, (self.ema_short / self.ema_middle) <= (1 - (self.p.shortMiddle - 1)), (
                    self.ema_middle / self.ema_long) <= (1 - (self.p.middleLong - 1)))


        #long_sl_price = self.fractal_bullish[0]
        #long_tp_price = ((self.price[0] - long_sl_price) * self.p.trades_ratio) + self.price[0]
        #qty = (self.p.risk * cash) / (self.price[0] - long_sl_price)

        #commission = 0.0008
        #leverage = 15
        #balance = self.broker.getvalue()
        #short_sl_price = self.fractal_bearish[0]
        #short_tp_price = ((self.price[0] - short_sl_price) * self.p.trades_ratio) + self.price[0]
        #qty = (self.p.risk * cash) / (self.price[0] - short_sl_price)
        #margin = (qty * self.price[0]) / leverage
        #percentage_used = margin / balance
        #leverage_net = percentage_used * leverage
        #risk_brut_real = (commission * leverage_net) / (1 - (0.64 / self.p.trades_ratio))
        #risk_net_real = risk_brut_real - commission * leverage
        #risk_net = leverage * (((margin * (self.price[0] - short_sl_price)) / cash) - commission)



    def notify_order(self, order):

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

        leverage = self.broker.getcommissioninfo(self.data).get_leverage()
        es = order.executed.size

        balance = self.broker.getvalue()
        margin = abs(es / leverage)
        # 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():

                long_sl_price = self.fractal_bullish[0]
                long_tp_price = ((self.price[0] - long_sl_price) * self.p.trades_ratio) + self.price[0]


                self.log('BUY EXECUTED, Price: %.4f, Cost: %.2f, Comm: %.2f, Size: %.2f, Balance: %.2f, Margin: %.2f, SL: %.2f, TP: %.2f' %
                    (order.executed.price,
                     order.executed.value,
                     order.executed.comm,
                     order.executed.size,
                     balance,
                     margin,
                     long_sl_price,
                     long_tp_price))

            elif order.issell():

                short_sl_price = self.fractal_bearish[0]
                short_tp_price = ((self.price[0] - short_sl_price) * self.p.trades_ratio) + self.price[0]

                self.log('SELL EXECUTED, Price: %.4f, Cost: %.2f, Comm: %.2f, Size: %.2f, Balance: %.2f, Margin: %.2f, SL: %.2f, TP: %.2f' %
                         (order.executed.price,
                          order.executed.value,
                          order.executed.comm,
                          order.executed.size,
                          balance,
                          margin,
                          short_sl_price,
                          short_tp_price))

            self.bar_executed = len(self)

        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log('Order Canceled/Margin/Rejected, Size: %.2f, Margin: %.2f' % (order.executed.size, margin))

        # Write down: no pending order
        self.order = None


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

        self.log('OPERATION PROFIT, GROSS %.2f, NET %.2f' %
                 (trade.pnl, trade.pnlcomm))

    def print_signal(self):
        self.log(
            f"o {self.datas[0].open[0]:7.2f} "
            f"h {self.datas[0].high[0]:7.2f} "
            f"l {self.datas[0].low[0]:7.2f} "
            f"c {self.datas[0].close[0]:7.2f} "
            f"v {self.datas[0].volume[0]:7.0f} "
        )


    def next(self):

        cash = self.broker.getvalue()
        long_sl_price = self.fractal_bullish
        long_tp_price = ((self.price - long_sl_price) * self.p.trades_ratio) + self.price
        qty_long = (self.p.risk * cash) / (self.price[0] - long_sl_price)

        short_sl_price = self.fractal_bearish
        short_tp_price = ((self.price - short_sl_price) * self.p.trades_ratio) + self.price
        qty_short = (self.p.risk * cash) / (self.price[0] - short_sl_price)

        # Check if an order is pending ... if yes, we cannot send a 2nd one
        if self.order:
            return

        # Check if we are in the market
        if not self.position:

            if self.buy_strat:

                self.log('BUY CREATE AT, %.2f' % self.dataclose[0])

                self.order = self.buy_bracket(limitprice=long_tp_price, stopprice=long_sl_price,
                                              exectype=bt.Order.Market, size=qty_long)



            if self.sell_strat:

                self.log('SELL CREATE AT, %.2f' % self.dataclose[0])

                self.order = self.sell_bracket(limitprice=short_tp_price, stopprice=short_sl_price,
                                              exectype=bt.Order.Market, size=qty_short)



        else:

            if len(self) >= (self.bar_executed + 24):

                self.log('CLOSING CURRENT ORDER AT: %.2f BEEN THERE FOR TOO LONG NOW. ' % self.dataclose[0])

                self.order = self.close()

    def stop(self):
        self.log(
            '(RSI BUY: %2f) (RSI SELL: %2f) (RISK: %2f) (WIN/LOSS RATIO: %2f) (EMA SHORT: %2f) (EMA MIDDLE: %2f) (SHORT/MIDDLE: %4f) (MIDDLE/SHORT: %4f) (STOP: %4f) Ending Value %.2f' %
            (self.params.RSI_buy_limit, self.params.RSI_sell_limit, self.params.risk, self.params.trades_ratio,
             self.params.ema_short, self.params.ema_middle, self.params.shortMiddle, self.params.middleLong, self.params.stop, self.broker.getvalue()), doprint=True)




def printTradeAnalysis(analyzer):
    '''
    Function to print the Technical Analysis results in a nice format.
    '''
    # Get the results we are interested in
    total_open = analyzer.total.open
    total_closed = analyzer.total.closed
    total_won = analyzer.won.total
    total_lost = analyzer.lost.total
    win_streak = analyzer.streak.won.longest
    lose_streak = analyzer.streak.lost.longest
    pnl_net = round(analyzer.pnl.net.total, 2)
    strike_rate = round(total_won / total_closed, 4) * 100
    # Designate the rows
    h1 = ['Total Open', 'Total Closed', 'Total Won', 'Total Lost']
    h2 = ['Strike Rate', 'Win Streak', 'Losing Streak', 'PnL Net']
    r1 = [total_open, total_closed, total_won, total_lost]
    r2 = [strike_rate, win_streak, lose_streak, pnl_net]
    # Check which set of headers is the longest.
    if len(h1) > len(h2):
        header_length = len(h1)
    else:
        header_length = len(h2)
    # Print the rows
    print_list = [h1, r1, h2, r2]
    row_format = "{:<15}" * (header_length + 1)
    print("Trade Analysis Results:")
    for row in print_list:
        print(row_format.format('', *row))

def printSQN(analyzer):
    sqn_list = []
    sqn = round(analyzer.sqn, 2)
    print('SQN: {}'.format(sqn))
    sqn_list.append(sqn)
    sqn_list.sort()
    print(sqn_list)


account = 1000

if __name__ == '__main__':

    cerebro = bt.Cerebro()

    strats = cerebro.optstrategy(
        TestStrategy,
        shortMiddle=np.arange(1.0094, 1.0134, 0.0004),
        middleLong=np.arange(1.005, 1.0094, 0.0004))

    data = bt.feeds.PandasData(dataname=df)

    cerebro.adddata(data)

    # Set our desired cash start
    cerebro.broker.setcash(account)

    cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="ta")
    cerebro.addanalyzer(bt.analyzers.SQN, _name="sqn")
    cerebro.addanalyzer(bt.analyzers.PyFolio, _name='PyFolio')

    # Add a FixedSize sizer according to the stake

    # Set the commission
    cerebro.broker.setcommission(commission=0.0004, leverage=10)

    print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())

    strategies = cerebro.run()
    TestStrat = strategies[0]

    final_results_list = []
    for run in strategies:
        for strategy in run:
            value = round(strategy.broker.get_value(), 2)
            PnL = round(value - account, 2)
            stop = round(strategy.params.stop, 4)
            shortMiddle = round(strategy.params.shortMiddle, 4)
            middleLong = round(strategy.params.middleLong, 4)
            ema_short = round(strategy.params.ema_short, 2)
            ema_middle = round(strategy.params.ema_middle, 2)
            final_results_list.append([stop, shortMiddle, middleLong, ema_short, ema_middle, PnL])

    by_SL_percent = sorted(final_results_list, key=lambda x: x[0])
    by_shortMiddle = sorted(final_results_list, key=lambda x: x[1])
    by_middleLong = sorted(final_results_list, key=lambda x: x[2])
    by_ema_short = sorted(final_results_list, key=lambda x: x[3])
    by_ema_middle = sorted(final_results_list, key=lambda x: x[4])
    by_PnL = sorted(final_results_list, key=lambda x: x[5], reverse=True)

    headers = ['STOP', 'SHORT/MIDDLE', 'MIDDLE/LONG', 'EMA SHORT', 'EMA MIDDLE', 'PROFIT']

    print('Results: Ordered by PROFIT:')
    for result in by_PnL:
        print(
            'STOP: {}, SHORT/MIDDLE: {}, MIDDLE/LONG: {}, EMA SHORT: {}, EMA MIDDLE: {}, PnL: {}'.format(result[0],
                                                                                                         result[1],
                                                                                                         result[2],
                                                                                                         result[3],
                                                                                                         result[4],
                                                                                                         result[5]))
        with open('Crossover_5.csv', 'w') as f:
            write = csv.writer(f)
            write.writerow(headers)
            write.writerows(by_PnL)

    portfolio_stats = TestStrat.analyzers.getbyname('PyFolio')
    returns, positions, transactions, gross_lev = portfolio_stats.get_pf_items()
    returns.index = returns.index.tz_convert(None)

    #quantstats.reports.html(returns, output='test9.html', title='XRPUSDT 15 minutes fractal strategy')

    printTradeAnalysis(TestStrat.analyzers.ta.get_analysis())
    printSQN(TestStrat.analyzers.sqn.get_analysis())

    portvalue = cerebro.broker.getvalue()

    pnl = portvalue - account
    profitability = ((portvalue / account) - 1) * 100

    print('----SUMMARY----')
    print('Final Portfolio Value: ${}'.format(round(portvalue, 2)))
    print('P/L: ${}'.format(round(pnl, 2)))
    print('Profitability: {} %'.format(round(profitability, 2)))

    cerebro.plot(style='candlestick')


print("--- %s seconds ---" % (time.time() - start_time))

Im posting that here because the forum dedicated to backtrader is broken, i can't register no matter what.
Sorry for my english, it is not my native language.
For those who have a good knoweldge with this library and can help me, thanks in advance ! :)
 

New Threads

Latest posts

Buy us a coffee!

Back
Top Bottom