Combining/weighing indicator values

[andreas-vester]

What would be the way to go in order to combine multiple indicator values of the same indicator type into one new indicator? Sounds weird? Here's an example ;-)

Let's say I want to compute a regular "rate-of-change" indicator. I am going to compute it twice. First with a lockback window size of 50, and second with a window size of 100. Now, I want to be able to weigh both indicator results by, let's say, 25% and 75%. I am finally interested in the weighted indicator values. That is sum(0.25 * win(50)-indicator-values, 0.75 * win(100)-indicator-values).

Now, for me there's the question of how to compute it? Should I make the calculation function of the actual indicator capable of weighing directly? In that case and with respect to params, I'm probably looking for window and weights to be of type Union[int, List[int]] and Optional[Union[int, List[int]]]=1, respectively. If simply providing window=100, this would compute the regular 100 period rate-of-change. Providing window=[50, 100], weights=[0.25, 0.75] should then weigh the individual indicator values.

The alternative would be to keep the actual calculation function as simple as possible and let it solely compute the regular rate-of-change indicator. The weighting could then occur after computing multiple instances of the roc-indicator. However, when following this route, I am wondering how to preserve all the class methods of the actual indicator?

[polakowo]

@andreas-vester the simplest way (let's say with moving averages): weight1 * vbt.MA.run(price, window1).ma + weight2 * vbt.MA.run(price, window2).ma. If you need to check multiple hyperparameters, create an indicator with four parameters, pass keep_pd=True, and repeat the above in your apply function (see https://vectorbt.dev/api/indicators/factory/#without-numba).

[andreas-vester]

the simplest way (let's say with moving averages): weight1 * vbt.MA.run(price, window1).ma + weight2 * vbt.MA.run(price, window2).ma

I fear that it is not as easy as that. Two remarks:

1. sma_weighted = weight1 * vbt.MA.run(price, window1).ma + weight2 * vbt.MA.run(price, window2).ma
sma_weighted is of type pd.DataFrame. With that, we're going to loose all class methods. I think it would be better if type(sma_weighted) equals vectorbt.indicators.basic.MA, wouldn't it?
2. If window1 is not equal to window2, I do get a pd.DataFrame full of np.nan. Can you replicate this on your end?

If you need to check multiple hyperparameters, create an indicator with four parameters, pass keep_pd=True, and repeat the above in your apply function (see https://vectorbt.dev/api/indicators/factory/#without-numba).

Not sure, if I fully get it. What do you actually mean with "hyperparameters"? Do you have a small example?

[andreas-vester]

2. If window1 is not equal to window2, I do get a pd.DataFrame full of np.nan. Can you replicate this on your end?

Forget about that. I solved the issue on my side. It had to do with np.nan within different places in the time series.

[andreas-vester]

All right, I am coming up with the following solution:

from typing import List, Optional, Union

import numpy as np
import pandas as pd
import vectorbt as vbt


def roc(
    timeseries: np.ndarray,
    windows: Union[int, List[int]],
    weights: List[int],
) -> np.ndarray:
    if isinstance(windows, int):
        windows = [windows]

    return sum(
        [
            vbt.nb.pct_change_nb(a=timeseries, n=window) * weight
            for (window, weight) in zip(windows, weights)
        ]
    )


rate_of_change = vbt.IndicatorFactory(
    short_name="roc",
    input_names=["price"],
    param_names=["window", "weights"],
    output_names=["roc"],
).from_apply_func(roc, weights=[[1]])


df = pd.DataFrame(data=np.random.randint(10, 100, [15, 2]))

indicator1 = rate_of_change.run(
    price=df, window=[[2, 5], [3, 7, 10]], weights=[[0.25, 0.75], [0.5, 0.3, 0.2]]
)

I was fighting quite a bit with getting the actual lists into the roc function. According to the docs https://vectorbt.dev/api/indicators/factory/#parameters, I was of the opinion that I need to pass

rate_of_change = vbt.IndicatorFactory(
    short_name="roc",
    input_names=["price"],
    param_names=["window", "weights"],
    output_names=["roc"],
).from_apply_func(
    roc,
    weights=[[1]],
    param_settings=dict(
        window=dict(is_array_like=False), weights=dict(is_array_like=False)
    ),
)

and then call the function like this:

indicator = rate_of_change.run(price=df, window=[3, 7, 10], weights=[0.5, 0.3, 0.2])

However, I learned that it doesn't make any difference if I set the values to True or False. The roc function always receives scalar values. What actually is this setting for? Could you provide a small example?
Passing the parameters as a list of lists did the trick.

Anyway, that's basically the result I was looking for in the first place. Performing the weightings in the roc function has the advantage that the final indicator has all the class method of a regular indicator.

Your thoughts?

[polakowo]

@andreas-vester is_array_like works only with NumPy arrays. Try setting it to True and passing window=np.array([3, 7, 10]), this window will be considered as a single value. With is_array_like=False, this window will be considered as three different parameter values. So basically is_array_like=True means whenever I pass a single NumPy array, consider it as one parameter value, not multiple. Your solution is fine. You could go further and make it Numba-compiled.

[andreas-vester]

is_array_like works only with NumPy arrays.

Aaah, got it :-)

You could go further and make it Numba-compiled.

Would that mean to simply put the @njit decorator above the roc function? And is it fair to say that I can make it numba-compiled because I am utilizing a loop? I am not really familiar with numba.

If you need to check multiple hyperparameters, create an indicator with four parameters, pass keep_pd=True

Would you mind to elaborate on this? Just for my understanding... - Thanks :-)

[andreas-vester]

Here's a more robust solution that takes into account, that the user should be able to provide a range of different formats with respect to lookback periods and weightings.

The following cases will be covered:

• single computation with single lookback period, no weightings:
  periods = 5
• multiple computations with single lookback periods, no weightings:
  periods = [3, 7]
• single computation with multiple lookback periods and weightings:
  periods = [4, 8]
weights = [0.4, 0.6]
• multiple computatios with multiple lookback periods and weightings:
  periods = [[4, 8], [3, 5, 7]]
weights = [[0.4, 0.6], [0.5, 0.3, 0.2]]

from typing import List, Optional, Union

import numpy as np
import pandas as pd
import pytest
import vectorbt as vbt


def roc(
    data: Union[pd.Series, pd.DataFrame],
    periods: Union[int, List[int], List[List[int]]] = 1,
    weights: Union[List[float], List[List[float]]] = None,
) -> np.ndarray:
    def _roc(
        timeseries: np.ndarray,
        windows: Union[int, List[int]],
        wgts: Optional[List[int]] = None,
    ) -> np.ndarray:
        if wgts is None:
            return vbt.nb.pct_change_nb(a=timeseries, n=windows)
        else:
            return sum(
                [
                    vbt.nb.pct_change_nb(a=timeseries, n=window) * weight
                    for (window, weight) in zip(windows, wgts)
                ]
            )

    rate_of_change = vbt.IndicatorFactory(
        short_name="roc",
        input_names=["price"],
        param_names=["window", "weights"],
        output_names=["roc"],
    ).from_apply_func(
        _roc,
        weights=None,
        param_settings=dict(
            window=dict(is_array_like=True), weights=dict(is_array_like=True)
        ),
    )

    # Transform input arguments

    # Case 1: regular roc (single or multiple), no weightings
    if (
        isinstance(periods, int)
        or (not any(isinstance(period, list) for period in periods))
        and weights is None
    ):
        return rate_of_change.run(price=data, window=periods)

    # Case 2: weighted roc
    else:
        # single roc, eg. periods=[3, 6] and weights=[0.3, 0.7]
        if not any(isinstance(period, list) for period in periods):
            periods = np.array(periods)
            weights = np.array(weights)

        # multiple rocs, e.g. periods = [[4, 8], [3, 5, 7]] and
        # weights = [[0.4, 0.6], [0.5, 0.3, 0.2]]
        else:
            periods = [np.array(period) for period in periods]
            weights = [np.array(weight) for weight in weights]
        return rate_of_change.run(price=data, window=periods, weights=weights)

I also performed some unit tests:

@pytest.fixture(scope="module")
def df():
    return pd.DataFrame(
        data=np.random.randint(10, 100, [15, 3]), columns=["A", "B", "C"]
    )


def test_with_default_params(df):
    rate_of_change = roc(data=df).roc

    assert len(rate_of_change.columns) == 3
    assert rate_of_change.columns.names == ["roc_window", None]
    assert rate_of_change.columns.get_level_values(0).to_list() == [1, 1, 1]
    assert rate_of_change.columns.get_level_values(1).to_list() == ["A", "B", "C"]
    assert pd.isna(rate_of_change.iloc[:1, :]).all().all()


def test_single_period(df):
    periods = 5
    rate_of_change = roc(data=df, periods=periods).roc

    assert len(rate_of_change.columns) == 3
    assert rate_of_change.columns.names == ["roc_window", None]
    assert rate_of_change.columns.get_level_values(0).to_list() == [5, 5, 5]
    assert rate_of_change.columns.get_level_values(1).to_list() == ["A", "B", "C"]
    assert pd.isna(rate_of_change.iloc[:5, :]).all().all()


def test_multiple_periods(df):
    periods = [3, 7]
    rate_of_change = roc(data=df, periods=periods).roc

    assert len(rate_of_change.columns) == 6
    assert rate_of_change.columns.names == ["roc_window", None]
    assert rate_of_change.columns.get_level_values(0).to_list() == [3, 3, 3, 7, 7, 7]
    assert rate_of_change.columns.get_level_values(1).to_list() == [
        "A",
        "B",
        "C",
        "A",
        "B",
        "C",
    ]
    assert pd.isna(rate_of_change.iloc[:3, :]).all().all()
    assert pd.isna(rate_of_change.iloc[:7, 3::]).all().all()


def test_single_weighted_roc(df):
    periods = [4, 8]
    weights = [0.4, 0.6]
    rate_of_change = roc(data=df, periods=periods, weights=weights).roc

    assert len(rate_of_change.columns) == 3
    assert rate_of_change.columns.names == ["roc_window", "roc_weights", None]
    assert rate_of_change.columns.get_level_values(0).to_list() == [
        "array_0",
        "array_0",
        "array_0",
    ]
    assert rate_of_change.columns.get_level_values(1).to_list() == [
        "array_0",
        "array_0",
        "array_0",
    ]
    assert rate_of_change.columns.get_level_values(2).to_list() == ["A", "B", "C"]
    assert pd.isna(rate_of_change.iloc[:4, :]).all().all()


def test_multiple_weighted_roc(df):
    periods = [[4, 8], [3, 5, 7]]
    weights = [[0.4, 0.6], [0.5, 0.3, 0.2]]
    rate_of_change = roc(data=df, periods=periods, weights=weights).roc

    assert len(rate_of_change.columns) == 6
    assert rate_of_change.columns.names == ["roc_window", "roc_weights", None]
    assert rate_of_change.columns.get_level_values(0).to_list() == [
        "array_0",
        "array_0",
        "array_0",
        "array_1",
        "array_1",
        "array_1",
    ]
    assert rate_of_change.columns.get_level_values(0).to_list() == [
        "array_0",
        "array_0",
        "array_0",
        "array_1",
        "array_1",
        "array_1",
    ]
    assert rate_of_change.columns.get_level_values(2).to_list() == [
        "A",
        "B",
        "C",
        "A",
        "B",
        "C",
    ]
    assert pd.isna(rate_of_change.iloc[:8, :3]).all().all()
    assert pd.isna(rate_of_change.iloc[:7, 3:]).all().all()

I still need some help in order to make this code numba-compiled. Would be highly appreciated.