Indicators

The IndicatorFactory class is one of the most powerful components in the VBT ecosystem. It can wrap any indicator function, making it both parameterizable and analyzable.

An indicator is a pipeline that performs the following steps:

• Accepts input arrays (for example, opening and closing prices).
• Accepts parameters either as scalars or arrays (for example, window size).
• Accepts other relevant arguments and keyword arguments.
• Broadcasts input arrays against each other or to a specific shape.
• Broadcasts parameters against each other to create a fixed set of parameter combinations.
• Calculates output arrays for each parameter combination using the input arrays, producing the same shape (for example, a rolling average).
• Concatenates the output arrays of all parameter combinations along the columns.
• Converts the results back to the Pandas format.

Let's manually create an indicator that takes two time series, computes their normalized moving averages, and returns the difference between the two. We will test different shapes as well as parameter combinations to see how broadcasting can be leveraged:

from vectorbtpro import *

def mov_avg_crossover(ts1, ts2, w1, w2):
    ts1, ts2 = vbt.broadcast(ts1, ts2)  

    w1, w2 = vbt.broadcast(  
        vbt.to_1d_array(w1),
        vbt.to_1d_array(w2))

    ts1_mas = []
    for w in w1:
        ts1_mas.append(ts1.vbt.rolling_mean(w) / ts1)  
    ts2_mas = []
    for w in w2:
        ts2_mas.append(ts2.vbt.rolling_mean(w) / ts2)

    ts1_ma = pd.concat(ts1_mas, axis=1)  
    ts2_ma = pd.concat(ts2_mas, axis=1)

    ts1_ma.columns = vbt.combine_indexes((  
        pd.Index(w1, name="ts1_window"),
        ts1.columns))
    ts2_ma.columns = vbt.combine_indexes((
        pd.Index(w2, name="ts2_window"),
        ts2.columns))

    return ts1_ma.vbt - ts2_ma  

def generate_index(n):  
    return vbt.date_range("2020-01-01", periods=n)

ts1 = pd.Series([1, 2, 3, 4, 5, 6, 7], index=generate_index(7))
ts2 = pd.DataFrame({
    'a': [5, 4, 3, 2, 3, 4, 5],
    'b': [2, 3, 4, 5, 4, 3, 2]
}, index=generate_index(7))
w1 = 2
w2 = [3, 4]

mov_avg_crossover(ts1, ts2, w1, w2)

ts1_window                                       2
ts2_window                   3                   4
                   a         b         a         b
2020-01-01       NaN       NaN       NaN       NaN
2020-01-02       NaN       NaN       NaN       NaN
2020-01-03 -0.500000  0.083333       NaN       NaN
2020-01-04 -0.625000  0.075000 -0.875000  0.175000
2020-01-05  0.011111 -0.183333 -0.100000 -0.100000
2020-01-06  0.166667 -0.416667  0.166667 -0.416667
2020-01-07  0.128571 -0.571429  0.228571 -0.821429

Pretty neat! We just built a flexible pipeline that can handle arbitrary input and parameter combinations. The resulting DataFrame shows each column as a specific window combination applied to each column in both ts1 and ts2. But is this pipeline user-friendly? 🤔 Dealing with broadcasting, output concatenation, and column hierarchies makes this process very similar to working with regular Pandas code.

The pipeline above can be easily standardized using IndicatorBase.run_pipeline. This method conveniently prepares inputs, parameters, and columns. However, you still need to perform the calculation and output concatenation yourself by providing a custom_func. Let's update the example:

def custom_func(ts1, ts2, w1, w2):
    ts1_mas = []
    for w in w1:
        ts1_mas.append(vbt.nb.rolling_mean_nb(ts1, w) / ts1)  
    ts2_mas = []
    for w in w2:
        ts2_mas.append(vbt.nb.rolling_mean_nb(ts2, w) / ts2)

    ts1_ma = np.column_stack(ts1_mas)  
    ts2_ma = np.column_stack(ts2_mas)

    return ts1_ma - ts2_ma  

outputs = vbt.IndicatorBase.run_pipeline(
    num_ret_outputs=1,
    custom_func=custom_func,
    inputs=dict(ts1=ts1, ts2=ts2),
    params=dict(w1=w1, w2=w2)
)
(type(outputs[0]), *outputs[1:])

(vectorbtpro.base.wrapping.ArrayWrapper,
 [array([[1, 1],
         [2, 2],
         [3, 3],
         [4, 4],
         [5, 5],
         [6, 6],
         [7, 7]]),
  array([[5, 2],
         [4, 3],
         [3, 4],
         [2, 5],
         [3, 4],
         [4, 3],
         [5, 2]])],
 array([0, 1, 0, 1]),
 [],
 [array([[        nan,         nan,         nan,         nan],
         [        nan,         nan,         nan,         nan],
         [-0.5       ,  0.08333333,         nan,         nan],
         [-0.625     ,  0.075     , -0.875     ,  0.175     ],
         [ 0.01111111, -0.18333333, -0.1       , -0.1       ],
         [ 0.16666667, -0.41666667,  0.16666667, -0.41666667],
         [ 0.12857143, -0.57142857,  0.22857143, -0.82142857]])],
 [[2, 2], [3, 4]],
 [Index([2, 2, 2, 2], dtype='int64'), Index([3, 3, 4, 4], dtype='int64')],
 [])