VectorBT Pro - Custom Simulator 1: Basic Candlestick Strategy

Importing the Dependencies

import vectorbtpro as vbt
import numpy as np
import pandas as pd
from numba import njit
import talib
import datetime as dt
import time
from collections import namedtuple
import itertools
import math
from vectorbtpro.records.nb import col_map_nb
from vectorbtpro.portfolio import nb as pf_nb, enums as pf_enums
import plotly.io as pio
from numba import njit

Strategy Rules 📏

In this strategy we will go long 🟩 on a Bullish Candle and go short 🟥 on a Bearish Candle.

If already in a long position and a bearish candle is present the long position should be closed and vice versa.

Otherwise the position should be held.

Obtaining The Data

data = vbt.YFData.fetch("BTC-USD", end="2022-01-01")
Open = data.get("Open").to_numpy()
High = data.get("High").to_numpy()
Low = data.get("Low").to_numpy()
Close = data.get("Close").to_numpy()

Coding The Rules

A Bullish Candle 🟩 is where the close is greater the the open (The price increased)

df['IsBull'] = df['Close'] > df['Open']

Converting the above piece of code to a NumPY Array

IsBull = df['IsBull'].to_numpy()

Understanding the Custom Simulator

Bullish Candle 🟩 & Not In A Position ⚪

IsBullArray[i] and exec_state.position == 0

---

Bearish Candle 🟥 & Not In A Position ⚪

not IsBullArray[i] and exec_state.position == 0

---

Bullish Candle 🟩 & In A Short Position 🔴

IsBullArray[i] and exec_state.position < 0

---

Bearish Candle 🟥 & In A Long Position 🟢

not IsBullArray[i] and exec_state.position > 0

---

Custom Simulator Code

def custom_simulator(open_, high_ , low_ , close_ , IsBullArray,init_cash = 10000):
     
    order_records = np.empty((2663,1), dtype = vbt.pf_enums.order_dt)
    order_counts = np.full(1, 0, dtype=np.int_)

    exec_state = vbt.pf_enums.ExecState(
                        cash = float(init_cash),
                        position = 0.0,
                        debt = 0.0,
                        locked_cash = 0.0,
                        free_cash = float(init_cash),
                        val_price = np.nan,
                        value = np.nan)
    
    
    for i in range(len(close_)):
        
        price_area = vbt.pf_enums.PriceArea(open  = open_[i],
                                            high = high_[i], 
                                            low = low_[i], 
                                            close = close_[i])
        
        value_price = price_area.close
        value = exec_state.cash + (exec_state.position * value_price)
        
        exec_state = vbt.pf_enums.ExecState(
                                cash = exec_state.cash,
                                position = exec_state.position,
                                debt = exec_state.debt,
                                locked_cash = exec_state.locked_cash,
                                free_cash = exec_state.free_cash,
                                val_price = value_price,
                                value = value)
        
        if IsBullArray[i] and exec_state.position == 0:
            #CODE THAT WILL ENTER LONG POSITION

            order_result , exec_state = enter_position(
                                                                execution_state_ = exec_state,
                                                                price_area_ = price_area,
                                                                percent_risk_ = 1,
                                                                group_ = 0, column_= 0, bar_ = i,
                                                                direction = 'Buy',
                                                                order_records_= order_records,
                                                                order_counts_ = order_counts)

        elif not IsBullArray[i] and exec_state.position == 0:
            #CODE THAT WILL ENTER SHORT POSITION
            order_result , exec_state = enter_position(
                                                                execution_state_ = exec_state,
                                                                price_area_ = price_area,
                                                                percent_risk_ = 1,
                                                                group_ = 0, column_= 0, bar_ = i,
                                                                direction = 'Sell',
                                                                order_records_= order_records,
                                                                order_counts_ = order_counts)
               

        elif IsBullArray[i] and exec_state.position < 0:
            #CODE THAT WILL CLOSE THE SHORT POSITION HERE
            order_result , exec_state = close_full_position(
                                                                execution_state_ = exec_state,
                                                                price_area_ = price_area,
                                                                group_ = 0, column_= 0, bar_ = i,
                                                                order_records_ = order_records,
                                                                order_counts_  = order_counts)    
            

        elif not IsBullArray[i] and exec_state.position >0:
            order_result , exec_state = close_full_position(
                                                                execution_state_ = exec_state,
                                                                price_area_ = price_area,
                                                                group_ = 0, column_= 0, bar_ = i,
                                                                order_records_= order_records,
                                                                order_counts_ = order_counts)    
        else:
            continue
        
                
    return vbt.nb.repartition_nb(order_records, order_counts)

Running The Custom Simulator

custom_simulator(open_ = Open, 
                 high_ = High, 
                 low_  = Low, 
                 close_  = Close, 
                 IsBullArray = IsBull,
                 init_cash = 100000)

Output of Custom Simulation / Backtest

array([(   0, 0,    0, 2.18658566,   457.33401489, 0., 1),
       (   1, 0,    3, 2.18658566,   408.9039917 , 0., 0),
       (   2, 0,    4, 2.51004568,   398.8210144 , 0., 1), ...,
       (1818, 0, 2659, 0.0218352 , 47588.85546875, 0., 1),
       (1819, 0, 2661, 0.0218352 , 47178.125     , 0., 0),
       (1820, 0, 2662, 0.02244184, 46306.4453125 , 0., 1)],
      dtype={'names':['id','col','idx','size','price','fees','side'], 'formats':['<i4','<i4','<i4','<f8','<f8','<f8','<i4'], 'offsets':[0,4,8,16,24,32,40], 'itemsize':48, 'aligned':True})

Summary / Key Points 💡

---

The custom simulator is relatively simple to understand , all we do is iterate through the length of the closing_prices.

---

Upon each iteration we update the execution_state

---

We then index into the IsBullArray along with checking the value of the exec_state.position as shown above to match the logic of our strategy.

---

After we use our prebuilt functions enter_position , close_position to enter and close positions as per the strategy has defined.