Generation¶

Signals are an additional level of abstraction added on top of orders: instead of specifying every bit of information on what needs to be ordered at each timestamp, we can first decide on what a typical order should look like, and then choose the timing of issuing such an order. The latter decision process can be realized through signals, which in the vectorbt's world are represented by a boolean mask where True means "order" and False means "no order". Additionally, we can change the meaning of each signal statically, or dynamically based on the current simulation state; for example, we can instruct the simulator to ignore an "order" signal if we're already in the market, which cannot be done by using the "from-orders" method alone. Finally, vectorbt loves data science, and so comparing multiple strategies with the same trading conditions but different signal permutations (i.e., order timings and directions) is much easier, less error-prone, and generally leads to fairer experiments.

Since we constantly buy and sell things, the ideal scenario would be to incorporate an order direction into each signal as well. But we cannot represent three states ("order to buy", "order to sell", and "no order") by using booleans - a data type with just two values. Thus, signals are usually distributed across two or more boolean arrays, where each array represents a different decision dimension. The most popular way to define signals is by using two direction-unaware arrays: entries and exits. Those two arrays have a different meaning based on the direction specified using a separate variable. For instance, when only the long direction is enabled, an entry signal opens a new long position and an exit signal closes it; when both directions are enabled, an entry signal opens a new long position and an exit signal reverses it to open a short one. To better control the decision on whether to reverse the current position or just close it out, we can define four direction-aware arrays: long entries, long exits, short entries, and short exits, which guarantees the most flexibility.

For example, to open a long position, close it, open a short position, and reverse it, the signals would look like this:

The same strategy can be also defined using an entry signal, an exit signal, and a direction:

But why not choosing an integer data type where a positive number means "order to buy", negative number means "order to sell", and zero means "no order", like done in backtrader, for example? Boolean arrays are much easier to generate and maintain by the user, but also, a boolean NumPy array requires 8x less memory than a 64-bit signed integer NumPy array. Furthermore, it's so much more convenient to combine and analyze masks than integer arrays! For example, we can use the logical OR (| in NumPy) operation to combine two masks, or sum the elements in a mask to get the number of signals since booleans are a subtype of integers and behave just like regular integers in most math expressions.

Comparison¶

Generating signals properly can sometimes be orders of magnitude more difficult than simulating them. This is because we have to take into account not only their distribution, but also how they interact across multiple boolean arrays. For example, setting both an entry and an exit at the same timestamp will effectively eliminate both. That's why vectorbt deploys numerous functions and techniques to support us in this regard.

Signal generation usually starts with comparing two or more numeric arrays. Remember that by comparing entire arrays, we're iterating over each row and column (= element) in a vectorized manner, and compare their scalar values at that one element. So, essentially, we're just running the same comparison operation on each single element across all the arrays that are being compared together. Let's start our first example with Bollinger Bands run on two separate assets. At each timestamp, we'll place a signal whenever the low price is below the lower band, with an expectation that the price will reverse back to its rolling mean:

>>> from vectorbtpro import *

>>> data = vbt.BinanceData.pull(
...     ["BTCUSDT", "ETHUSDT"], 
...     start="2021-01-01",
...     end="2022-01-01"
... )
>>> data.get("Low")
symbol                      BTCUSDT  ETHUSDT
Open time                                   
2021-01-01 00:00:00+00:00  28624.57   714.29
2021-01-02 00:00:00+00:00  28946.53   714.91
2021-01-03 00:00:00+00:00  31962.99   768.71
...                             ...      ...
2021-12-29 00:00:00+00:00  46096.99  3604.20
2021-12-30 00:00:00+00:00  45900.00  3585.00
2021-12-31 00:00:00+00:00  45678.00  3622.29

[365 rows x 2 columns]

>>> bb = vbt.talib("BBANDS").run(
...     data.get("Close"),
...     timeperiod=vbt.Default(14),  # (1)!
...     nbdevup=vbt.Default(2),
...     nbdevdn=vbt.Default(2)
... )
>>> bb.lowerband  # (2)!
symbol                          BTC-USD      ETH-USD
Date                                                
2021-04-23 00:00:00+00:00           NaN          NaN
2021-04-24 00:00:00+00:00           NaN          NaN
2021-04-25 00:00:00+00:00           NaN          NaN
...                                 ...          ...
2022-04-21 00:00:00+00:00  38987.326323  2912.894415
2022-04-22 00:00:00+00:00  38874.059308  2898.681307
2022-04-23 00:00:00+00:00  38915.417003  2903.756905

[366 rows x 2 columns]

>>> mask = data.get("Low") < bb.lowerband  # (3)!
>>> mask
symbol                     BTCUSDT  ETHUSDT
Open time                                  
2021-01-01 00:00:00+00:00    False    False
2021-01-02 00:00:00+00:00    False    False
2021-01-03 00:00:00+00:00    False    False
...                            ...      ...
2021-12-29 00:00:00+00:00    False     True
2021-12-30 00:00:00+00:00    False     True
2021-12-31 00:00:00+00:00    False    False

[365 rows x 2 columns]

>>> mask.sum()  # (4)!
symbol
BTCUSDT    36
ETHUSDT    28
dtype: int64

1. Wrap each parameter with Default to hide its column level. Alternatively, we can pass a list of the parameters to hide with hide_params.
2. Get the lower band as a Pandas object. We can list all the output names of an indicator using bb.output_names.
3. Compare two numeric arrays element-wise
4. Get the number of signals in each column for a better overview

This operation has generated a mask that has a true value whenever the low price dips below the lower band. Such an array can already be used in simulation! But let's see what happens when we try to compare the lower band that has been generated for multiple combinations of the (upper and lower) multiplier:

>>> bb_mult = vbt.talib("BBANDS").run(
...     data.get("Close"),
...     timeperiod=vbt.Default(14),
...     nbdevup=[2, 3],
...     nbdevdn=[2, 3]  # (1)!
... )
>>> mask = data.get("Low") < bb_mult.lowerband
ValueError: Can only compare identically-labeled DataFrame objects

1. Two parameter combinations: (14, 2, 2) and (14, 3, 3)

The problem lies in Pandas being unable to compare DataFrames with different columns - the left DataFrame contains the columns BTCUSDT and ETHUSDT while the right DataFrame coming from the Bollinger Bands indicator now contains the columns (2, 2, BTCUSDT), (2, 2, ETHUSDT), (3, 3, BTCUSDT), and (3, 3, ETHUSDT). So, what's the solution? Right - vectorbt! By appending vbt to the left operand, we are comparing the accessor object of type BaseAccessor instead of the DataFrame itself. This will trigger the so-called magic method __lt__ of that accessor, which takes the DataFrame under the accessor and the DataFrame on the right, and combines them with BaseAccessor.combine and numpy.less as combine_func. This, in turn, will broadcast the shapes and indexes of both DataFrames using the vectorbt's powerful broadcasting mechanism, effectively circumventing the limitation of Pandas.

As the result, vectorbt will compare (2, 2, BTCUSDT) and (3, 3, BTCUSDT) only with BTCUSDT and (2, 2, ETHSDT) and (3, 3, ETHSDT) only with ETHSDT, and this using NumPy - faster!

>>> mask = data.get("Low").vbt < bb_mult.lowerband  # (1)!
>>> mask
bbands_nbdevup                          2               3
bbands_nbdevdn                          2               3
symbol                    BTCUSDT ETHUSDT BTCUSDT ETHUSDT
Open time                                                
2021-01-01 00:00:00+00:00   False   False   False   False
2021-01-02 00:00:00+00:00   False   False   False   False
2021-01-03 00:00:00+00:00   False   False   False   False
...                           ...     ...     ...     ...
2021-12-29 00:00:00+00:00   False    True   False   False
2021-12-30 00:00:00+00:00   False    True   False   False
2021-12-31 00:00:00+00:00   False   False   False   False

[365 rows x 4 columns]

>>> mask.sum()
bbands_nbdevup  bbands_nbdevdn  symbol 
2               2               BTCUSDT    53
                                ETHUSDT    48
3               3               BTCUSDT    10
                                ETHUSDT     9
dtype: int64

1. vbt must be always called on the left operand

As you might have recalled from the documentation on indicators, each indicator attaches a couple of helper methods for comparison - {name}_above, {name}_equal, and {name}_below, which do basically the same as we did above:

>>> mask = bb_mult.lowerband_above(data.get("Low"))  # (1)!
>>> mask.sum()
bbands_nbdevup  bbands_nbdevdn  symbol 
2               2               BTCUSDT    53
                                ETHUSDT    48
3               3               BTCUSDT    10
                                ETHUSDT     9
dtype: int64

1. Our indicator doesn't have an input or output for the low price, thus we need to reverse the comparison order and return whether the lower band is above the low price

Thresholds¶

To compare a numeric array against two or more scalar thresholds (making them parameter combinations), we can use the same approach by either appending vbt, or by calling the method BaseAccessor.combine. Let's calculate the bandwidth of our single-combination indicator, which is the upper band minus the lower band divided by the middle band, and check whether it's higher than two different thresholds:

