Open the recipe for a walk-through on how to code this

🦉

Step3 Backtesting and Analysis

Open Recipe

Recall from the previous step, here is what we defined about the signal:

Buy signal: When the TM_TRADER_GRADE shows an up-trend
Sell signal: When the TM_TRADER_GRADE shows a down-trend

EMA could be one of the notable ways to follow the trend. Therefore, we represent our signals in mathematical ways

Buy Signal: When the EMA(3) of the TM_TRADER_GRADE >= EMA(7) of the TM_TRADER_GRADE
Sell Signal: When the EMA(3) of the TM_TRADER_GRADE < EMA(7) of the TM_TRADER_GRADE

A professional and comprehensive backtesting framework must be grasped if you want to act as a professional quantitative trader. Please refer to [Simple Backtesting](##Simple Backtesting) for people with a limited background in backtesting. If you are a professional quantitative trader, please refer to the [Advanced Backtesting](##Advanced Backtesting).

3.1 Simple Backtesting

3.1.1 Backtesting

To implement a simple backtesting process, we first calculated the daily return and our signal (True: Buy Signal, False: Sell Signal).

Then, for each day, we calculate the daily portfolio return.

For example, given 2021-01-01, we received the True Signal at the start of the day, then, we will assume we earn DailyReturnPCT, 1.2837% for that day.

import pandas as pd

# Read data
data = pd.read_csv('data/TMdata.csv', index_col=0, parse_dates=True)

# Calculate daily return and Signal
data['DailyReturnPCT'] = data['Close']/data['Open']-1
data['EMA_TraderGrade_Fastline'] = data['DailyReturnPCT'].ewm(span=3).mean()
data['EMA_TraderGrade_Slowline'] = data['DailyReturnPCT'].ewm(span=7).mean()
data['Signal'] = data['EMA_TraderGrade_Fastline'] >= data['EMA_TraderGrade_Slowline']

# Calculate portfolio return
data['PortfolioReturnPCT'] = data.apply(lambda x: x['DailyReturnPCT'] if x['Signal'] else -x['DailyReturnPCT'], axis=1)

	DATE	Open	High	Low	Close	Volume	TA_GRADE	QUANT_GRADE	TM_TRADER_GRADE	DailyReturnPCT	EMA_TraderGrade_Fastline	EMA_TraderGrade_Slowline	Signal	PortfolioReturnPCT
0	2021-01-01	28933.86	29611.51	28649.46	29305.29	298.357085	95.941921	52.077850	87.169107	1.283721	87.169107	87.169107	True	1.283721
1	2021-01-02	29321.41	33271.38	28955.02	32162.01	667.482338	96.085594	52.550240	87.378523	9.687802	87.308718	87.288773	True	9.687802
2	2021-01-03	32174.75	34776.59	31942.96	33031.74	701.951087	96.124244	53.365372	87.572470	2.663548	87.459433	87.411453	True	2.663548
3	2021-01-04	33034.55	33593.47	27500.00	32005.27	987.279894	82.142057	54.996140	76.712873	-3.115768	81.727934	83.498829	False	3.115768

3.1.2 Analysis

Here is our approximate backtesting result

The average daily return will be 0.6014%, or 24.54% annual return rate.

data['PortfolioReturnPCT'].mean()
# 0.6014

The compound annual return calculation: (1+0.0006014)^365 - 1 = 0.2454

The total return will be 449.22%

data['PortfolioReturnPCT'].sum()
# 449.22

Amazing! We successfully built a profitable trading strategy!

However, we should admit that the simple backtesting process was inaccurate because it ignored many factors, for example, the commission fee. Please move to the next section if you want an accurate backtesting result.

3.2 Advanced Backtesting

3.2.1 Building Signal

We introduced a professional backtesting framework to simulate live trading: Backtrader.

Here is the main logic for the strategy:

class EMACloseSignal(bt.Indicator):
    lines = ('trader_grade_A', 'trader_grade_B', 'signal')
    params = (('period_A', None),('period_B', None),)
    
    def __init__(self):
        self.lines.trader_grade_A = bt.talib.EMA(self.datas[0].trader_grade, timeperiod=self.p.period_A, subplot=True)
        self.lines.trader_grade_B = bt.talib.EMA(self.datas[0].trader_grade, timeperiod=self.p.period_B, subplot=True)
        self.lines.signal = self.lines.trader_grade_A > self.lines.trader_grade_B

We defined the rule:

When the faster EMA line, trader_grade_A, of the trader grade crosses over the slower EMA line, trader_grade_B, we should be in the Long position. Otherwise, we should hold the Short position.

3.2.2 Building Trading Logic

Here is the Long-Short strategy

def next(self):
    
    # ......

    # We are not in the market
    if not self.position:
        if self.signal[0]: # Buy signal
            self.log(f"BUY CREATE, {self.dataclose[0]}")
            self.order = self.buy(size=size) # Long position

        else: # Sell signal
            self.log(f"SELL CREATE, {self.dataclose[0]}")
            self.order = self.sell(size=size) # Short position

    # We in a short position but given a Buy signal
    elif self.signal[0] and self.position.size < 0:

        self.log(f"SELL Close, {self.dataclose[0]}")
        self.order = self.close() # we close our current position

    # We in a long position but given a Sell signal
    elif not self.signal[0] and self.position.size > 0:
            
        self.log(f"BUY Close, {self.dataclose[0]}")
        self.order = self.close()

Here is the explanation for the above code:

  if we do not hold position:
    
      if it is a buy signal:
          we buy!!! --> Long position

      else it is a sell signal:
          we sell!!! --> Short position

  else if the signal is buy but we hold the short position :

      we closed our short position

  else if the signal is sell but we hold the long position:

      we closed our long position

It is straightforward, right? Let's run it and see our potential ROI with the strategy.

3.2.3 Backtesting parameter

Before we start our backtesting, we still need to set some parameters to simulate a live trading environment.

Starting cash will be set with $100,000

cerebro.broker.setcash(100000.0)

The commission will be set with 0.04% referred to as the USD-M Futures Trading - USDT Maker/Taker from Binance

cerebro.broker.setcommission(commission=0.0004)

3.2.4 Analysis

Wow, pretty good! Our capital increased from $100,000 to $173,332.71 over nearly two years.

Starting Portfolio Value: 100000.00
Final Portfolio Value: 173332.71
Sharpe Ratio: 4.084129867662106
DrawDown: 40.249658800933624

Sharpe Ratio reached almost 4.1!

Now that we can say this trading strategy works, what will be next?

We just need to implement it in our real account and prepare to watch the ROI move!

Appendix

Backtesting.py

import backtrader as bt
import backtrader.feeds as btfeeds


class TM_API_data(btfeeds.GenericCSVData):
    # add only three columns called date, Bitcoin, Ethereum
    lines = ('ta_grade','quant_grade','trader_grade')

    params = (
        ('nullvalue', float('NaN')),
        ('dtformat', '%Y-%m-%d'),
        # ('tmformat', '%H:%M:%S'),

        ('datetime', 0),
        # ('time', 0),
        ('open', 1),
        ('high', 2),
        ('low', 3),
        ('close', 4),
        ('volume', 5),
        ('openinterest', -1),
        ('ta_grade', 6),
        ('quant_grade', 7),
        ('trader_grade', 8),
    )

class EMACloseSignal(bt.Indicator):
    lines = ('trader_grade_A', 'trader_grade_B', 'signal')
    params = (('period_A', None),('period_B', None),)
    
    def __init__(self):
        self.lines.trader_grade_A = bt.talib.EMA(self.datas[0].trader_grade, timeperiod=self.p.period_A, subplot=True)
        self.lines.trader_grade_B = bt.talib.EMA(self.datas[0].trader_grade, timeperiod=self.p.period_B, subplot=True)
        self.lines.signal = self.lines.trader_grade_A > self.lines.trader_grade_B



class TMStrategy(bt.Strategy):
    params = (('period_A', None),('period_B', None),)

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

    def __init__(self):
        """Initialize the strategy"""
        self.dataclose = self.datas[0].close

        # for single strategy
        self.myindicator = EMACloseSignal(self.datas[0], period_A=self.p.period_A, period_B=self.p.period_B, plot=True)
        self.signal = self.myindicator.lines.signal
        

        # To keep track of pending orders
        self.order = None
        self.buyprice = None
        self.buycomm = None

    def calculate_size(self,slipage = 0.1):
        # full position
        cash = self.broker.get_cash()
        size = cash/(self.data.close[0] * (1+slipage))
        return size

    def next(self):
        """Run once per data bar"""
        size = self.calculate_size(slipage=0)
        self.log(f"Close, {self.dataclose[0]}, ta_grade, {self.datas[0].trader_grade[0]}, signal, {self.signal[0]}, tg_A, {self.myindicator.lines.trader_grade_A[0]}, tg_B, {self.myindicator.lines.trader_grade_B[0]}")

        # Check if an order is pending
        if self.order:
            return

        # We are not in the market
        if not self.position:
            if self.signal[0]: # Buy signal
                self.log(f"BUY CREATE, {self.dataclose[0]}")
                self.order = self.buy(size=size) # Long position

            else: # Sell signal
                self.log(f"SELL CREATE, {self.dataclose[0]}")
                self.order = self.sell(size=size) # Short position

        # We in a short position but given a Buy signal
        elif self.signal[0] and self.position.size < 0:

            self.log(f"SELL Close, {self.dataclose[0]}")
            self.order = self.close() # we close our current position

        # We in a long position but given a Sell signal
        elif not self.signal[0] and self.position.size > 0:
                
            self.log(f"BUY Close, {self.dataclose[0]}")
            self.order = self.close()


    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

        # 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(
                    'BUY EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
                    (order.executed.price,
                     order.executed.value,
                     order.executed.comm))
                
                self.buyprice = order.executed.price
                self.buycomm = order.executed.comm
                
            elif order.issell() : # Sell
                self.log('SELL EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
                    (order.executed.price,
                     order.executed.value,
                     order.executed.comm))

            elif order.isclose():
                self.log('Order closed EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
                    (order.executed.price,
                     order.executed.value,
                     order.executed.comm))
            self.bar_executed = len(self)

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

        # 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 runstart():
    cerebro = bt.Cerebro(cheat_on_open=True)
    cerebro.addstrategy(TMStrategy, period_A=3, period_B=7)
    # cerebro.optstrategy(TMStrategy, period_A=range(2, 10), period_B=range(2, 10))

    cerebro.addanalyzer(bt.analyzers.PyFolio, _name='pyfolio')
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='SharpeRatio')
    cerebro.addanalyzer(bt.analyzers.Returns, _name='Returns')
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='DrawDown')

    cerebro.broker.setcash(100000.0)
    cerebro.adddata(TM_API_data(dataname='data/TMdata.csv'))
    cerebro.broker.setcommission(commission=0.0004)

    print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
    result = cerebro.run()
    print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())

    # # for optimization
    # for strat in result:
    #     print('Parameters:', strat[0].p.period_A, strat[0].p.period_B)
    #     print('Sharpe Ratio:', strat[0].analyzers.SharpeRatio.get_analysis()['sharperatio'])
    #     print('Returns:', strat[0].analyzers.Returns.get_analysis()['rtot'])

    # for single strategy
    strat = result[0]
    print('Sharpe Ratio:', strat.analyzers.SharpeRatio.get_analysis()['sharperatio'])
    print('Returns:', strat.analyzers.Returns.get_analysis()['rtot'])
    print('DrawDown:', strat.analyzers.DrawDown.get_analysis()['max']['drawdown'])

    # pyfoliozer 
    pyfoliozer = strat.analyzers.getbyname('pyfolio')
    returns, positions, transactions, gross_lev = pyfoliozer.get_pf_items()
    returns.to_csv('result/returns.csv')
    positions.to_csv('result/positions.csv')
    transactions.to_csv('result/transactions.csv')
    gross_lev.to_csv('result/gross_lev.csv')

    cerebro.plot() 
    

if __name__ == '__main__':
    runstart()