>>> bandwidth = (bb.upperband - bb.lowerband) / bb.middleband

>>> mask = bandwidth.vbt > vbt.Param([0.15, 0.3], name="threshold")  # (1)!
>>> mask.sum()
threshold  symbol 
0.15       BTCUSDT    253
           ETHUSDT    316
0.30       BTCUSDT     65
           ETHUSDT    136
dtype: int64

>>> mask = bandwidth.vbt.combine(
...     [0.15, 0.3],  # (2)!
...     combine_func=np.greater, 
...     keys=pd.Index([0.15, 0.3], name="threshold")  # (3)!
... )
>>> mask.sum()
threshold  symbol 
0.15       BTCUSDT    253
           ETHUSDT    316
0.30       BTCUSDT     65
           ETHUSDT    136
dtype: int64

1. Passes both objects to BaseAccessor.combine, broadcasts them using broadcast, and combines them using numpy.greater. Thanks to broadcasting, each value in vbt.Param is combined with each column in the array. This works only for scalars!
2. When passing a list, the DataFrame gets compared to each item in the list
3. Argument keys is used to append a column level describing the items in the list

The latest example works also on arrays instead of scalars. Or, we can use pandas.concat to manually stack the results of any comparison to treat them as separate combinations:

>>> mask = pd.concat(
...     (bandwidth > 0.15, bandwidth > 0.3), 
...     keys=pd.Index([0.15, 0.3], name="threshold"), 
...     axis=1
... )
>>> mask.sum()
threshold  symbol 
0.15       BTCUSDT    253
           ETHUSDT    316
0.30       BTCUSDT     65
           ETHUSDT    136
dtype: int64

Crossovers¶

So far we have touched basic vectorized comparison operations, but there is one operation that comes disproportionally often in technical analysis: crossovers. A crossover refers to a situation where two time series cross each other. There are two ways of finding the crossovers: naive and native. The naive approach compares both time series in a vectorized manner and then selects the first True value out of each "partition" of True values. A partition in the vectorbt's vocabulary for signal processing is just a bulk of consecutive True values produced by the comparison. While we already know how to do the first operation, the second one can be achieved with the help of the accessor for signals - SignalsAccessor, accessible via the attribute vbt.signals on any Pandas object.

In particular, we will be using the method SignalsAccessor.first, which takes a mask, assigns a rank to each True value in each partition using SignalsAccessor.pos_rank (enumerated from 0 to the length of the respective partition), and then keeps only those True values that have the rank 0. Let's get the crossovers of the lower price dipping below the lower band:

>>> low_below_lband = data.get("Low") < bb.lowerband
>>> mask = low_below_lband.vbt.signals.first()
>>> mask.sum()
symbol
BTCUSDT    21
ETHUSDT    20
dtype: int64

To make sure that the operation was successful, let's plot the BTCUSDT column of both time series using GenericAccessor.plot and the generated signals using SignalsSRAccessor.plot_as_markers:

>>> btc_low = data.get("Low", "BTCUSDT").rename("Low")  # (1)!
>>> btc_lowerband = bb.lowerband["BTCUSDT"].rename("Lower Band")
>>> btc_mask = mask["BTCUSDT"].rename("Signals")

>>> fig = btc_low.vbt.plot()  # (2)!
>>> btc_lowerband.vbt.plot(fig=fig)
>>> btc_mask.vbt.signals.plot_as_markers(
...     y=btc_low, 
...     trace_kwargs=dict(
...         marker=dict(
...             color="#DFFF00"
...         )
...     ),
...     fig=fig
... )  # (3)!
>>> fig.show()

1. Give the column another name for the legend
2. First plotting method returns a figure, which needs to be passed to each subsequent plotting method
3. We're using btc_low as the Y-values at which to place the markers

But here's the catch: if the first low value is already below the first lower band value, it will also yield a crossover signal. To fix that, we need to pass after_false=True, which will discard the first partition if there is no False value before it.

>>> mask = low_below_lband.vbt.signals.first(after_false=True)
>>> mask.sum()
symbol
BTCUSDT    21
ETHUSDT    20
dtype: int64

And here's another catch: if the first bunch of values in the indicator are NaN, which results in False values in the mask, and the first value after the last NaN yields True, then the after_false argument becomes ineffective. To account for this, we need to manually set those values in the mask to True. Let's illustrate this issue on sample data:

>>> sample_low = pd.Series([10, 9, 8, 9, 8])
>>> sample_lband = pd.Series([np.nan, np.nan, 9, 8, 9])
>>> sample_mask = sample_low < sample_lband
>>> sample_mask.vbt.signals.first(after_false=True)  # (1)!
0    False
1    False
2     True
3    False
4     True
dtype: bool

>>> sample_mask[sample_lband.ffill().isnull()] = True  # (2)!
>>> sample_mask.vbt.signals.first(after_false=True)
0    False
1    False
2    False
3    False
4     True
dtype: bool

1. The first crossover shouldn't happen because we don't know what happens at those NaN, while the second crossover is perfectly valid
2. Forward fill the indicator values to keep only the first NaN values, and set the mask to True at those positions to make after_false effective again

Or, we can remove the buffer, do the operation, and then add the buffer back:

>>> buffer = sample_lband.ffill().isnull().sum(axis=0).max()  # (1)!
>>> buffer
2

>>> sample_buf_mask = sample_low.iloc[buffer:] < sample_lband.iloc[buffer:]
>>> sample_buf_mask = sample_buf_mask.vbt.signals.first(after_false=True)
>>> sample_mask = sample_low.vbt.wrapper.fill(False)
>>> sample_mask.loc[sample_buf_mask.index] = sample_buf_mask
>>> sample_mask
0    False
1    False
2    False
3    False
4     True
dtype: bool

1. Find the maximum length of each first consecutive series of NaN values across all columns

But here comes another issue: what happens if our data contains gaps and we encounter a NaN in the middle of a partition? We should make the second part of the partition False as forward-filling that NaN value would make waiting for a confirmation problematic. But also, doing so many operations on bigger arrays just for getting the crossovers is quite resource-expensive. Gladly, vectorbt deploys its own Numba-compiled function crossed_above_nb for finding the crossovers in an iterative manner, which is the second, native way. To use this function, we can use the methods GenericAccessor.crossed_above and GenericAccessor.crossed_below, accessible via the attribute vbt on any Pandas object:

>>> mask = data.get("Low").vbt.crossed_below(bb.lowerband, wait=1)  # (1)!
>>> mask.sum()
symbol
BTCUSDT    15
ETHUSDT    11
dtype: int64

1. Wait one bar for confirmation

As with other comparison methods, each indicator has the helper methods {name}_crossed_above and {name}_crossed_below for generating the crossover masks:

>>> mask = bb.lowerband_crossed_above(data.get("Low"), wait=1)
>>> mask.sum()
symbol
BTCUSDT    15
ETHUSDT    11
dtype: int64

Logical operators¶

Once we've generated two or more masks (conditions), we can combine them into a single mask using logical operators. Common logical operators include

• AND (& or numpy.logical_and): for each element, returns True whenever all the conditions are True
• OR (| or numpy.logical_or): for each element, returns True whenever any of the conditions are True
• NOT (~ or numpy.logical_not): for each element, returns True whenever the condition is False
• XOR (^ or numpy.logical_xor): for each element, returns True whenever only one of the conditions is True

For example, let's combine four conditions for a signal: the low price dips below the lower band AND the bandwidth is above some threshold (= a downward breakout while expanding), OR, the high price rises above the upper band AND the bandwidth is below some threshold (= an upward breakout while squeezing):

>>> cond1 = data.get("Low") < bb.lowerband
>>> cond2 = bandwidth > 0.3
>>> cond3 = data.get("High") > bb.upperband
>>> cond4 = bandwidth < 0.15

>>> mask = (cond1 & cond2) | (cond3 & cond4)
>>> mask.sum()
symbol
BTCUSDT    25
ETHUSDT    13
dtype: int64

To test multiple thresholds and to broadcast exclusively using vectorbt:

>>> cond1 = data.get("Low").vbt < bb.lowerband
>>> cond2 = bandwidth.vbt > vbt.Param([0.3, 0.3, 0.4, 0.4], name="cond2_th")  # (1)!
>>> cond3 = data.get("High").vbt > bb.upperband
>>> cond4 = bandwidth.vbt < vbt.Param([0.1, 0.2, 0.1, 0.2], name="cond4_th")  # (2)!

>>> mask = (cond1.vbt & cond2).vbt | (cond3.vbt & cond4)  # (3)!
>>> mask.sum()
cond2_th  cond4_th  symbol 
0.3       0.1       BTCUSDT    11
                    ETHUSDT    10
          0.2       BTCUSDT    28
                    ETHUSDT    27
0.4       0.1       BTCUSDT     9
                    ETHUSDT     5
          0.2       BTCUSDT    26
                    ETHUSDT    22
dtype: int64

1. Test two thresholds in the second condition, and repeat them to compare each one to both thresholds in the fourth condition
2. Test two thresholds in the fourth condition, and tile them to compare each one to both thresholds in the second condition
3. Notice how vbt is appended to each operand on the left. Adding it to any operand on the right is redundant.

Cartesian product¶

Combining two or more arrays using a Cartesian product is a bit more complex since every array has the column level symbol that shouldn't be combined with itself. But here's the trick. First, convert the columns of each array into their integer positions. Then, split each position array into "blocks" (smaller arrays). Blocks will be combined with each other, but the positions within each block won't; that is, each block acts as a parameter combination. Combine then all blocks using a combinatorial function of choice (see itertools for various options, or generate_param_combs), and finally, flatten each array with blocks and use it for column selection. Sounds complex? Yes. Difficult to implement? No!

>>> cond1 = data.get("Low").vbt < bb.lowerband
>>> cond2 = bandwidth.vbt > vbt.Param([0.3, 0.4], name="cond2_th")  # (1)!
>>> cond3 = data.get("High").vbt > bb.upperband
>>> cond4 = bandwidth.vbt < vbt.Param([0.1, 0.2], name="cond4_th")

>>> i1 = np.split(np.arange(len(cond1.columns)), len(cond1.columns) // 2)  # (2)!
>>> i2 = np.split(np.arange(len(cond2.columns)), len(cond2.columns) // 2)
>>> i3 = np.split(np.arange(len(cond3.columns)), len(cond3.columns) // 2)
>>> i4 = np.split(np.arange(len(cond4.columns)), len(cond4.columns) // 2)

>>> i1
[array([0, 1])]
>>> i2
[array([0, 1]), array([2, 3])]
>>> i3
[array([0, 1])]
>>> i4
[array([0, 1]), array([2, 3])]

>>> i1, i2, i3, i4 = zip(*product(i1, i2, i3, i4))  # (3)!

>>> i1
(array([0, 1]), array([0, 1]), array([0, 1]), array([0, 1]))
>>> i2
(array([0, 1]), array([0, 1]), array([2, 3]), array([2, 3]))
>>> i3
(array([0, 1]), array([0, 1]), array([0, 1]), array([0, 1]))
>>> i4
(array([0, 1]), array([2, 3]), array([0, 1]), array([2, 3]))

>>> i1 = np.asarray(i1).flatten()  # (4)!
>>> i2 = np.asarray(i2).flatten()
>>> i3 = np.asarray(i3).flatten()
>>> i4 = np.asarray(i4).flatten()

>>> i1
[0 1 0 1 0 1 0 1]
>>> i2
[0 1 0 1 2 3 2 3]
>>> i3
[0 1 0 1 0 1 0 1]
>>> i4
[0 1 2 3 0 1 2 3]

>>> cond1 = cond1.iloc[:, i1]  # (5)!
>>> cond2 = cond2.iloc[:, i2]
>>> cond3 = cond3.iloc[:, i3]
>>> cond4 = cond4.iloc[:, i4]

>>> mask = (cond1.vbt & cond2).vbt | (cond3.vbt & cond4)  # (6)!
>>> mask.sum()
cond2_th  cond4_th  symbol 
0.3       0.1       BTCUSDT    11
                    ETHUSDT    10
          0.2       BTCUSDT    28
                    ETHUSDT    27
0.4       0.1       BTCUSDT     9
                    ETHUSDT     5
          0.2       BTCUSDT    26
                    ETHUSDT    22
dtype: int64

1. Each DataFrame should contain only unique parameter combinations (and columns in general)
2. Split each position array into blocks with symbols
3. Combine all blocks (= parameter combinations) using the Cartesian product
4. Flatten each array with blocks to get an array that can be used in indexing
5. Using the final positions, select the columns from each DataFrame, which will make each DataFrame have the same number of columns
6. All columns have the same length but still different labels let the vectorbt's broadcaster do the alignment job

In newer versions of VBT the same effect can be achieved with a single call of BaseAccessor.cross:

>>> cond1 = data.get("Low").vbt < bb.lowerband
>>> cond2 = bandwidth.vbt > vbt.Param([0.3, 0.4], name="cond2_th")
>>> cond3 = data.get("High").vbt > bb.upperband
>>> cond4 = bandwidth.vbt < vbt.Param([0.1, 0.2], name="cond4_th")

>>> cond1, cond2, cond3, cond4 = vbt.pd_acc.x(cond1, cond2, cond3, cond4)  # (1)!
>>> mask = (cond1.vbt & cond2).vbt | (cond3.vbt & cond4)

1. x is an alias method for cross. Another way: cond1.vbt.x(cond2, cond3, cond4).

But probably an easier and less error-prone approach would be to build an indicator that would handle parameter combinations for us

For this, we will write an indicator expression similar to the code we wrote for a single parameter combination, and use IndicatorFactory.from_expr to auto-build an indicator by parsing that expression. The entire logic including the specification of all inputs, parameters, and outputs is encapsulated in the expression itself. We'll use the annotation @res_talib_bbands to resolve the specification of the inputs and parameters expected by the TA-Lib's BBANDS indicator and "copy" them over to our indicator by also prepending the prefix talib to the parameter names. Then, we will perform our usual signal generation logic by substituting the custom parameters cond2_th and cond4_th with their single values, and return the whole thing as an output mask annotated accordingly.

>>> MaskGenerator = vbt.IF.from_expr("""
... upperband, middleband, lowerband = @res_talib_bbands
... bandwidth = (upperband - lowerband) / middleband
... cond1 = low < lowerband
... cond2 = bandwidth > @p_cond2_th
... cond3 = high > upperband
... cond4 = bandwidth < @p_cond4_th
... @out_mask:(cond1 & cond2) | (cond3 & cond4)
... """)

>>> vbt.phelp(MaskGenerator.run, incl_doc=False)  # (1)!
Indicator.run(
    high,
    low,
    close,
    cond2_th,
    cond4_th,
    bbands_timeperiod=Default(value=5),
    bbands_nbdevup=Default(value=2),
    bbands_nbdevdn=Default(value=2),
    bbands_matype=Default(value=0),
    bbands_timeframe=Default(value=None),
    short_name='custom',
    hide_params=None,
    hide_default=True,
    **kwargs
)

>>> mask_generator = MaskGenerator.run(
...     high=data.get("High"),
...     low=data.get("Low"),
...     close=data.get("Close"),
...     cond2_th=[0.3, 0.4],
...     cond4_th=[0.1, 0.2],
...     bbands_timeperiod=vbt.Default(14),
...     param_product=True
... )  # (2)!
>>> mask_generator.mask.sum()
custom_cond2_th  custom_cond4_th  symbol 
0.3              0.1              BTCUSDT    11
                                  ETHUSDT    10
                 0.2              BTCUSDT    28
                                  ETHUSDT    27
0.4              0.1              BTCUSDT     9
                                  ETHUSDT     5
                 0.2              BTCUSDT    26
                                  ETHUSDT    22
dtype: int64

1. Take a look at the arguments that the run method expects
2. Run the indicator on the Cartesian product of all parameters

Shifting¶

To compare the current value to any previous (not future!) value, we can use forward shifting. Also, we can use it to shift the final mask to postpone the order execution. For example, let's generate a signal whenever the low price dips below the lower band AND the bandwidth change (i.e., the difference between the current and the previous bandwidth) is positive:

>>> cond1 = data.get("Low") < bb.lowerband
>>> cond2 = bandwidth > bandwidth.shift(1)  # (1)!

>>> mask = cond1 & cond2
>>> mask.sum()
symbol
BTCUSDT    42
ETHUSDT    39
dtype: int64

1. The shifted array holds the previous values

Another way to shift observations is by selecting the first observation in a rolling window. This is particularly useful when the rolling window has a variable size, for example, based on a frequency. Let's do the same as above but determine the change in the bandwidth in relation to one week ago instead of yesterday:

>>> cond2 = bandwidth > bandwidth.rolling("7d").apply(lambda x: x[0])

>>> mask = cond1 & cond2
>>> mask.sum()
symbol
BTCUSDT    33
ETHUSDT    28
dtype: int64

The approach above is a move in the right direction, but it introduces two potential issues: all windows will be either 6 days long or less, while the performance of rolling and applying such a custom Python function using Pandas is not satisfactory, to say the least. The first issue can be solved by rolling a window of 8 days, and checking the timestamp of the first observation being exactly 7 days behind the current timestamp:

>>> def exactly_ago(sr):  # (1)!
...     if sr.index[0] == sr.index[-1] - vbt.timedelta("7d"):
...         return sr.iloc[0]
...     return np.nan

>>> cond_7d_ago = bandwidth.rolling("8d").apply(exactly_ago, raw=False)
>>> cond2 = bandwidth > cond_7d_ago

>>> mask = cond1 & cond2
>>> mask.sum()
symbol
BTCUSDT    29
ETHUSDT    26
dtype: int64

1. By passing raw=False, the input will be a Pandas Series

The second issue can be solved by looping with Numba. However, the main challenge lies in solving those two issues simultaneously because we want to access the timestamp of the first observation, which requires us to work on a Pandas Series instead of a NumPy array, and Numba cannot work on Pandas Series

Thus, we will use the vectorbt's accessor method GenericAccessor.rolling_apply, which offers two modes: regular and meta. The regular mode rolls over the data of a Pandas object just like Pandas does it, and does not give us any information about the current window The meta mode rolls over the metadata of a Pandas object, so we can easily select the data from any array corresponding to the current window

>>> @njit
... def exactly_ago_meta_nb(from_i, to_i, col, index, freq, arr):  # (1)!
...     if index[from_i] == index[to_i - 1] - freq:  # (2)!
...         return arr[from_i, col]  # (3)!
...     return np.nan

>>> cond_7d_ago = vbt.pd_acc.rolling_apply(
...     "8d",
...     exactly_ago_meta_nb,
...     bandwidth.index.values,  # (4)!
...     vbt.timedelta("7d").to_timedelta64(),
...     vbt.to_2d_array(bandwidth),
...     wrapper=bandwidth.vbt.wrapper  # (5)!
... )
>>> cond2 = bandwidth > cond_7d_ago

>>> mask = cond1 & cond2
>>> mask.sum()
symbol
BTCUSDT    29
ETHUSDT    26
dtype: int64

1. The meta function must take three arguments: the current window start index, window end index, and column
2. Window end index is exclusive, thus use to_i - 1 to get the last index in the window
3. Use the current column index to select the column. Remember to make all arrays two-dimensional.
4. Here come our three user-defined arguments
5. Wrapper contains the metadata we want to iterate over and to construct the final Pandas object

And if this approach (rightfully) intimidates you, there is a dead simple method GenericAccessor.ago, which is capable of forward-shifting the array using any delta:

>>> cond2 = bandwidth > bandwidth.vbt.ago("7d")

>>> mask = cond1 & cond2
>>> mask.sum()
symbol
BTCUSDT    29
ETHUSDT    26
dtype: int64

>>> bandwidth.iloc[-8]
symbol
BTCUSDT    0.125477
ETHUSDT    0.096458
Name: 2021-12-24 00:00:00+00:00, dtype: float64

>>> bandwidth.vbt.ago("7d").iloc[-1]
symbol
BTCUSDT    0.125477
ETHUSDT    0.096458
Name: 2021-12-31 00:00:00+00:00, dtype: float64

Truth value testing¶

But what if we want to test whether a certain condition was met during a certain period of time in the past? For this, we need to create an expanding or a rolling window, and do truth value testing using numpy.any or numpy.all within this window. But since Pandas doesn't implement the rolling aggregation using any and all, we need to be more creative and treat booleans as integers: use max for a logical OR and min for a logical AND. Also, don't forget to cast the resulting array to a boolean data type to generate a valid mask.

Let's place a signal whenever the low price goes below the lower band AND there was a downward crossover of the close price with the middle band in the past 5 candles:

>>> cond2 = data.get("Close").vbt.crossed_below(bb.middleband)
>>> cond2 = cond2.rolling(5, min_periods=1).max().astype(bool)

>>> mask = cond1 & cond2
>>> mask.sum()
symbol
BTCUSDT    36
ETHUSDT    28
dtype: int64

If the window size is fixed, we can also use GenericAccessor.rolling_any and GenericAccessor.rolling_all, which are tailored for computing rolling truth testing operations:

>>> cond2 = data.get("Close").vbt.crossed_below(bb.middleband)
>>> cond2 = cond2.vbt.rolling_any(5)

>>> mask = cond1 & cond2
>>> mask.sum()
symbol
BTCUSDT    36
ETHUSDT    28
dtype: int64

Another way of doing the same rolling operations is by using the accessor method GenericAccessor.rolling_apply and specifying reduce_func_nb as "any" or "all" string. We should use the argument wrap_kwargs to instruct vectorbt to fill NaNs with False and change the data type. This method allows flexible windows to be passed. Again, let's roll a window of 5 days:

>>> cond2 = data.get("Close").vbt.crossed_below(bb.middleband)
>>> cond2 = cond2.vbt.rolling_apply(
...     "5d", "any",  # (1)!
...     minp=1, 
...     wrap_kwargs=dict(fillna=0, dtype=bool)
... )

>>> mask = cond1 & cond2
>>> mask.sum()
symbol
BTCUSDT    36
ETHUSDT    28
dtype: int64

1. "any" translates to any_reduce_nb

Let's do something more complex: check whether the bandwidth contracted to 10% or less at any point during a month using an expanding window, and reset the window at the beginning of the next month; this way, we make the first timestamp of the month a time anchor for our condition. For this, we'll overload the vectorbt's resampling logic, which allows aggregating values by mapping any source index (anchor points in our example) to any target index (our index).

>>> anchor_points = data.wrapper.get_index_points(  # (1)!
...     every="M", 
...     start=0,  # (2)!
...     exact_start=True
... )
>>> anchor_points
array([  0,  31,  59,  90, 120, 151, 181, 212, 243, 273, 304, 334])

>>> left_bound = np.full(len(data.wrapper.index), np.nan)  # (3)!
>>> left_bound[anchor_points] = anchor_points
>>> left_bound = vbt.dt.to_ns(vbt.nb.ffill_1d_nb(left_bound))
>>> left_bound = bandwidth.index[left_bound]
>>> left_bound
DatetimeIndex(['2021-01-01 00:00:00+00:00', '2021-01-01 00:00:00+00:00',
               '2021-01-01 00:00:00+00:00', '2021-01-01 00:00:00+00:00',
               '2021-01-01 00:00:00+00:00', '2021-01-01 00:00:00+00:00',
               ...
               '2021-12-01 00:00:00+00:00', '2021-12-01 00:00:00+00:00',
               '2021-12-01 00:00:00+00:00', '2021-12-01 00:00:00+00:00',
               '2021-12-01 00:00:00+00:00', '2021-12-01 00:00:00+00:00'],
              dtype='datetime64[ns, UTC]', ...)

>>> right_bound = data.wrapper.index  # (4)!
>>> right_bound
DatetimeIndex(['2021-01-01 00:00:00+00:00', '2021-01-02 00:00:00+00:00',
               '2021-01-03 00:00:00+00:00', '2021-01-04 00:00:00+00:00',
               '2021-01-05 00:00:00+00:00', '2021-01-06 00:00:00+00:00',
               ...
               '2021-12-26 00:00:00+00:00', '2021-12-27 00:00:00+00:00',
               '2021-12-28 00:00:00+00:00', '2021-12-29 00:00:00+00:00',
               '2021-12-30 00:00:00+00:00', '2021-12-31 00:00:00+00:00'],
              dtype='datetime64[ns, UTC]', ...)

>>> mask = (bandwidth <= 0.1).vbt.resample_between_bounds(  # (5)!
...     left_bound, 
...     right_bound,
...     "any",
...     closed_lbound=True,  # (6)!
...     closed_rbound=True,
...     wrap_kwargs=dict(fillna=0, dtype=bool)
... )
>>> mask.index = right_bound
>>> mask.astype(int).vbt.ts_heatmap().show()

1. Find the position of the first timestamp in each month using ArrayWrapper.get_index_ranges
2. Remove this and the next line to not include the first month if it's incomplete
3. The left bound of each window will be the respective anchor point (month start). Create an integer array of the same size as our index, fill the anchor points at their positions, and forward fill them.
4. The right bound of each window is the current index
5. Use GenericAccessor.resample_between_bounds to map the mask values to their windows, and aggregate each window using the "any" operation
6. Make both bounds inclusive to include every single timestamp in a month

We can observe how the signal for the bandwidth touching the 10% mark propagates through each month, and then the calculation gets reset and repeated.

Periodically¶

To set signals periodically, such as at 18:00 of each Tuesday, we have multiple options. The first approach involves comparing various attributes of the source and target datetime. For example, to get the timestamps that correspond to each Tuesday, we can compare pandas.DatetimeIndex.weekday to 1 (Monday is 0 and Sunday is 6):

>>> min_data = vbt.BinanceData.pull(  # (1)!
...     ["BTCUSDT", "ETHUSDT"], 
...     start="2021-01-01 UTC",  # (2)!
...     end="2021-02-01 UTC",
...     timeframe="1h"
... )
>>> index = min_data.wrapper.index
>>> tuesday_index = index[index.weekday == 1]
>>> tuesday_index
DatetimeIndex(['2021-01-05 00:00:00+00:00', '2021-01-05 01:00:00+00:00',
               '2021-01-05 02:00:00+00:00', '2021-01-05 03:00:00+00:00',
               '2021-01-05 04:00:00+00:00', '2021-01-05 05:00:00+00:00',
               ...
               '2021-01-26 18:00:00+00:00', '2021-01-26 19:00:00+00:00',
               '2021-01-26 20:00:00+00:00', '2021-01-26 21:00:00+00:00',
               '2021-01-26 22:00:00+00:00', '2021-01-26 23:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)

1. Fetch hourly data for a better illustration
2. Don't forget about the UTC timezone for crypto

Now, we need to select only those timestamps that happen at one specific time:

>>> tuesday_1800_index = tuesday_index[tuesday_index.hour == 18]
>>> tuesday_1800_index
DatetimeIndex(['2021-01-05 18:00:00+00:00', '2021-01-12 18:00:00+00:00',
               '2021-01-19 18:00:00+00:00', '2021-01-26 18:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)

Since each attribute comparison produces a mask, we can get our signals by pure logical operations. Let's get the timestamps that correspond to each Tuesday 17:30 by comparing the weekday of each timestamp to Tuesday AND the hour of each timestamp to 17 AND the minute of each timestamp to 30:

>>> tuesday_1730_index = index[
...     (index.weekday == 1) & 
...     (index.hour == 17) & 
...     (index.minute == 30)
... ]
>>> tuesday_1730_index
DatetimeIndex([], dtype='datetime64[ns, UTC]', name='Open time', freq='H')

As we see, both conditions combined produced no exact matches because our index is hourly. But what if we wanted to get the previous or next timestamp if there was no exact match? Clearly, the approach above wouldn't work. Instead, we'll use the function pandas.Index.get_indexer, which takes an array with index labels, and searches for their corresponding positions in the index. For example, let's get the position of August 7th in our index:

>>> index.get_indexer([vbt.timestamp("2021-01-07", tz=index.tz)])  # (1)!
array([144])

1. Don't forget to provide the timezone if the index is timezone-aware!

But looking for an index that doesn't exist will return -1:

>>> index.get_indexer([vbt.timestamp("2021-01-07 17:30:00", tz=index.tz)]) 
array([-1])

To get either the exact match or the previous one, we can pass method='ffill'. Conversely, to get the next one, we can pass method='bfill':

>>> index[index.get_indexer(
...     [vbt.timestamp("2021-01-07 17:30:00", tz=index.tz)],
...     method="ffill"
... )]
DatetimeIndex(['2021-01-07 17:00:00+00:00'], ...)

>>> index[index.get_indexer(
...     [vbt.timestamp("2021-01-07 17:30:00", tz=index.tz)],
...     method="bfill"
... )]
DatetimeIndex(['2021-01-07 18:00:00+00:00'], ...)

Returning to our example, we need first to generate the target index for our query, which we're about to search in the source index: use the function pandas.date_range to get the timestamp of each Tuesday midnight, and then add a timedelta of 17 hours and 30 minutes. Next, transform the target index into positions (row indices) at which our signals will be placed. Then, we extract the Pandas symbol wrapper from our data instance and use it to fill a new mask that has the same number of columns as we have symbols. Finally, set True at the generated positions of that mask:

>>> each_tuesday = vbt.date_range(index[0], index[-1], freq="tuesday")  # (1)!
>>> each_tuesday_1730 = each_tuesday + pd.Timedelta(hours=17, minutes=30)  # (2)!
>>> each_tuesday_1730
DatetimeIndex(['2021-01-05 17:30:00+00:00', '2021-01-12 17:30:00+00:00',
               '2021-01-19 17:30:00+00:00', '2021-01-26 17:30:00+00:00'],
              dtype='datetime64[ns, UTC]', freq=None)

>>> positions = index.get_indexer(each_tuesday_1730, method="bfill")

>>> min_symbol_wrapper = min_data.get_symbol_wrapper()  # (3)!
>>> mask = min_symbol_wrapper.fill(False)  # (4)!
>>> mask.iloc[positions] = True  # (5)!
>>> mask.sum()
symbol
BTCUSDT    4
ETHUSDT    4
dtype: int64

1. Timezone is embedded into both timestamps and will be used automatically
2. Adding a timedelta to a datetime will produce a new datetime
3. Use Data.get_symbol_wrapper to get a Pandas wrapper where columns are symbols
4. Use ArrayWrapper.fill to create an array with the same shape as the wrapper, and fill it with False values
5. Positions are integer indices of rows, thus use iloc to select the elements at those rows

Let's make sure that all signals match 18:00 on Tuesday, which is the first date after the requested 17:30 on Tuesday in an hourly index:

>>> mask[mask.any(axis=1)].index.strftime("%A %T")  # (1)!
Index(['Tuesday 18:00:00', 'Tuesday 18:00:00', 'Tuesday 18:00:00',
       'Tuesday 18:00:00'],
      dtype='object', name='Open time')

1. Details of the string format can be found here

The above solution is only required when only a single time boundary is known. For example, if we want 17:30 on Tuesday or later, we know only the left boundary while the right boundary is infinity (or we might get no data point after this datetime at all). When both time boundaries are known, we can easily use the first approach and combine it with the vectorbt's signal selection mechanism. For example, let's place a signal at 17:00 on Tuesday or later, but not later than 17:00 on Wednesday. This would require us placing signals from the left boundary all the way to the right boundary, and then selecting the first signal out of that partition:

>>> tuesday_after_1700 = (index.weekday == 1) & (index.hour >= 17)
>>> wednesday_before_1700 = (index.weekday == 2) & (index.hour < 17)
>>> main_cond = tuesday_after_1700 | wednesday_before_1700
>>> mask = min_symbol_wrapper.fill(False)
>>> mask[main_cond] = True
>>> mask = mask.vbt.signals.first()
>>> mask[mask.any(axis=1)].index.strftime("%A %T")
Index(['Tuesday 17:00:00', 'Tuesday 17:00:00', 'Tuesday 17:00:00',
       'Tuesday 17:00:00'],
      dtype='object', name='Open time')

The third and final approach is the vectorbt's one

It's relying on the two accessor methods BaseAccessor.set and BaseAccessor.set_between, which allow us to conditionally set elements of an array in a more intuitive manner.

Place a signal at 17:30 each Tuesday or later:

>>> mask = min_symbol_wrapper.fill(False)
>>> mask.vbt.set(
...     True, 
...     every="tuesday", 
...     at_time="17:30", 
...     inplace=True
... )
>>> mask[mask.any(axis=1)].index.strftime("%A %T")
Index(['Tuesday 18:00:00', 'Tuesday 18:00:00', 'Tuesday 18:00:00',
       'Tuesday 18:00:00'],
      dtype='object', name='Open time')

Place a signal after 18:00 each Tuesday (exclusive):

>>> mask = min_symbol_wrapper.fill(False)
>>> mask.vbt.set(
...     True, 
...     every="tuesday", 
...     at_time="18:00", 
...     add_delta=pd.Timedelta(nanoseconds=1),  # (1)!
...     inplace=True
... )
>>> mask[mask.any(axis=1)].index.strftime("%A %T")
Index(['Tuesday 19:00:00', 'Tuesday 19:00:00', 'Tuesday 19:00:00',
       'Tuesday 19:00:00'],
      dtype='object', name='Open time')

1. Add a nanosecond to exclude 18:00

Fill signals between 12:00 each Monday and 17:00 each Tuesday:

>>> mask = min_symbol_wrapper.fill(False)
>>> mask.vbt.set_between(
...     True, 
...     every="monday", 
...     start_time="12:00", 
...     end_time="17:00", 
...     add_end_delta=pd.Timedelta(days=1),  # (1)!
...     inplace=True
... )
>>> mask[mask.any(axis=1)].index.strftime("%A %T")
Index(['Monday 12:00:00', 'Monday 13:00:00', 'Monday 14:00:00',
       'Monday 15:00:00', 'Monday 16:00:00', 'Monday 17:00:00',
       'Monday 18:00:00', 'Monday 19:00:00', 'Monday 20:00:00',
       ...
       'Tuesday 10:00:00', 'Tuesday 11:00:00', 'Tuesday 12:00:00',
       'Tuesday 13:00:00', 'Tuesday 14:00:00', 'Tuesday 15:00:00',
       'Tuesday 16:00:00'],
      dtype='object', name='Open time', length=116)

1. Add a day to get Wednesday

Place a signal exactly at the midnight of January 7th, 2021:

>>> mask = min_symbol_wrapper.fill(False)
>>> mask.vbt.set(
...     True, 
...     on="January 7th 2021 UTC",  # (1)!
...     indexer_method=None,  # (2)!
...     inplace=True
... )
>>> mask[mask.any(axis=1)].index
DatetimeIndex(['2021-01-07 00:00:00+00:00'], ...)

1. Human-readable datetime strings are accepted
2. The default method is bfill, which takes the next timestamp if there was no exact match

Fill signals between 12:00 on January 1st/7th and 12:00 on January 2nd/8th, 2021:

>>> mask = min_symbol_wrapper.fill(False)
>>> mask.vbt.set_between(
...     True, 
...     start=["2021-01-01 12:00:00", "2021-01-07 12:00:00"],  # (1)!
...     end=["2021-01-02 12:00:00", "2021-01-08 12:00:00"],
...     inplace=True
... )
>>> mask[mask.any(axis=1)].index
DatetimeIndex(['2021-01-01 12:00:00+00:00', '2021-01-01 13:00:00+00:00',
               '2021-01-01 14:00:00+00:00', '2021-01-01 15:00:00+00:00',
               '2021-01-01 16:00:00+00:00', '2021-01-01 17:00:00+00:00',
               ...
               '2021-01-08 06:00:00+00:00', '2021-01-08 07:00:00+00:00',
               '2021-01-08 08:00:00+00:00', '2021-01-08 09:00:00+00:00',
               '2021-01-08 10:00:00+00:00', '2021-01-08 11:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)

1. The first range is built from the first element in start and end, and the second range is built from the second element. We could have also used human-readable datetime strings but those take time to map (to enable, see settings.datetime).

Fill signals in the first 2 hours of each week:

>>> mask = min_symbol_wrapper.fill(False)
>>> mask.vbt.set_between(
...     True, 
...     every="monday",
...     split_every=False,  # (1)!
...     add_end_delta="2h",
...     inplace=True
... )
>>> mask[mask.any(axis=1)].index
DatetimeIndex(['2021-01-04 00:00:00+00:00', '2021-01-04 01:00:00+00:00',
               '2021-01-11 00:00:00+00:00', '2021-01-11 01:00:00+00:00',
               '2021-01-18 00:00:00+00:00', '2021-01-18 01:00:00+00:00',
               '2021-01-25 00:00:00+00:00', '2021-01-25 01:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)

1. Otherwise, ranges will be built from each pair of week starts

See the API documentation for more examples.

Iteratively¶

With Numba, we can write an iterative logic that performs just as well as its vectorized counterparts. But which approach is better? There is no clear winner, although using vectors is an overall more effective and user-friendlier approach because it abstracts away looping over data and automates various mechanisms associated with index and columns. Just think about how powerful the concept of broadcasting is, and how many more lines of code it would require implementing something similar iteratively. Numba also doesn't allow us to work with labels and complex data types, only with numeric data, which requires skills and creativity in designing (efficient!) algorithms.

Moreover, most vectorized and also non-vectorized but compiled functions are specifically tailored at one specific task and perform it reliably, while writing an own loop makes you responsible to implement every bit of the logic correctly. Vectors are like Lego bricks that require almost zero effort to construct even the most breathtaking castles, while custom loops require learning how to design each Lego brick first

Nevertheless, the most functionality in vectorbt is powered by loops, not vectors - we should rename vectorbt to loopbt, really The main reason is plain and simple: most of the operations cannot be realized through vectors because they either introduce path dependencies, require complex data structures, use intermediate calculations or data buffers, periodically need to call a third-party function, or all of these together. Another reason is certainly efficiency: we can design algorithms that loop of the data only once, while performing the same logic using vectors would read the whole data sometimes a dozen of times. The same goes for memory consumption! Finally, defining and running a strategy at each time step is exactly how we would proceed in the real world (and in any other backtesting framework too), and we as traders should strive to mimic the real world as closely as possible.

Enough talking! Let's implement the first example from Logical operators using a custom loop. Unless our signals are based on multiple assets or some other column grouping, we should always start with one column only:

>>> @njit  # (1)!
... def generate_mask_1d_nb(  # (2)!
...     high, low,  # (3)!
...     uband, mband, lband,  # (4)!
...     cond2_th, cond4_th  # (5)!
... ):
...     out = np.full(high.shape, False)  # (6)!
...     
...     for i in range(high.shape[0]):  # (7)!
...         # (8)!
...         bandwidth = (uband[i] - lband[i]) / mband[i]
...         cond1 = low[i] < lband[i]
...         cond2 = bandwidth > cond2_th
...         cond3 = high[i] > uband[i]
...         cond4 = bandwidth < cond4_th
...         signal = (cond1 and cond2) or (cond3 and cond4)  # (9)!
...         out[i] = signal  # (10)!
...         
...     return out

>>> mask = generate_mask_1d_nb(
...     data.get("High")["BTCUSDT"].values,  # (11)!
...     data.get("Low")["BTCUSDT"].values,
...     bb.upperband["BTCUSDT"].values,
...     bb.middleband["BTCUSDT"].values,
...     bb.lowerband["BTCUSDT"].values,
...     0.30,
...     0.15
... )
>>> symbol_wrapper = data.get_symbol_wrapper()
>>> mask = symbol_wrapper["BTCUSDT"].wrap(mask)  # (12)!
>>> mask.sum()
25

1. Don't forget to decorate with @njit to make the function Numba-compiled
2. Good convention is to append a suffix nb to Numba-compiled functions and 1d_nb to those that work only on one column of data
3. Data arrays required by our logic
4. Bollinger Bands arrays required by our logic
5. Thresholds. Can be also defined as keyword arguments with default values.
6. Create a boolean array of the same shape as each of our arrays, and fill it with the default value False (no signal)
7. Iterate over the rows (= timestamps)
8. Here comes the main logic. We perform all operations on one element at a time instead of arrays, which is great for memory.
9. When working with single values, we need to replace & with and, | with or, and ~ with not
10. Write the current signal to the array
11. Since we're working with a one-dimensional Numba-compiled function, we must pass one-dimensional NumPy arrays
12. Since we generated the mask for the symbol BTCUSDT only, select the same column from the wrapper and wrap the array

We've got the same number of signals as previously - magic!

To make the function work on multiple columns, we can then write another Numba-compiled function that iterates over columns and calls generate_mask_1d_nb on each:

>>> @njit
... def generate_mask_nb(  # (1)!
...     high, low,
...     uband, mband, lband,
...     cond2_th, cond4_th
... ):
...     out = np.empty(high.shape, dtype=np.bool_)  # (2)!
...     
...     for col in range(high.shape[1]):  # (3)!
...         out[:, col] = generate_mask_1d_nb(  # (4)!
...             high[:, col], low[:, col],
...             uband[:, col], mband[:, col], lband[:, col],
...             cond2_th, cond4_th
...         )
...         
...     return out

>>> mask = generate_mask_nb(
...     vbt.to_2d_array(data.get("High")),  # (5)!
...     vbt.to_2d_array(data.get("Low")),
...     vbt.to_2d_array(bb.upperband),
...     vbt.to_2d_array(bb.middleband),
...     vbt.to_2d_array(bb.lowerband),
...     0.30,
...     0.15
... )
>>> mask = symbol_wrapper.wrap(mask)
>>> mask.sum()
symbol
BTCUSDT    25
ETHUSDT    13
dtype: int64

1. Remove 1d from the name if the function takes two-dimensional arrays
2. Create an empty boolean array that will be gradually filled with results from generate_mask_1d_nb. When using np.empty instead of np.full, make sure to override each single value, otherwise the elements that haven't been overridden will remain uninitialized and basically garbage.
3. Iterate over the columns (= assets)
4. Select the current column from each array and call our one-dimensional function
5. Use to_2d_array to cast any array to two dimensions and convert to NumPy

Probably a more "vectorbtonic" way is to create a stand-alone indicator where we can specify the function and what data it expects and returns, and the indicator factory will take care of everything else for us!

>>> MaskGenerator = vbt.IF(  # (1)!
...     input_names=["high", "low", "uband", "mband", "lband"],
...     param_names=["cond2_th", "cond4_th"],
...     output_names=["mask"]
... ).with_apply_func(generate_mask_1d_nb, takes_1d=True)  # (2)!
>>> mask_generator = MaskGenerator.run(  # (3)!
...     data.get("High"),
...     data.get("Low"),
...     bb.upperband,
...     bb.middleband,
...     bb.lowerband,
...     [0.3, 0.4],
...     [0.1, 0.2],
...     param_product=True  # (4)!
... )
>>> mask_generator.mask.sum()
custom_cond2_th  custom_cond4_th  symbol 
0.3              0.1              BTCUSDT    11
                                  ETHUSDT    10
                 0.2              BTCUSDT    28
                                  ETHUSDT    27
0.4              0.1              BTCUSDT     9
                                  ETHUSDT     5
                 0.2              BTCUSDT    26
                                  ETHUSDT    22
dtype: int64

1. Create the facade of the indicator and specify the apply function
2. We could have also used generate_mask_nb and takes_1d=False
3. Run the indicator. Notice that we don't have care about the proper type and shape of each array!
4. Test the Cartesian product of various threshold combinations

But what about shifting and truth value testing? Simple use cases such as fixed shifts and windows can be implemented quite easily. Below, we're comparing the current value to the value some number of ticks before:

>>> @njit
... def value_ago_1d_nb(arr, ago):
...     out = np.empty(arr.shape, dtype=np.float_)  # (1)!
...     for i in range(out.shape[0]):
...         if i - ago >= 0:  # (2)!
...             out[i] = arr[i - ago]
...         else:
...             out[i] = np.nan  # (3)!
...     return out

>>> arr = np.array([1, 2, 3])
>>> value_ago_1d_nb(arr, 1)
array([nan, 1., 2.])

1. Use np.empty if we can guarantee to override each element in the array. Also remember to set the data type of the array to floating as soon as NaN values are involved!
2. Before accessing the previous element, make sure that it's within the bounds of the array
3. If the previous element is outside the bounds, set it to NaN

And here's how to test if any condition was true inside a fixed window (= variable time interval):

>>> @njit
... def any_in_window_1d_nb(arr, window):
...     out = np.empty(arr.shape, dtype=np.bool_)  # (1)!
...     for i in range(out.shape[0]):
...         from_i = max(0, i + 1 - window)  # (2)!
...         to_i = i + 1  # (3)!
...         out[i] = np.any(arr[from_i:to_i])  # (4)!
...     return out

>>> arr = np.array([False, True, True, False, False])
>>> any_in_window_1d_nb(arr, 2)
array([False, True, True, True, False])

1. We can make the empty array boolean because there are no NaN values involved
2. Get the left bound of the window
3. Get the right bound of the window
4. Use the bounds to select all elements in the window and test whether there is a True value among them

As soon as dates and time are involved, such as to compare the current value to the value exactly 5 days ago, a better approach is to pre-calculate as many intermediate steps as possible. But there is also a possibility to work with a datetime-like index in Numba directly. Here's how to test if any condition was true inside a variable window (= fixed time interval):

>>> @njit
... def any_in_var_window_1d_nb(arr, index, freq):  # (1)!
...     out = np.empty(arr.shape, dtype=np.bool_)
...     from_i = 0
...     for i in range(out.shape[0]):
...         if index[from_i] <= index[i] - freq:  # (2)!
...             for j in range(from_i + 1, index.shape[0]):  # (3)!
...                 if index[j] > index[i] - freq:
...                     from_i = j
...                     break  # (4)!
...         to_i = i + 1
...         out[i] = np.any(arr[from_i:to_i])
...     return out

>>> arr = np.array([False, True, True, False, False])
>>> index = vbt.date_range("2020", freq="5min", periods=len(arr)).values  # (5)!
>>> freq = vbt.timedelta("10min").to_timedelta64()  # (6)!
>>> any_in_var_window_1d_nb(arr, index, freq)
array([False, True, True, True, False])

1. Take an array of the data type np.datetime64 as index and a constant of the type np.timedelta64 as frequency
2. Test whether the previous left bound is still valid, that is, inside the current time interval
3. If not, search for the current left bound using another loop that iterates over index
4. Once found, set the left bound to from_i and abort the second loop
5. Define an index with the 5-minute timeframe and convert it to a NumPy datetime array
6. Define a frequency of 10 minutes and convert it to a NumPy timedelta value

Remember that Numba (and thus vectorbt) has far more features for processing numeric data than datetime/timedelta data. But gladly, datetime/timedelta data can be safely converted into integer data outside Numba, and many functions will continue to work just as before:

>>> any_in_var_window_1d_nb(arr, vbt.dt.to_ns(index), vbt.dt.to_ns(freq))
array([False, True, True, True, False])

Why so? By converting a datetime/timedelta into an integer, we're extracting the total number of nanoseconds representing that object. For a datetime, the integer value becomes the number of nanoseconds after the Unix Epoch, which is 00:00:00 UTC on 1 January 1970:

>>> vbt.dt.to_ns(index)  # (1)!
array([1577836800000000000, 1577837100000000000, 1577837400000000000,
       1577837700000000000, 1577838000000000000])

>>> vbt.dt.to_ns(index - np.datetime64(0, "ns")) # (2)!
array([1577836800000000000, 1577837100000000000, 1577837400000000000,
       1577837700000000000, 1577838000000000000])

>>> vbt.dt.to_ns(freq)  # (3)!
600000000000

>>> vbt.dt.to_ns(freq) / 1000 / 1000 / 1000 / 60  # (4)!
10.0

1. Index as timedelta in nanoseconds after 1970
2. This is the same as converting datetime into timedelta by subtracting the Unix Epoch
3. Frequency as timedelta in nanoseconds
4. Convert nanoseconds into minutes

Generators¶

Writing own loops is powerful and makes fun, but even here vectorbt has functions that may make our life easier, especially for generating signals. The most flexible out of all helper functions is the Numba-compiled function generate_nb and its accessor class method SignalsAccessor.generate, which takes a target shape, initializes a boolean output array of that shape and fills it with Falsevalues, then iterates over the columns, and for each column, it calls a so-called "placement function" - a regular UDF that changes the mask in place. After the change, the placement function should return either the position of the last placed signal or -1 for no signal.

All the information about the current iteration is being passed via a context of the type GenEnContext, which contains the current segment of the output mask that can be modified in place, the range start (inclusive) that corresponds to that segment, the range end (exclusive), column, but also the full output mask for the user to be able to make patches wherever they want. This way, vectorbt abstracts away both preparing the array and looping over the columns, and assists the user in selecting the right subset of the output data to modify.

Let's place a signal at 17:00 (UTC) of each Tuesday:

>>> @njit
... def place_func_nb(c, index):  # (1)!
...     last_i = -1  # (2)!
...     for out_i in range(len(c.out)):  # (3)!
...         i = c.from_i + out_i  # (4)!
...         weekday = vbt.dt_nb.weekday_nb(index[i])  # (5)!
...         hour = vbt.dt_nb.hour_nb(index[i])
...         if weekday == 1 and hour == 17:  # (6)!
...             c.out[out_i] = True  # (7)!
...             last_i = out_i
...     return last_i  # (8)!

>>> mask = vbt.pd_acc.signals.generate(  # (9)!
...     symbol_wrapper.shape,  # (10)!
...     place_func_nb,
...     vbt.dt.to_ns(symbol_wrapper.index),  # (11)!
...     wrapper=symbol_wrapper  # (12)!
... )
>>> mask.sum()
symbol
BTCUSDT    0
ETHUSDT    0
dtype: int64

1. A placement function takes a context and optionally other user-defined arguments
2. Create a variable to track the position of the last placed signal (if any)
3. Iterate over the output array with a local index
4. Get the global index that we'll use to get the current timestamp. For example, if the array segment has 3 elements (len(out)) and the start index of the segment is 2 (from_i), then the first element corresponds to the position 2 (i) and the last element to the position 5.
5. dt_nb contains Numba-compiled functions for working datetimes and timedeltas, but mostly requiring them to have the integer representation!
6. Check whether the current timestamp is 17:00 on Tuesday
7. If yes, place a signal using the local index, otherwise continue with the next timestamp
8. Make sure that the returned index is local, not global!
9. Call SignalsAccessor.generate as a class method of the accessor SignalsAccessor
10. Provide the shape to iterate over
11. Provide the index in the integer representation (= the total number of nanoseconds)
12. Provide the wrapper to wrap the final NumPy array

But our index is a daily index, thus there can't be any signal. Instead, let's place a signal at the next possible timestamp:

>>> @njit
... def place_func_nb(c, index):
...     last_i = -1
...     for out_i in range(len(c.out)):
...         i = c.from_i + out_i
...         weekday = vbt.dt_nb.weekday_nb(index[i])
...         hour = vbt.dt_nb.hour_nb(index[i])
...         if weekday == 1 and hour == 17:
...             c.out[out_i] = True
...             last_i = out_i
...         else:
...             past_target_midnight = vbt.dt_nb.past_weekday_nb(index[i], 1)  # (1)!
...             past_target = past_target_midnight + 17 * vbt.dt_nb.h_ns  # (2)!
...             if (i > 0 and index[i - 1] < past_target) and \
...                 index[i] > past_target:  # (3)!
...                 c.out[out_i] = True
...                 last_i = out_i
...     return last_i

>>> mask = vbt.pd_acc.signals.generate(
...     symbol_wrapper.shape,
...     place_func_nb,
...     vbt.dt.to_ns(symbol_wrapper.index),
...     wrapper=symbol_wrapper
... )
>>> mask.sum()
symbol
BTCUSDT    52
ETHUSDT    52
dtype: int64

>>> mask.index[mask.any(axis=1)].strftime('%A %m/%d/%Y')  # (4)!
Index(['Wednesday 01/06/2021', ..., 'Wednesday 12/29/2021'],
      dtype='object', name='Open time')

1. Get the timestamp of the midnight on the previous Tuesday
2. Get the timestamp of 17:00 on the previous Tuesday
3. Check if the previous timestamp was before and the current timestamp is after the target time
4. All the generated signals must be on Wednesdays

The most fascinating part about the snippet above is that the entire datetime logic is being performed using just regular integers!

But what about multiple parameter combinations? We cannot pass the function above to the indicator factory because it doesn't look like an apply function. But vectorbt's got our back! There is an entire subclass of the indicator factory tailed at signal generation - SignalFactory. This class supports multiple generation modes that can be specified using the argument mode of the type FactoryMode. In our case, the mode is FactoryMode.Entries because our function generates signals based on the target shape only, and not based on other signal arrays. Furthermore, the signal factory accepts any additional inputs, parameters, and in-outputs to build the skeleton of our future indicator class.

The signal factory has the class method SignalFactory.with_place_func comparable to from_apply_func we've got used to. In fact, it takes a placement function and generates a custom function that does all the pre- and post-processing around generate_nb (note that other modes have other generation functions). This custom function, for example, prepares the arguments and assigns them to their correct positions in the placement function call. It's then forwarded down to IndicatorFactory.with_custom_func. As a result, we receive an indicator class with a run method that can be applied on any user-defined shape and any grid of parameter combinations. Sounds handy, right?

Let's parametrize our exact-match placement function with two parameters: weekday and hour.

>>> @njit
... def place_func_nb(c, weekday, hour, index):  # (1)!
...     last_i = -1
...     for out_i in range(len(c.out)):
...         i = c.from_i + out_i
...         weekday_now = vbt.dt_nb.weekday_nb(index[i])
...         hour_now = vbt.dt_nb.hour_nb(index[i])
...         if weekday_now == weekday and hour_now == hour:
...             c.out[out_i] = True
...             last_i = out_i
...     return last_i

>>> EntryGenerator = vbt.SignalFactory(
...     mode="entries",
...     param_names=["weekday", "hour"]
... ).with_place_func(
...     entry_place_func_nb=place_func_nb,  # (2)!
...     entry_settings=dict(  # (3)!
...         pass_params=["weekday", "hour"],
...     ),
...     var_args=True  # (4)!
... )
>>> entry_generator = EntryGenerator.run(
...     symbol_wrapper.shape,  # (5)!
...     1, 
...     [0, 17],  # (6)!
...     vbt.dt.to_ns(symbol_wrapper.index),  # (7)!
...     input_index=symbol_wrapper.index,  # (8)!
...     input_columns=symbol_wrapper.columns
... )
>>> entry_generator.entries.sum()
custom_weekday  custom_hour   
1               0            0    52
                             1    52
                17           0     0
                             1     0
dtype: int64

1. Each indicator function must first accept the input shape, then (optionally) any inputs, in-outputs, parameters, and only then additional arguments
2. When in the mode FactoryMode.Entries, we need to pass the placement function as entry_place_func_nb
3. Additionally, we need to provide settings to instruct the custom function to pass specific inputs, in-outputs, and parameters to the placement function. Without it, it will pass nothing!
4. Enable variable arguments to be able to pass our index as an additional positional argument
5. Input shape is required if there are no inputs or in-outputs to determine it from
6. Test two time combinations: 00:00 and 17:00
7. Can be passed thanks to var_args
8. Pass Pandas metadata for wrapping output arrays

The indicator function was able to match all midnight times but none afternoon times, which makes sense because our index is daily and thus contains midnight times only. We can easily plot the indicator using the attached plot method, which knows how to visualize each mode:

>>> entry_generator.plot(column=(2, 0, "BTCUSDT")).show()

Exits¶

After populating the position entry mask, we should decide on the position exit mask. When exits do not rely on entries, we can use the generator introduced above. In other cases though, we might have a logic that makes an exit signal fully depend on the entry signal. For example, an exit signal representing a stop loss exists solely because of the entry signal that defined that stop loss condition. There is also no guarantee that an exit can be found for an entry at all. Thus, this mode should only be used for cases where entries do not depend on exits, but exits depend on entries. The generation is then done using the Numba-compiled function generate_ex_nb and its accessor instance method SignalsAccessor.generate_exits. The passed context is now of the type GenExContext and also includes the input mask and various generator-related arguments.

The generator takes an entry mask array, and in each column, it visits each entry signal and calls a UDF to place one or more exit signals succeeding it. Do you recall how we had to accept from_i and to_i in the placement functions above? The previous mode always passed 0 as from_i and len(index) as to_i because we had all the freedom to define our signals across the entire column. Here, the passed from_i will usually be the next index after the previous entry, while the passed to_i will usually be the index of the next entry, thus effectively limiting our decision field to the space between each pair of entries.

Let's generate an entry each quarter and an exit at the next date:

>>> @njit
... def exit_place_func_nb(c):
...     c.out[0] = True  # (1)!
...     return 0

>>> entries = symbol_wrapper.fill(False)
>>> entries.vbt.set(True, every="Q", inplace=True)
>>> entries.index[entries.any(axis=1)]
DatetimeIndex(['2021-03-31 00:00:00+00:00', '2021-06-30 00:00:00+00:00',
               '2021-09-30 00:00:00+00:00', '2021-12-31 00:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)

>>> exits = entries.vbt.signals.generate_exits(exit_place_func_nb)  # (2)!
>>> exits.index[exits.any(axis=1)]
DatetimeIndex(['2021-04-01 00:00:00+00:00', '2021-07-01 00:00:00+00:00',
               '2021-10-01 00:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)

1. Place an exit at the first position in the segment
2. In contrast to SignalsAccessor.generate, this method is bound to a Pandas object with entries and is using its metadata

We can control the distance to the entry signal using wait, which defaults to 1. Let's instruct vectorbt to start each segment at the same timestamp as the entry:

>>> exits = entries.vbt.signals.generate_exits(
...     exit_place_func_nb,
...     wait=0
... )
>>> exits.index[exits.any(axis=1)]
DatetimeIndex(['2021-03-31 00:00:00+00:00', '2021-06-30 00:00:00+00:00',
               '2021-09-30 00:00:00+00:00', '2021-12-31 00:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)

And below is how to implement a variable waiting time based on a frequency. Let's wait exactly 7 days before placing an exit:

>>> @njit
... def exit_place_func_nb(c, index, wait_td):
...     for out_i in range(len(c.out)):
...         i = c.from_i + out_i
...         if index[i] >= index[c.from_i] + wait_td:  # (1)!
...             return out_i  # (2)!
...     return -1

>>> exits = entries.vbt.signals.generate_exits(
...     exit_place_func_nb,
...     vbt.dt.to_ns(entries.index),  # (3)!
...     vbt.dt.to_ns(vbt.timedelta("7d")),
...     wait=0
... )
>>> exits.index[exits.any(axis=1)]
DatetimeIndex(['2021-04-07 00:00:00+00:00', '2021-07-07 00:00:00+00:00',
               '2021-10-07 00:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)

1. Check whether a sufficient amount of time has passed since the entry. For c.from_i to be the index of the entry, we must set wait to zero.
2. When returning the index, the generator will set the respective element to True automatically
3. Our additional arguments come here

But what happens with the exit condition for the previous entry if the next entry is less than 7 days away? Will the exit still be placed? No!

>>> entries = symbol_wrapper.fill(False)
>>> entries.vbt.set(True, every="5d", inplace=True)
>>> exits = entries.vbt.signals.generate_exits(
...     exit_place_func_nb,
...     vbt.dt.to_ns(entries.index),
...     vbt.dt.to_ns(vbt.timedelta("7d")),
...     wait=0
... )
>>> exits.index[exits.any(axis=1)]
DatetimeIndex([], dtype='datetime64[ns, UTC]', name='Open time', freq='D')

By default, each segment is limited by the two entries surrounding it. To make it infinite, we can disable until_next:

>>> exits = entries.vbt.signals.generate_exits(
...     exit_place_func_nb,
...     vbt.dt.to_ns(entries.index),
...     vbt.dt.to_ns(vbt.timedelta("7d")),
...     wait=0,
...     until_next=False
... )
>>> exits.index[exits.any(axis=1)]
DatetimeIndex(['2021-01-08 00:00:00+00:00', '2021-01-13 00:00:00+00:00',
               '2021-01-18 00:00:00+00:00', '2021-01-23 00:00:00+00:00',
               '2021-01-28 00:00:00+00:00', '2021-02-02 00:00:00+00:00',
               ...
               '2021-12-04 00:00:00+00:00', '2021-12-09 00:00:00+00:00',
               '2021-12-14 00:00:00+00:00', '2021-12-19 00:00:00+00:00',
               '2021-12-24 00:00:00+00:00', '2021-12-29 00:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)

In the case above, the generated signals follow the following schema: entry1, entry2, exit1, entry3, exit2, and so on. Whenever those signals are passed to the simulator, it will execute entry1 and ignore entry2 because there was no exit prior to it - we're still in the market. It will then rightfully execute exit1. But then, it will open a new position with entry3 and close it with exit2 right after, which was originally designed for entry2 (that has been ignored). To avoid this mistake, we should enable skip_until_exit to avoid processing any future entry signal that comes before an exit for any past entry signal. This would match the simulation order.

>>> exits = entries.vbt.signals.generate_exits(
...     exit_place_func_nb,
...     vbt.dt.to_ns(entries.index),
...     vbt.dt.to_ns(vbt.timedelta("7d")),
...     wait=0,
...     until_next=False,
...     skip_until_exit=True
... )
>>> exits.index[exits.any(axis=1)]
DatetimeIndex(['2021-01-08 00:00:00+00:00', '2021-01-18 00:00:00+00:00',
               '2021-01-28 00:00:00+00:00', '2021-02-07 00:00:00+00:00',
               '2021-02-17 00:00:00+00:00', '2021-02-27 00:00:00+00:00',
               ...
               '2021-11-04 00:00:00+00:00', '2021-11-14 00:00:00+00:00',
               '2021-11-24 00:00:00+00:00', '2021-12-04 00:00:00+00:00',
               '2021-12-14 00:00:00+00:00', '2021-12-24 00:00:00+00:00'],
              dtype='datetime64[ns, UTC]', name='Open time', freq=None)