from datetime import datetime
import numpy as np
import pandas as pd
import pytest
from numba import njit
import vectorbt as vbt
from tests.utils import record_arrays_close
from vectorbt.generic import nb as generic_nb
from vectorbt.generic.enums import range_dt
seed = 42
day_dt = np.timedelta64(86400000000000)
mask = pd.DataFrame([
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False],
[False, True, False]
], index=pd.Index([
datetime(2020, 1, 1),
datetime(2020, 1, 2),
datetime(2020, 1, 3),
datetime(2020, 1, 4),
datetime(2020, 1, 5)
]), columns=['a', 'b', 'c'])
ts = pd.Series([1., 2., 3., 2., 1.], index=mask.index)
price = pd.DataFrame({
'open': [10, 11, 12, 11, 10],
'high': [11, 12, 13, 12, 11],
'low': [9, 10, 11, 10, 9],
'close': [11, 12, 11, 10, 9]
})
group_by = pd.Index(['g1', 'g1', 'g2'])
def setup_module():
vbt.settings.numba['check_func_suffix'] = True
vbt.settings.caching.enabled = False
vbt.settings.caching.whitelist = []
vbt.settings.caching.blacklist = []
def teardown_module():
vbt.settings.reset()
class TestAccessors:
def test_indexing(self):
assert mask.vbt.signals['a'].total() == mask['a'].vbt.signals.total()
def test_freq(self):
assert mask.vbt.signals.wrapper.freq == day_dt
assert mask['a'].vbt.signals.wrapper.freq == day_dt
assert mask.vbt.signals(freq='2D').wrapper.freq == day_dt * 2
assert mask['a'].vbt.signals(freq='2D').wrapper.freq == day_dt * 2
assert pd.Series([False, True]).vbt.signals.wrapper.freq is None
assert pd.Series([False, True]).vbt.signals(freq='3D').wrapper.freq == day_dt * 3
assert pd.Series([False, True]).vbt.signals(freq=np.timedelta64(4, 'D')).wrapper.freq == day_dt * 4
@pytest.mark.parametrize(
"test_n",
[1, 2, 3, 4, 5],
)
def test_fshift(self, test_n):
pd.testing.assert_series_equal(mask['a'].vbt.signals.fshift(test_n), mask['a'].shift(test_n, fill_value=False))
np.testing.assert_array_equal(
mask['a'].vbt.signals.fshift(test_n).values,
generic_nb.fshift_1d_nb(mask['a'].values, test_n, fill_value=False)
)
pd.testing.assert_frame_equal(mask.vbt.signals.fshift(test_n), mask.shift(test_n, fill_value=False))
@pytest.mark.parametrize(
"test_n",
[1, 2, 3, 4, 5],
)
def test_bshift(self, test_n):
pd.testing.assert_series_equal(
mask['a'].vbt.signals.bshift(test_n),
mask['a'].shift(-test_n, fill_value=False))
np.testing.assert_array_equal(
mask['a'].vbt.signals.bshift(test_n).values,
generic_nb.bshift_1d_nb(mask['a'].values, test_n, fill_value=False)
)
pd.testing.assert_frame_equal(mask.vbt.signals.bshift(test_n), mask.shift(-test_n, fill_value=False))
def test_empty(self):
pd.testing.assert_series_equal(
pd.Series.vbt.signals.empty(5, index=np.arange(10, 15), name='a'),
pd.Series(np.full(5, False), index=np.arange(10, 15), name='a')
)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.empty((5, 3), index=np.arange(10, 15), columns=['a', 'b', 'c']),
pd.DataFrame(np.full((5, 3), False), index=np.arange(10, 15), columns=['a', 'b', 'c'])
)
pd.testing.assert_series_equal(
pd.Series.vbt.signals.empty_like(mask['a']),
pd.Series(np.full(mask['a'].shape, False), index=mask['a'].index, name=mask['a'].name)
)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.empty_like(mask),
pd.DataFrame(np.full(mask.shape, False), index=mask.index, columns=mask.columns)
)
def test_generate(self):
@njit
def choice_func_nb(from_i, to_i, col, n):
if col == 0:
return np.arange(from_i, to_i)
elif col == 1:
return np.full(1, from_i)
else:
return np.full(1, to_i - n)
pd.testing.assert_series_equal(
pd.Series.vbt.signals.generate(5, choice_func_nb, 1, index=mask['a'].index, name=mask['a'].name),
pd.Series(
np.array([True, True, True, True, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
with pytest.raises(Exception):
_ = pd.Series.vbt.signals.generate((5, 2), choice_func_nb, 1)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.generate(
(5, 3), choice_func_nb, 1, index=mask.index, columns=mask.columns),
pd.DataFrame(
np.array([
[True, True, False],
[True, False, False],
[True, False, False],
[True, False, False],
[True, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.generate(
(5, 3), choice_func_nb, 1, pick_first=True, index=mask.index, columns=mask.columns),
pd.DataFrame(
np.array([
[True, True, False],
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
def test_generate_both(self):
@njit
def entry_func_nb(from_i, to_i, col, temp_int):
temp_int[0] = from_i
return temp_int[:1]
@njit
def exit_func_nb(from_i, to_i, col, temp_int):
temp_int[0] = from_i
return temp_int[:1]
temp_int = np.empty((mask.shape[0],), dtype=np.int64)
en, ex = pd.Series.vbt.signals.generate_both(
5, entry_func_nb, (temp_int,), exit_func_nb, (temp_int,),
index=mask['a'].index, name=mask['a'].name)
pd.testing.assert_series_equal(
en,
pd.Series(
np.array([True, False, True, False, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
pd.testing.assert_series_equal(
ex,
pd.Series(
np.array([False, True, False, True, False]),
index=mask['a'].index,
name=mask['a'].name
)
)
en, ex = pd.DataFrame.vbt.signals.generate_both(
(5, 3), entry_func_nb, (temp_int,), exit_func_nb, (temp_int,),
index=mask.index, columns=mask.columns)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[False, False, False],
[True, True, True],
[False, False, False],
[True, True, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[True, True, True],
[False, False, False],
[True, True, True],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
en, ex = pd.Series.vbt.signals.generate_both(
(5,), entry_func_nb, (temp_int,), exit_func_nb, (temp_int,),
index=mask['a'].index, name=mask['a'].name, entry_wait=1, exit_wait=0)
pd.testing.assert_series_equal(
en,
pd.Series(
np.array([True, True, True, True, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
pd.testing.assert_series_equal(
ex,
pd.Series(
np.array([True, True, True, True, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
en, ex = pd.Series.vbt.signals.generate_both(
(5,), entry_func_nb, (temp_int,), exit_func_nb, (temp_int,),
index=mask['a'].index, name=mask['a'].name, entry_wait=0, exit_wait=1)
pd.testing.assert_series_equal(
en,
pd.Series(
np.array([True, True, True, True, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
pd.testing.assert_series_equal(
ex,
pd.Series(
np.array([False, True, True, True, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
@njit
def entry_func2_nb(from_i, to_i, col, temp_int):
temp_int[0] = from_i
if from_i + 1 < to_i:
temp_int[1] = from_i + 1
return temp_int[:2]
return temp_int[:1]
@njit
def exit_func2_nb(from_i, to_i, col, temp_int):
temp_int[0] = from_i
if from_i + 1 < to_i:
temp_int[1] = from_i + 1
return temp_int[:2]
return temp_int[:1]
en, ex = pd.DataFrame.vbt.signals.generate_both(
(5, 3), entry_func2_nb, (temp_int,), exit_func2_nb, (temp_int,),
entry_pick_first=False, exit_pick_first=False,
index=mask.index, columns=mask.columns)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[True, True, True],
[False, False, False],
[False, False, False],
[True, True, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[False, False, False],
[True, True, True],
[True, True, True],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
def test_generate_exits(self):
@njit
def choice_func_nb(from_i, to_i, col, temp_int):
temp_int[0] = from_i
return temp_int[:1]
temp_int = np.empty((mask.shape[0],), dtype=np.int64)
pd.testing.assert_series_equal(
mask['a'].vbt.signals.generate_exits(choice_func_nb, temp_int),
pd.Series(
np.array([False, True, False, False, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_exits(choice_func_nb, temp_int),
pd.DataFrame(
np.array([
[False, False, False],
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_exits(choice_func_nb, temp_int, wait=0),
pd.DataFrame(
np.array([
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False],
[False, True, False]
]),
index=mask.index,
columns=mask.columns
)
)
@njit
def choice_func2_nb(from_i, to_i, col, temp_int):
for i in range(from_i, to_i):
temp_int[i - from_i] = i
return temp_int[:to_i - from_i]
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_exits(choice_func2_nb, temp_int, until_next=False, pick_first=False),
pd.DataFrame(
np.array([
[False, False, False],
[True, False, False],
[True, True, False],
[True, True, True],
[True, True, True]
]),
index=mask.index,
columns=mask.columns
)
)
mask2 = pd.Series([True, True, True, True, True], index=mask.index)
pd.testing.assert_series_equal(
mask2.vbt.signals.generate_exits(choice_func_nb, temp_int, until_next=False, skip_until_exit=True),
pd.Series(
np.array([False, True, False, True, False]),
index=mask.index
)
)
def test_clean(self):
entries = pd.DataFrame([
[True, False, True],
[True, False, False],
[True, True, True],
[False, True, False],
[False, True, True]
], index=mask.index, columns=mask.columns)
exits = pd.Series([True, False, True, False, True], index=mask.index)
pd.testing.assert_frame_equal(
entries.vbt.signals.clean(),
pd.DataFrame(
np.array([
[True, False, True],
[False, False, False],
[False, True, True],
[False, False, False],
[False, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.clean(entries),
pd.DataFrame(
np.array([
[True, False, True],
[False, False, False],
[False, True, True],
[False, False, False],
[False, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
entries.vbt.signals.clean(exits)[0],
pd.DataFrame(
np.array([
[False, False, False],
[True, False, False],
[False, False, False],
[False, True, False],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
entries.vbt.signals.clean(exits)[1],
pd.DataFrame(
np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False],
[True, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
entries.vbt.signals.clean(exits, entry_first=False)[0],
pd.DataFrame(
np.array([
[False, False, False],
[True, False, False],
[False, False, False],
[False, True, False],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
entries.vbt.signals.clean(exits, entry_first=False)[1],
pd.DataFrame(
np.array([
[False, True, False],
[False, False, False],
[False, False, False],
[False, False, False],
[True, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.clean(entries, exits)[0],
pd.DataFrame(
np.array([
[False, False, False],
[True, False, False],
[False, False, False],
[False, True, False],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.clean(entries, exits)[1],
pd.DataFrame(
np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False],
[True, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
with pytest.raises(Exception):
_ = pd.Series.vbt.signals.clean(entries, entries, entries)
def test_generate_random(self):
pd.testing.assert_series_equal(
pd.Series.vbt.signals.generate_random(
5, n=3, seed=seed, index=mask['a'].index, name=mask['a'].name),
pd.Series(
np.array([False, True, True, False, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
with pytest.raises(Exception):
_ = pd.Series.vbt.signals.generate_random((5, 2), n=3)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.generate_random(
(5, 3), n=3, seed=seed, index=mask.index, columns=mask.columns),
pd.DataFrame(
np.array([
[False, False, True],
[True, True, True],
[True, True, False],
[False, True, True],
[True, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.generate_random(
(5, 3), n=[0, 1, 2], seed=seed, index=mask.index, columns=mask.columns),
pd.DataFrame(
np.array([
[False, False, True],
[False, False, True],
[False, False, False],
[False, True, False],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_series_equal(
pd.Series.vbt.signals.generate_random(
5, prob=0.5, seed=seed, index=mask['a'].index, name=mask['a'].name),
pd.Series(
np.array([True, False, False, False, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
with pytest.raises(Exception):
_ = pd.Series.vbt.signals.generate_random((5, 2), prob=3)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.generate_random(
(5, 3), prob=0.5, seed=seed, index=mask.index, columns=mask.columns),
pd.DataFrame(
np.array([
[True, True, True],
[False, True, False],
[False, False, False],
[False, False, True],
[True, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.generate_random(
(5, 3), prob=[0., 0.5, 1.], seed=seed, index=mask.index, columns=mask.columns),
pd.DataFrame(
np.array([
[False, True, True],
[False, True, True],
[False, False, True],
[False, False, True],
[False, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
with pytest.raises(Exception):
pd.DataFrame.vbt.signals.generate_random((5, 3))
pd.testing.assert_frame_equal(
pd.DataFrame.vbt.signals.generate_random(
(5, 3), prob=[0., 0.5, 1.], pick_first=True, seed=seed, index=mask.index, columns=mask.columns),
pd.DataFrame(
np.array([
[False, True, True],
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
def test_generate_random_both(self):
en, ex = pd.Series.vbt.signals.generate_random_both(
5, n=2, seed=seed, index=mask['a'].index, name=mask['a'].name)
pd.testing.assert_series_equal(
en,
pd.Series(
np.array([True, False, True, False, False]),
index=mask['a'].index,
name=mask['a'].name
)
)
pd.testing.assert_series_equal(
ex,
pd.Series(
np.array([False, True, False, False, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both(
(5, 3), n=2, seed=seed, index=mask.index, columns=mask.columns)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[False, False, False],
[True, True, False],
[False, False, True],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[True, True, True],
[False, False, False],
[False, True, False],
[True, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both(
(5, 3), n=[0, 1, 2], seed=seed, index=mask.index, columns=mask.columns)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[False, False, True],
[False, True, False],
[False, False, False],
[False, False, True],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[False, False, True],
[False, False, False],
[False, True, False],
[False, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both((2, 3), n=2, seed=seed, entry_wait=1, exit_wait=0)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[True, True, True],
])
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[True, True, True],
[True, True, True]
])
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both((3, 3), n=2, seed=seed, entry_wait=0, exit_wait=1)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[True, True, True],
[False, False, False]
])
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[True, True, True],
[True, True, True],
])
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both((7, 3), n=2, seed=seed, entry_wait=2, exit_wait=2)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[False, False, False],
[False, False, False],
[False, False, False],
[True, True, True],
[False, False, False],
[False, False, False]
])
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[False, False, False],
[True, True, True],
[False, False, False],
[False, False, False],
[False, False, False],
[True, True, True]
])
)
)
n = 10
a = np.full(n * 2, 0.)
for i in range(10000):
en, ex = pd.Series.vbt.signals.generate_random_both(1000, n, entry_wait=2, exit_wait=2)
_a = np.empty((n * 2,), dtype=np.int64)
_a[0::2] = np.flatnonzero(en)
_a[1::2] = np.flatnonzero(ex)
a += _a
greater = a > 10000000 / (2 * n + 1) * np.arange(0, 2 * n)
less = a < 10000000 / (2 * n + 1) * np.arange(2, 2 * n + 2)
assert np.all(greater & less)
en, ex = pd.Series.vbt.signals.generate_random_both(
5, entry_prob=0.5, exit_prob=1., seed=seed, index=mask['a'].index, name=mask['a'].name)
pd.testing.assert_series_equal(
en,
pd.Series(
np.array([True, False, False, False, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
pd.testing.assert_series_equal(
ex,
pd.Series(
np.array([False, True, False, False, False]),
index=mask['a'].index,
name=mask['a'].name
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both(
(5, 3), entry_prob=0.5, exit_prob=1., seed=seed, index=mask.index, columns=mask.columns)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[False, False, False],
[False, False, False],
[False, False, True],
[True, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[True, True, True],
[False, False, False],
[False, False, False],
[False, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both(
(5, 3), entry_prob=[0., 0.5, 1.], exit_prob=[0., 0.5, 1.],
seed=seed, index=mask.index, columns=mask.columns)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[False, True, True],
[False, False, False],
[False, False, True],
[False, False, False],
[False, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[False, True, True],
[False, False, False],
[False, False, True],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both(
(5, 3), entry_prob=1., exit_prob=1., exit_wait=0,
seed=seed, index=mask.index, columns=mask.columns)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[True, True, True],
[True, True, True],
[True, True, True],
[True, True, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[True, True, True],
[True, True, True],
[True, True, True],
[True, True, True],
[True, True, True]
]),
index=mask.index,
columns=mask.columns
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both(
(5, 3), entry_prob=1., exit_prob=1., entry_pick_first=False, exit_pick_first=True,
seed=seed, index=mask.index, columns=mask.columns)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[True, True, True],
[True, True, True],
[True, True, True],
[True, True, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
en, ex = pd.DataFrame.vbt.signals.generate_random_both(
(5, 3), entry_prob=1., exit_prob=1., entry_pick_first=True, exit_pick_first=False,
seed=seed, index=mask.index, columns=mask.columns)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(
np.array([
[True, True, True],
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(
np.array([
[False, False, False],
[True, True, True],
[True, True, True],
[True, True, True],
[True, True, True]
]),
index=mask.index,
columns=mask.columns
)
)
with pytest.raises(Exception):
pd.DataFrame.vbt.signals.generate_random((5, 3))
def test_generate_random_exits(self):
pd.testing.assert_series_equal(
mask['a'].vbt.signals.generate_random_exits(seed=seed),
pd.Series(
np.array([False, False, True, False, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_random_exits(seed=seed),
pd.DataFrame(
np.array([
[False, False, False],
[False, False, False],
[True, True, False],
[False, False, False],
[True, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_random_exits(seed=seed, wait=0),
pd.DataFrame(
np.array([
[True, False, False],
[False, False, False],
[False, True, False],
[False, False, True],
[True, True, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_series_equal(
mask['a'].vbt.signals.generate_random_exits(prob=1., seed=seed),
pd.Series(
np.array([False, True, False, False, True]),
index=mask['a'].index,
name=mask['a'].name
)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_random_exits(prob=1., seed=seed),
pd.DataFrame(
np.array([
[False, False, False],
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_random_exits(prob=[0., 0.5, 1.], seed=seed),
pd.DataFrame(
np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, True, True],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_random_exits(prob=1., wait=0, seed=seed),
pd.DataFrame(
np.array([
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False],
[False, True, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_random_exits(prob=1., until_next=False, seed=seed),
pd.DataFrame(
np.array([
[False, False, False],
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
def test_generate_stop_exits(self):
e = pd.Series([True, False, False, False, False, False])
t = pd.Series([2, 3, 4, 3, 2, 1]).astype(np.float64)
pd.testing.assert_series_equal(
e.vbt.signals.generate_stop_exits(t, -0.1),
pd.Series(np.array([False, False, False, False, False, True]))
)
pd.testing.assert_series_equal(
e.vbt.signals.generate_stop_exits(t, -0.1, trailing=True),
pd.Series(np.array([False, False, False, True, False, False]))
)
pd.testing.assert_series_equal(
e.vbt.signals.generate_stop_exits(t, -0.1, trailing=True, pick_first=False),
pd.Series(np.array([False, False, False, True, True, True]))
)
pd.testing.assert_frame_equal(
e.vbt.signals.generate_stop_exits(t.vbt.tile(3), [np.nan, -0.5, -1.], trailing=True, pick_first=False),
pd.DataFrame(np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False],
[False, True, False],
[False, True, False]
]))
)
pd.testing.assert_series_equal(
e.vbt.signals.generate_stop_exits(t, -0.1, trailing=True, exit_wait=3),
pd.Series(np.array([False, False, False, False, True, False]))
)
pd.testing.assert_series_equal(
e.vbt.signals.generate_stop_exits(4 - t, 0.1),
pd.Series(np.array([False, False, False, False, False, True]))
)
pd.testing.assert_series_equal(
e.vbt.signals.generate_stop_exits(4 - t, 0.1, trailing=True),
pd.Series(np.array([False, False, False, True, False, False]))
)
pd.testing.assert_series_equal(
e.vbt.signals.generate_stop_exits(4 - t, 0.1, trailing=True, pick_first=False),
pd.Series(np.array([False, False, False, True, True, True]))
)
pd.testing.assert_frame_equal(
e.vbt.signals.generate_stop_exits((4 - t).vbt.tile(3), [np.nan, 0.5, 1.], trailing=True, pick_first=False),
pd.DataFrame(np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, True, True],
[False, True, True],
[False, True, True]
]))
)
pd.testing.assert_series_equal(
e.vbt.signals.generate_stop_exits(4 - t, 0.1, trailing=True, exit_wait=3),
pd.Series(np.array([False, False, False, False, True, False]))
)
e = pd.Series([True, True, True, True, True, True])
en, ex = e.vbt.signals.generate_stop_exits(t, -0.1, trailing=True, chain=True)
pd.testing.assert_series_equal(
en,
pd.Series(np.array([True, False, False, False, True, False]))
)
pd.testing.assert_series_equal(
ex,
pd.Series(np.array([False, False, False, True, False, True]))
)
en, ex = e.vbt.signals.generate_stop_exits(t, -0.1, trailing=True, entry_wait=2, chain=True)
pd.testing.assert_series_equal(
en,
pd.Series(np.array([True, False, False, False, False, True]))
)
pd.testing.assert_series_equal(
ex,
pd.Series(np.array([False, False, False, True, False, False]))
)
en, ex = e.vbt.signals.generate_stop_exits(t, -0.1, trailing=True, exit_wait=2, chain=True)
pd.testing.assert_series_equal(
en,
pd.Series(np.array([True, False, False, False, True, False]))
)
pd.testing.assert_series_equal(
ex,
pd.Series(np.array([False, False, False, True, False, False]))
)
e2 = pd.Series([True, True, True, True, True, True])
t2 = pd.Series([6, 5, 4, 3, 2, 1]).astype(np.float64)
ex = e2.vbt.signals.generate_stop_exits(t2, -0.1, until_next=False, pick_first=False)
pd.testing.assert_series_equal(
ex,
pd.Series(np.array([False, True, True, True, True, True]))
)
def test_generate_ohlc_stop_exits(self):
with pytest.raises(Exception):
_ = mask.vbt.signals.generate_ohlc_stop_exits(ts, sl_stop=-0.1)
with pytest.raises(Exception):
_ = mask.vbt.signals.generate_ohlc_stop_exits(ts, tp_stop=-0.1)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_stop_exits(ts, -0.1),
mask.vbt.signals.generate_ohlc_stop_exits(ts, sl_stop=0.1)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_stop_exits(ts, -0.1, trailing=True),
mask.vbt.signals.generate_ohlc_stop_exits(ts, sl_stop=0.1, sl_trail=True)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_stop_exits(ts, 0.1),
mask.vbt.signals.generate_ohlc_stop_exits(ts, tp_stop=0.1)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_stop_exits(ts, 0.1),
mask.vbt.signals.generate_ohlc_stop_exits(ts, sl_stop=0.1, reverse=True)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_stop_exits(ts, 0.1, trailing=True),
mask.vbt.signals.generate_ohlc_stop_exits(ts, sl_stop=0.1, sl_trail=True, reverse=True)
)
pd.testing.assert_frame_equal(
mask.vbt.signals.generate_stop_exits(ts, -0.1),
mask.vbt.signals.generate_ohlc_stop_exits(ts, tp_stop=0.1, reverse=True)
)
def _test_ohlc_stop_exits(**kwargs):
out_dict = {'stop_price': np.nan, 'stop_type': -1}
result = mask.vbt.signals.generate_ohlc_stop_exits(
price['open'], price['high'], price['low'], price['close'],
out_dict=out_dict, **kwargs
)
if isinstance(result, tuple):
_, ex = result
else:
ex = result
return result, out_dict['stop_price'], out_dict['stop_type']
ex, stop_price, stop_type = _test_ohlc_stop_exits()
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_type,
pd.DataFrame(np.array([
[-1, -1, -1],
[-1, -1, -1],
[-1, -1, -1],
[-1, -1, -1],
[-1, -1, -1]
]), index=mask.index, columns=mask.columns)
)
ex, stop_price, stop_type = _test_ohlc_stop_exits(sl_stop=0.1)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, True],
[True, False, False]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, 10.8],
[9.9, np.nan, np.nan]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_type,
pd.DataFrame(np.array([
[-1, -1, -1],
[-1, -1, -1],
[-1, -1, -1],
[-1, -1, 0],
[0, -1, -1]
]), index=mask.index, columns=mask.columns)
)
ex, stop_price, stop_type = _test_ohlc_stop_exits(sl_stop=0.1, sl_trail=True)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, True, True],
[True, False, False]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, 11.7, 10.8],
[9.9, np.nan, np.nan]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_type,
pd.DataFrame(np.array([
[-1, -1, -1],
[-1, -1, -1],
[-1, -1, -1],
[-1, 1, 1],
[1, -1, -1]
]), index=mask.index, columns=mask.columns)
)
ex, stop_price, stop_type = _test_ohlc_stop_exits(tp_stop=0.1)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(np.array([
[False, False, False],
[True, False, False],
[False, True, False],
[False, False, False],
[False, False, False]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan],
[11.0, np.nan, np.nan],
[np.nan, 12.1, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_type,
pd.DataFrame(np.array([
[-1, -1, -1],
[2, -1, -1],
[-1, 2, -1],
[-1, -1, -1],
[-1, -1, -1]
]), index=mask.index, columns=mask.columns)
)
ex, stop_price, stop_type = _test_ohlc_stop_exits(sl_stop=0.1, sl_trail=True, tp_stop=0.1)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(np.array([
[False, False, False],
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan],
[11.0, np.nan, np.nan],
[np.nan, 12.1, np.nan],
[np.nan, np.nan, 10.8],
[9.9, np.nan, np.nan]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_type,
pd.DataFrame(np.array([
[-1, -1, -1],
[2, -1, -1],
[-1, 2, -1],
[-1, -1, 1],
[1, -1, -1]
]), index=mask.index, columns=mask.columns)
)
ex, stop_price, stop_type = _test_ohlc_stop_exits(
sl_stop=[np.nan, 0.1, 0.2], sl_trail=True, tp_stop=[np.nan, 0.1, 0.2])
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(np.array([
[False, False, False],
[False, False, False],
[False, True, False],
[False, False, False],
[False, False, True]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, 12.1, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, 9.6]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_type,
pd.DataFrame(np.array([
[-1, -1, -1],
[-1, -1, -1],
[-1, 2, -1],
[-1, -1, -1],
[-1, -1, 1]
]), index=mask.index, columns=mask.columns)
)
ex, stop_price, stop_type = _test_ohlc_stop_exits(sl_stop=0.1, sl_trail=True, tp_stop=0.1, exit_wait=0)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(np.array([
[True, False, False],
[False, False, False],
[False, True, False],
[False, False, True],
[True, True, False]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_price,
pd.DataFrame(np.array([
[9.0, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, 12.1, np.nan],
[np.nan, np.nan, 11.7],
[10.8, 9.0, np.nan]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_type,
pd.DataFrame(np.array([
[1, -1, -1],
[-1, -1, -1],
[-1, 2, -1],
[-1, -1, 1],
[1, 1, -1]
]), index=mask.index, columns=mask.columns)
)
(en, ex), stop_price, stop_type = _test_ohlc_stop_exits(
sl_stop=0.1, sl_trail=True, tp_stop=0.1, chain=True)
pd.testing.assert_frame_equal(
en,
pd.DataFrame(np.array([
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False],
[False, True, False]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
ex,
pd.DataFrame(np.array([
[False, False, False],
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan],
[11.0, np.nan, np.nan],
[np.nan, 12.1, np.nan],
[np.nan, np.nan, 10.8],
[9.9, np.nan, np.nan]
]), index=mask.index, columns=mask.columns)
)
pd.testing.assert_frame_equal(
stop_type,
pd.DataFrame(np.array([
[-1, -1, -1],
[2, -1, -1],
[-1, 2, -1],
[-1, -1, 1],
[1, -1, -1]
]), index=mask.index, columns=mask.columns)
)
def test_between_ranges(self):
ranges = mask.vbt.signals.between_ranges()
record_arrays_close(
ranges.values,
np.array([
(0, 0, 0, 3, 1), (1, 1, 1, 4, 1)
], dtype=range_dt)
)
assert ranges.wrapper == mask.vbt.wrapper
mask2 = pd.DataFrame([
[True, True, True],
[True, True, True],
[False, False, False],
[False, False, False],
[False, False, False]
], index=mask.index, columns=mask.columns)
other_mask = pd.DataFrame([
[False, False, False],
[True, False, False],
[True, True, False],
[False, True, True],
[False, False, True]
], index=mask.index, columns=mask.columns)
ranges = mask2.vbt.signals.between_ranges(other=other_mask)
record_arrays_close(
ranges.values,
np.array([
(0, 0, 0, 1, 1), (1, 0, 1, 1, 1), (2, 1, 0, 2, 1),
(3, 1, 1, 2, 1), (4, 2, 0, 3, 1), (5, 2, 1, 3, 1)
], dtype=range_dt)
)
assert ranges.wrapper == mask2.vbt.wrapper
ranges = mask2.vbt.signals.between_ranges(other=other_mask, from_other=True)
record_arrays_close(
ranges.values,
np.array([
(0, 0, 1, 1, 1), (1, 0, 1, 2, 1), (2, 1, 1, 2, 1),
(3, 1, 1, 3, 1), (4, 2, 1, 3, 1), (5, 2, 1, 4, 1)
], dtype=range_dt)
)
assert ranges.wrapper == mask2.vbt.wrapper
def test_partition_ranges(self):
mask2 = pd.DataFrame([
[False, False, False],
[True, False, False],
[True, True, False],
[False, True, True],
[True, False, True]
], index=mask.index, columns=mask.columns)
ranges = mask2.vbt.signals.partition_ranges()
record_arrays_close(
ranges.values,
np.array([
(0, 0, 1, 3, 1), (1, 0, 4, 4, 0), (2, 1, 2, 4, 1), (3, 2, 3, 4, 0)
], dtype=range_dt)
)
assert ranges.wrapper == mask2.vbt.wrapper
def test_between_partition_ranges(self):
mask2 = pd.DataFrame([
[True, False, False],
[True, True, False],
[False, True, True],
[True, False, True],
[False, True, False]
], index=mask.index, columns=mask.columns)
ranges = mask2.vbt.signals.between_partition_ranges()
record_arrays_close(
ranges.values,
np.array([
(0, 0, 1, 3, 1), (1, 1, 2, 4, 1)
], dtype=range_dt)
)
assert ranges.wrapper == mask2.vbt.wrapper
def test_pos_rank(self):
pd.testing.assert_series_equal(
(~mask['a']).vbt.signals.pos_rank(),
pd.Series([-1, 0, 1, -1, 0], index=mask['a'].index, name=mask['a'].name)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.pos_rank(),
pd.DataFrame(
np.array([
[-1, 0, 0],
[0, -1, 1],
[1, 0, -1],
[-1, 1, 0],
[0, -1, 1]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.pos_rank(after_false=True),
pd.DataFrame(
np.array([
[-1, -1, -1],
[0, -1, -1],
[1, 0, -1],
[-1, 1, 0],
[0, -1, 1]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.pos_rank(allow_gaps=True),
pd.DataFrame(
np.array([
[-1, 0, 0],
[0, -1, 1],
[1, 1, -1],
[-1, 2, 2],
[2, -1, 3]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.pos_rank(reset_by=mask['a'], allow_gaps=True),
pd.DataFrame(
np.array([
[-1, 0, 0],
[0, -1, 1],
[1, 1, -1],
[-1, 0, 0],
[0, -1, 1]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.pos_rank(reset_by=mask, allow_gaps=True),
pd.DataFrame(
np.array([
[-1, 0, 0],
[0, -1, 1],
[1, 0, -1],
[-1, 1, 0],
[0, -1, 1]
]),
index=mask.index,
columns=mask.columns
)
)
def test_partition_pos_rank(self):
pd.testing.assert_series_equal(
(~mask['a']).vbt.signals.partition_pos_rank(),
pd.Series([-1, 0, 0, -1, 1], index=mask['a'].index, name=mask['a'].name)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.partition_pos_rank(),
pd.DataFrame(
np.array([
[-1, 0, 0],
[0, -1, 0],
[0, 1, -1],
[-1, 1, 1],
[1, -1, 1]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.partition_pos_rank(after_false=True),
pd.DataFrame(
np.array([
[-1, -1, -1],
[0, -1, -1],
[0, 0, -1],
[-1, 0, 0],
[1, -1, 0]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.partition_pos_rank(reset_by=mask['a']),
pd.DataFrame(
np.array([
[-1, 0, 0],
[0, -1, 0],
[0, 1, -1],
[-1, 0, 0],
[0, -1, 0]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.partition_pos_rank(reset_by=mask),
pd.DataFrame(
np.array([
[-1, 0, 0],
[0, -1, 0],
[0, 0, -1],
[-1, 0, 0],
[0, -1, 0]
]),
index=mask.index,
columns=mask.columns
)
)
def test_pos_rank_fns(self):
pd.testing.assert_frame_equal(
(~mask).vbt.signals.first(),
pd.DataFrame(
np.array([
[False, True, True],
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.nth(1),
pd.DataFrame(
np.array([
[False, False, False],
[False, False, True],
[True, False, False],
[False, True, False],
[False, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.nth(2),
pd.DataFrame(
np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, False]
]),
index=mask.index,
columns=mask.columns
)
)
pd.testing.assert_frame_equal(
(~mask).vbt.signals.from_nth(0),
pd.DataFrame(
np.array([
[False, True, True],
[True, False, True],
[True, True, False],
[False, True, True],
[True, False, True]
]),
index=mask.index,
columns=mask.columns
)
)
def test_pos_rank_mapped(self):
mask2 = pd.DataFrame([
[True, False, False],
[True, True, False],
[False, True, True],
[True, False, True],
[False, True, False]
], index=mask.index, columns=mask.columns)
mapped = mask2.vbt.signals.pos_rank_mapped()
np.testing.assert_array_equal(
mapped.values,
np.array([0, 1, 0, 0, 1, 0, 0, 1])
)
np.testing.assert_array_equal(
mapped.col_arr,
np.array([0, 0, 0, 1, 1, 1, 2, 2])
)
np.testing.assert_array_equal(
mapped.idx_arr,
np.array([0, 1, 3, 1, 2, 4, 2, 3])
)
assert mapped.wrapper == mask2.vbt.wrapper
def test_partition_pos_rank_mapped(self):
mask2 = pd.DataFrame([
[True, False, False],
[True, True, False],
[False, True, True],
[True, False, True],
[False, True, False]
], index=mask.index, columns=mask.columns)
mapped = mask2.vbt.signals.partition_pos_rank_mapped()
np.testing.assert_array_equal(
mapped.values,
np.array([0, 0, 1, 0, 0, 1, 0, 0])
)
np.testing.assert_array_equal(
mapped.col_arr,
np.array([0, 0, 0, 1, 1, 1, 2, 2])
)
np.testing.assert_array_equal(
mapped.idx_arr,
np.array([0, 1, 3, 1, 2, 4, 2, 3])
)
assert mapped.wrapper == mask2.vbt.wrapper
def test_nth_index(self):
assert mask['a'].vbt.signals.nth_index(0) == pd.Timestamp('2020-01-01 00:00:00')
pd.testing.assert_series_equal(
mask.vbt.signals.nth_index(0),
pd.Series([
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-02 00:00:00'),
pd.Timestamp('2020-01-03 00:00:00')
], index=mask.columns, name='nth_index', dtype='datetime64[ns]')
)
pd.testing.assert_series_equal(
mask.vbt.signals.nth_index(-1),
pd.Series([
pd.Timestamp('2020-01-04 00:00:00'),
pd.Timestamp('2020-01-05 00:00:00'),
pd.Timestamp('2020-01-03 00:00:00')
], index=mask.columns, name='nth_index', dtype='datetime64[ns]')
)
pd.testing.assert_series_equal(
mask.vbt.signals.nth_index(-2),
pd.Series([
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-02 00:00:00'),
np.nan
], index=mask.columns, name='nth_index', dtype='datetime64[ns]')
)
pd.testing.assert_series_equal(
mask.vbt.signals.nth_index(0, group_by=group_by),
pd.Series([
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-03 00:00:00')
], index=['g1', 'g2'], name='nth_index', dtype='datetime64[ns]')
)
pd.testing.assert_series_equal(
mask.vbt.signals.nth_index(-1, group_by=group_by),
pd.Series([
pd.Timestamp('2020-01-05 00:00:00'),
pd.Timestamp('2020-01-03 00:00:00')
], index=['g1', 'g2'], name='nth_index', dtype='datetime64[ns]')
)
def test_norm_avg_index(self):
assert mask['a'].vbt.signals.norm_avg_index() == -0.25
pd.testing.assert_series_equal(
mask.vbt.signals.norm_avg_index(),
pd.Series([-0.25, 0.25, 0.0], index=mask.columns, name='norm_avg_index')
)
pd.testing.assert_series_equal(
mask.vbt.signals.norm_avg_index(group_by=group_by),
pd.Series([0.0, 0.0], index=['g1', 'g2'], name='norm_avg_index')
)
def test_index_mapped(self):
mapped = mask.vbt.signals.index_mapped()
np.testing.assert_array_equal(
mapped.values,
np.array([0, 3, 1, 4, 2])
)
np.testing.assert_array_equal(
mapped.col_arr,
np.array([0, 0, 1, 1, 2])
)
np.testing.assert_array_equal(
mapped.idx_arr,
np.array([0, 3, 1, 4, 2])
)
assert mapped.wrapper == mask.vbt.wrapper
def test_total(self):
assert mask['a'].vbt.signals.total() == 2
pd.testing.assert_series_equal(
mask.vbt.signals.total(),
pd.Series([2, 2, 1], index=mask.columns, name='total')
)
pd.testing.assert_series_equal(
mask.vbt.signals.total(group_by=group_by),
pd.Series([4, 1], index=['g1', 'g2'], name='total')
)
def test_rate(self):
assert mask['a'].vbt.signals.rate() == 0.4
pd.testing.assert_series_equal(
mask.vbt.signals.rate(),
pd.Series([0.4, 0.4, 0.2], index=mask.columns, name='rate')
)
pd.testing.assert_series_equal(
mask.vbt.signals.rate(group_by=group_by),
pd.Series([0.4, 0.2], index=['g1', 'g2'], name='rate')
)
def test_total_partitions(self):
assert mask['a'].vbt.signals.total_partitions() == 2
pd.testing.assert_series_equal(
mask.vbt.signals.total_partitions(),
pd.Series([2, 2, 1], index=mask.columns, name='total_partitions')
)
pd.testing.assert_series_equal(
mask.vbt.signals.total_partitions(group_by=group_by),
pd.Series([4, 1], index=['g1', 'g2'], name='total_partitions')
)
def test_partition_rate(self):
assert mask['a'].vbt.signals.partition_rate() == 1.0
pd.testing.assert_series_equal(
mask.vbt.signals.partition_rate(),
pd.Series([1.0, 1.0, 1.0], index=mask.columns, name='partition_rate')
)
pd.testing.assert_series_equal(
mask.vbt.signals.partition_rate(group_by=group_by),
pd.Series([1.0, 1.0], index=['g1', 'g2'], name='partition_rate')
)
def test_stats(self):
stats_index = pd.Index([
'Start', 'End', 'Period', 'Total', 'Rate [%]', 'First Index',
'Last Index', 'Norm Avg Index [-1, 1]', 'Distance: Min',
'Distance: Max', 'Distance: Mean', 'Distance: Std', 'Total Partitions',
'Partition Rate [%]', 'Partition Length: Min', 'Partition Length: Max',
'Partition Length: Mean', 'Partition Length: Std',
'Partition Distance: Min', 'Partition Distance: Max',
'Partition Distance: Mean', 'Partition Distance: Std'
], dtype='object')
pd.testing.assert_series_equal(
mask.vbt.signals.stats(),
pd.Series([
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-05 00:00:00'),
pd.Timedelta('5 days 00:00:00'),
1.6666666666666667,
33.333333333333336,
pd.Timestamp('2020-01-02 00:00:00'),
pd.Timestamp('2020-01-04 00:00:00'),
0.0,
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
np.nan,
1.6666666666666667,
100.0,
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('0 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
np.nan
],
index=stats_index,
name='agg_func_mean'
)
)
pd.testing.assert_series_equal(
mask.vbt.signals.stats(column='a'),
pd.Series([
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-05 00:00:00'),
pd.Timedelta('5 days 00:00:00'),
2,
40.0,
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-04 00:00:00'),
-0.25,
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
np.nan,
2,
100.0,
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('0 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
np.nan
],
index=stats_index,
name='a'
)
)
pd.testing.assert_series_equal(
mask.vbt.signals.stats(column='a', settings=dict(to_timedelta=False)),
pd.Series([
pd.Timestamp('2020-01-01 00:00:00'), pd.Timestamp('2020-01-05 00:00:00'), 5, 2, 40.0,
pd.Timestamp('2020-01-01 00:00:00'), pd.Timestamp('2020-01-04 00:00:00'), -0.25, 3.0,
3.0, 3.0, np.nan, 2, 100.0, 1.0, 1.0, 1.0, 0.0, 3.0, 3.0, 3.0, np.nan
],
index=stats_index,
name='a'
)
)
pd.testing.assert_series_equal(
mask.vbt.signals.stats(column='a', settings=dict(other=mask['b'], from_other=True)),
pd.Series([
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-05 00:00:00'),
pd.Timedelta('5 days 00:00:00'),
2,
40.0,
0,
0.0,
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-04 00:00:00'),
-0.25,
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('0 days 00:00:00'),
2,
100.0,
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('0 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
np.nan
],
index=pd.Index([
'Start', 'End', 'Period', 'Total', 'Rate [%]', 'Total Overlapping',
'Overlapping Rate [%]', 'First Index', 'Last Index',
'Norm Avg Index [-1, 1]', 'Distance <- Other: Min',
'Distance <- Other: Max', 'Distance <- Other: Mean',
'Distance <- Other: Std', 'Total Partitions', 'Partition Rate [%]',
'Partition Length: Min', 'Partition Length: Max',
'Partition Length: Mean', 'Partition Length: Std',
'Partition Distance: Min', 'Partition Distance: Max',
'Partition Distance: Mean', 'Partition Distance: Std'
], dtype='object'),
name='a'
)
)
pd.testing.assert_series_equal(
mask.vbt.signals.stats(column='g1', group_by=group_by),
pd.Series([
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-05 00:00:00'),
pd.Timedelta('5 days 00:00:00'),
4,
40.0,
pd.Timestamp('2020-01-01 00:00:00'),
pd.Timestamp('2020-01-05 00:00:00'),
0.0,
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('0 days 00:00:00'),
4,
100.0,
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('1 days 00:00:00'),
pd.Timedelta('0 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('3 days 00:00:00'),
pd.Timedelta('0 days 00:00:00')
],
index=stats_index,
name='g1'
)
)
pd.testing.assert_series_equal(
mask['c'].vbt.signals.stats(),
mask.vbt.signals.stats(column='c')
)
pd.testing.assert_series_equal(
mask['c'].vbt.signals.stats(),
mask.vbt.signals.stats(column='c', group_by=False)
)
pd.testing.assert_series_equal(
mask.vbt.signals(group_by=group_by)['g2'].stats(),
mask.vbt.signals(group_by=group_by).stats(column='g2')
)
pd.testing.assert_series_equal(
mask.vbt.signals(group_by=group_by)['g2'].stats(),
mask.vbt.signals.stats(column='g2', group_by=group_by)
)
stats_df = mask.vbt.signals.stats(agg_func=None)
assert stats_df.shape == (3, 22)
pd.testing.assert_index_equal(stats_df.index, mask.vbt.wrapper.columns)
pd.testing.assert_index_equal(stats_df.columns, stats_index)
@pytest.mark.parametrize(
"test_func,test_func_pd",
[
(lambda x, *args, **kwargs: x.AND(args, **kwargs), lambda x, y: x & y),
(lambda x, *args, **kwargs: x.OR(args, **kwargs), lambda x, y: x | y),
(lambda x, *args, **kwargs: x.XOR(args, **kwargs), lambda x, y: x ^ y)
],
)
def test_logical_funcs(self, test_func, test_func_pd):
pd.testing.assert_series_equal(
test_func(mask['a'].vbt.signals, True, [True, False, False, False, False]),
test_func_pd(test_func_pd(mask['a'], True), [True, False, False, False, False])
)
pd.testing.assert_frame_equal(
test_func(mask['a'].vbt.signals, True, [True, False, False, False, False], concat=True),
pd.concat((
test_func_pd(mask['a'], True),
test_func_pd(mask['a'], [True, False, False, False, False])
), axis=1, keys=[0, 1], names=['combine_idx'])
)
pd.testing.assert_frame_equal(
test_func(mask.vbt.signals, True, [[True], [False], [False], [False], [False]]),
test_func_pd(test_func_pd(mask, True),
np.broadcast_to([[True], [False], [False], [False], [False]], (5, 3)))
)
pd.testing.assert_frame_equal(
test_func(mask.vbt.signals, True, [[True], [False], [False], [False], [False]], concat=True),
pd.concat((
test_func_pd(mask, True),
test_func_pd(mask, np.broadcast_to([[True], [False], [False], [False], [False]], (5, 3)))
), axis=1, keys=[0, 1], names=['combine_idx'])
)
class TestFactory:
def test_entries(self):
@njit
def choice_nb(from_i, to_i, col, ts, in_out, n, arg, temp_idx_arr, kw):
in_out[from_i, col] = ts[from_i, col] * n + arg + kw
temp_idx_arr[0] = from_i
return temp_idx_arr[:1]
MySignals = vbt.SignalFactory(
mode='entries',
input_names=['ts2'],
in_output_names=['in_out2'],
param_names=['n2']
).from_choice_func(
entry_choice_func=choice_nb,
entry_settings=dict(
pass_inputs=['ts2'],
pass_in_outputs=['in_out2'],
pass_params=['n2'],
pass_kwargs=['temp_idx_arr2', ('kw2', 1000)],
pass_cache=True
),
in_output_settings=dict(
in_out2=dict(
dtype=np.float64
)
),
in_out2=np.nan,
var_args=True
)
my_sig = MySignals.run((5,), np.arange(5), [1, 0], 100)
pd.testing.assert_frame_equal(
my_sig.entries,
pd.DataFrame(np.array([
[True, True],
[False, False],
[False, False],
[False, False],
[False, False]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.in_out2,
pd.DataFrame(np.array([
[1100.0, 1100.0],
[np.nan, np.nan],
[np.nan, np.nan],
[np.nan, np.nan],
[np.nan, np.nan],
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
def test_exits(self):
@njit
def choice_nb(from_i, to_i, col, ts, in_out, n, arg, temp_idx_arr, kw):
in_out[from_i, col] = ts[from_i, col] * n + arg + kw
temp_idx_arr[0] = from_i
return temp_idx_arr[:1]
MySignals = vbt.SignalFactory(
mode='exits',
input_names=['ts2'],
in_output_names=['in_out2'],
param_names=['n2']
).from_choice_func(
exit_choice_func=choice_nb,
exit_settings=dict(
pass_inputs=['ts2'],
pass_in_outputs=['in_out2'],
pass_params=['n2'],
pass_kwargs=['temp_idx_arr2', ('kw2', 1000)],
pass_cache=True
),
in_output_settings=dict(
in_out2=dict(
dtype=np.float64
)
),
in_out2=np.nan,
var_args=True
)
e = np.array([True, False, True, False, True])
my_sig = MySignals.run(e, np.arange(5), [1, 0], 100)
pd.testing.assert_frame_equal(
my_sig.entries,
pd.DataFrame(np.array([
[True, True],
[False, False],
[True, True],
[False, False],
[True, True]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.exits,
pd.DataFrame(np.array([
[False, False],
[True, True],
[False, False],
[True, True],
[False, False]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.in_out2,
pd.DataFrame(np.array([
[np.nan, np.nan],
[1101.0, 1100.0],
[np.nan, np.nan],
[1103.0, 1100.0],
[np.nan, np.nan],
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
e = np.array([True, False, False, True, False, False])
my_sig = MySignals.run(e, np.arange(6), [1, 0], 100, wait=2)
pd.testing.assert_frame_equal(
my_sig.entries,
pd.DataFrame(np.array([
[True, True],
[False, False],
[False, False],
[True, True],
[False, False],
[False, False]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.exits,
pd.DataFrame(np.array([
[False, False],
[False, False],
[True, True],
[False, False],
[False, False],
[True, True]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.in_out2,
pd.DataFrame(np.array([
[np.nan, np.nan],
[np.nan, np.nan],
[1102.0, 1100.0],
[np.nan, np.nan],
[np.nan, np.nan],
[1105.0, 1100.0]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
def test_chain(self):
@njit
def choice_nb(from_i, to_i, col, ts, in_out, n, arg, temp_idx_arr, kw):
in_out[from_i, col] = ts[from_i, col] * n + arg + kw
temp_idx_arr[0] = from_i
return temp_idx_arr[:1]
MySignals = vbt.SignalFactory(
mode='chain',
input_names=['ts2'],
in_output_names=['in_out2'],
param_names=['n2']
).from_choice_func(
exit_choice_func=choice_nb,
exit_settings=dict(
pass_inputs=['ts2'],
pass_in_outputs=['in_out2'],
pass_params=['n2'],
pass_kwargs=['temp_idx_arr2', ('kw2', 1000)],
pass_cache=True
),
in_output_settings=dict(
in_out2=dict(
dtype=np.float64
)
),
in_out2=np.nan,
var_args=True
)
e = np.array([True, True, True, True, True])
my_sig = MySignals.run(e, np.arange(5), [1, 0], 100)
pd.testing.assert_frame_equal(
my_sig.entries,
pd.DataFrame(np.array([
[True, True],
[True, True],
[True, True],
[True, True],
[True, True]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.new_entries,
pd.DataFrame(np.array([
[True, True],
[False, False],
[True, True],
[False, False],
[True, True]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.exits,
pd.DataFrame(np.array([
[False, False],
[True, True],
[False, False],
[True, True],
[False, False]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.in_out2,
pd.DataFrame(np.array([
[np.nan, np.nan],
[1101.0, 1100.0],
[np.nan, np.nan],
[1103.0, 1100.0],
[np.nan, np.nan],
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
e = np.array([True, True, True, True, True, True])
my_sig = MySignals.run(e, np.arange(6), [1, 0], 100, wait=2)
pd.testing.assert_frame_equal(
my_sig.entries,
pd.DataFrame(np.array([
[True, True],
[True, True],
[True, True],
[True, True],
[True, True],
[True, True]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.new_entries,
pd.DataFrame(np.array([
[True, True],
[False, False],
[False, False],
[True, True],
[False, False],
[False, False]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.exits,
pd.DataFrame(np.array([
[False, False],
[False, False],
[True, True],
[False, False],
[False, False],
[True, True]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
pd.testing.assert_frame_equal(
my_sig.in_out2,
pd.DataFrame(np.array([
[np.nan, np.nan],
[np.nan, np.nan],
[1102.0, 1100.0],
[np.nan, np.nan],
[np.nan, np.nan],
[1105.0, 1100.0]
]), columns=pd.Index([1, 0], dtype='int64', name='custom_n2')
)
)
def test_both(self):
@njit
def cache_nb(ts1, ts2, in_out1, in_out2, n1, n2, arg0, temp_idx_arr0, kw0):
return arg0
@njit
def choice_nb(from_i, to_i, col, ts, in_out, n, arg, temp_idx_arr, kw, cache):
in_out[from_i, col] = ts[from_i, col] * n + arg + kw + cache
temp_idx_arr[0] = from_i
return temp_idx_arr[:1]
MySignals = vbt.SignalFactory(
input_names=['ts1', 'ts2'],
in_output_names=['in_out1', 'in_out2'],
param_names=['n1', 'n2']
).from_choice_func(
cache_func=cache_nb,
cache_settings=dict(
pass_inputs=['ts1', 'ts2'],
pass_in_outputs=['in_out1', 'in_out2'],
pass_params=['n1', 'n2'],
pass_kwargs=['temp_idx_arr0', ('kw0', 1000)]
),
entry_choice_func=choice_nb,
entry_settings=dict(
pass_inputs=['ts1'],
pass_in_outputs=['in_out1'],
pass_params=['n1'],
pass_kwargs=['temp_idx_arr1', ('kw1', 1000)],
pass_cache=True
),
exit_choice_func=choice_nb,
exit_settings=dict(
pass_inputs=['ts2'],
pass_in_outputs=['in_out2'],
pass_params=['n2'],
pass_kwargs=['temp_idx_arr2', ('kw2', 1000)],
pass_cache=True
),
in_output_settings=dict(
in_out1=dict(
dtype=np.float64
),
in_out2=dict(
dtype=np.float64
)
),
in_out1=np.nan,
in_out2=np.nan,
var_args=True,
require_input_shape=False
)
my_sig = MySignals.run(
np.arange(5), np.arange(5), [0, 1], [1, 0],
cache_args=(0,), entry_args=(100,), exit_args=(100,)
)
pd.testing.assert_frame_equal(
my_sig.entries,
pd.DataFrame(np.array([
[True, True],
[False, False],
[True, True],
[False, False],
[True, True]
]), columns=pd.MultiIndex.from_tuples(
[(0, 1), (1, 0)],
names=['custom_n1', 'custom_n2'])
)
)
pd.testing.assert_frame_equal(
my_sig.exits,
pd.DataFrame(np.array([
[False, False],
[True, True],
[False, False],
[True, True],
[False, False],
]), columns=pd.MultiIndex.from_tuples(
[(0, 1), (1, 0)],
names=['custom_n1', 'custom_n2'])
)
)
pd.testing.assert_frame_equal(
my_sig.in_out1,
pd.DataFrame(np.array([
[1100.0, 1100.0],
[np.nan, np.nan],
[1100.0, 1102.0],
[np.nan, np.nan],
[1100.0, 1104.0]
]), columns=pd.MultiIndex.from_tuples(
[(0, 1), (1, 0)],
names=['custom_n1', 'custom_n2'])
)
)
pd.testing.assert_frame_equal(
my_sig.in_out2,
pd.DataFrame(np.array([
[np.nan, np.nan],
[1101.0, 1100.0],
[np.nan, np.nan],
[1103.0, 1100.0],
[np.nan, np.nan],
]), columns=pd.MultiIndex.from_tuples(
[(0, 1), (1, 0)],
names=['custom_n1', 'custom_n2'])
)
)
my_sig = MySignals.run(
np.arange(7), np.arange(7), [0, 1], [1, 0],
cache_args=(0,), entry_args=(100,), exit_args=(100,),
entry_kwargs=dict(wait=2), exit_kwargs=dict(wait=2)
)
pd.testing.assert_frame_equal(
my_sig.entries,
pd.DataFrame(np.array([
[True, True],
[False, False],
[False, False],
[False, False],
[True, True],
[False, False],
[False, False]
]), columns=pd.MultiIndex.from_tuples(
[(0, 1), (1, 0)],
names=['custom_n1', 'custom_n2'])
)
)
pd.testing.assert_frame_equal(
my_sig.exits,
pd.DataFrame(np.array([
[False, False],
[False, False],
[True, True],
[False, False],
[False, False],
[False, False],
[True, True]
]), columns=pd.MultiIndex.from_tuples(
[(0, 1), (1, 0)],
names=['custom_n1', 'custom_n2'])
)
)
pd.testing.assert_frame_equal(
my_sig.in_out1,
pd.DataFrame(np.array([
[1100.0, 1100.0],
[np.nan, np.nan],
[np.nan, np.nan],
[np.nan, np.nan],
[1100.0, 1104.0],
[np.nan, np.nan],
[np.nan, np.nan]
]), columns=pd.MultiIndex.from_tuples(
[(0, 1), (1, 0)],
names=['custom_n1', 'custom_n2'])
)
)
pd.testing.assert_frame_equal(
my_sig.in_out2,
pd.DataFrame(np.array([
[np.nan, np.nan],
[np.nan, np.nan],
[1102.0, 1100.0],
[np.nan, np.nan],
[np.nan, np.nan],
[np.nan, np.nan],
[1106.0, 1100.0]
]), columns=pd.MultiIndex.from_tuples(
[(0, 1), (1, 0)],
names=['custom_n1', 'custom_n2'])
)
)
class TestGenerators:
def test_RAND(self):
rand = vbt.RAND.run(n=1, input_shape=(6,), seed=seed)
pd.testing.assert_series_equal(
rand.entries,
pd.Series(np.array([True, False, False, False, False, False]), name=1)
)
rand = vbt.RAND.run(n=[1, 2, 3], input_shape=(6,), seed=seed)
pd.testing.assert_frame_equal(
rand.entries,
pd.DataFrame(np.array([
[True, True, True],
[False, False, True],
[False, False, False],
[False, True, False],
[False, False, True],
[False, False, False]
]), columns=pd.Index([1, 2, 3], dtype='int64', name='rand_n')
)
)
rand = vbt.RAND.run(n=[np.array([1, 2]), np.array([3, 4])], input_shape=(8, 2), seed=seed)
pd.testing.assert_frame_equal(
rand.entries,
pd.DataFrame(np.array([
[False, False, True, False],
[True, False, False, False],
[False, False, False, True],
[False, True, True, False],
[False, False, False, False],
[False, False, False, True],
[False, False, True, True],
[False, True, False, True]
]), columns=pd.MultiIndex.from_tuples([
(1, 0),
(2, 1),
(3, 0),
(4, 1)
], names=['rand_n', None])
)
)
def test_RANDX(self):
randx = vbt.RANDX.run(mask, seed=seed)
pd.testing.assert_frame_equal(
randx.exits,
pd.DataFrame(np.array([
[False, False, False],
[False, False, False],
[True, True, False],
[False, False, False],
[True, False, True]
]), columns=mask.columns, index=mask.index)
)
def test_RANDNX(self):
randnx = vbt.RANDNX.run(n=1, input_shape=(6,), seed=seed)
pd.testing.assert_series_equal(
randnx.entries,
pd.Series(np.array([True, False, False, False, False, False]), name=1)
)
pd.testing.assert_series_equal(
randnx.exits,
pd.Series(np.array([False, True, False, False, False, False]), name=1)
)
randnx = vbt.RANDNX.run(n=[1, 2, 3], input_shape=(6,), seed=seed)
pd.testing.assert_frame_equal(
randnx.entries,
pd.DataFrame(np.array([
[True, True, True],
[False, False, False],
[False, True, True],
[False, False, False],
[False, False, True],
[False, False, False]
]), columns=pd.Index([1, 2, 3], dtype='int64', name='randnx_n')
)
)
pd.testing.assert_frame_equal(
randnx.exits,
pd.DataFrame(np.array([
[False, False, False],
[True, True, True],
[False, False, False],
[False, True, True],
[False, False, False],
[False, False, True]
]), columns=pd.Index([1, 2, 3], dtype='int64', name='randnx_n')
)
)
randnx = vbt.RANDNX.run(n=[np.array([1, 2]), np.array([3, 4])], input_shape=(8, 2), seed=seed)
pd.testing.assert_frame_equal(
randnx.entries,
pd.DataFrame(np.array([
[False, True, True, True],
[True, False, False, False],
[False, False, False, True],
[False, False, True, False],
[False, True, False, True],
[False, False, True, False],
[False, False, False, True],
[False, False, False, False]
]), columns=pd.MultiIndex.from_tuples([
(1, 0),
(2, 1),
(3, 0),
(4, 1)
], names=['randnx_n', None])
)
)
pd.testing.assert_frame_equal(
randnx.exits,
pd.DataFrame(np.array([
[False, False, False, False],
[False, False, True, True],
[False, False, False, False],
[False, True, False, True],
[False, False, True, False],
[True, False, False, True],
[False, False, True, False],
[False, True, False, True]
]), columns=pd.MultiIndex.from_tuples([
(1, 0),
(2, 1),
(3, 0),
(4, 1)
], names=['randnx_n', None])
)
)
def test_RPROB(self):
rprob = vbt.RPROB.run(prob=1, input_shape=(5,), seed=seed)
pd.testing.assert_series_equal(
rprob.entries,
pd.Series(np.array([True, True, True, True, True]), name=1)
)
rprob = vbt.RPROB.run(prob=[0, 0.5, 1], input_shape=(5,), seed=seed)
pd.testing.assert_frame_equal(
rprob.entries,
pd.DataFrame(np.array([
[False, True, True],
[False, True, True],
[False, False, True],
[False, False, True],
[False, False, True]
]), columns=pd.Index([0, 0.5, 1], dtype='float64', name='rprob_prob')
)
)
rprob = vbt.RPROB.run(prob=[np.array([0, 0.25]), np.array([0.75, 1])], input_shape=(5, 2), seed=seed)
pd.testing.assert_frame_equal(
rprob.entries,
pd.DataFrame(np.array([
[False, True, True, True],
[False, True, False, True],
[False, False, False, True],
[False, False, True, True],
[False, False, True, True]
]), columns=pd.MultiIndex.from_tuples([
('array_0', 0),
('array_0', 1),
('array_1', 0),
('array_1', 1)
], names=['rprob_prob', None])
)
)
def test_RPROBX(self):
rprobx = vbt.RPROBX.run(mask, prob=[0., 0.5, 1.], seed=seed)
pd.testing.assert_frame_equal(
rprobx.exits,
pd.DataFrame(np.array([
[False, False, False, False, False, False, False, False, False],
[False, False, False, False, False, False, True, False, False],
[False, False, False, False, True, False, False, True, False],
[False, False, False, False, False, False, False, False, True],
[False, False, False, False, False, False, True, False, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.0, 'a'),
(0.0, 'b'),
(0.0, 'c'),
(0.5, 'a'),
(0.5, 'b'),
(0.5, 'c'),
(1.0, 'a'),
(1.0, 'b'),
(1.0, 'c')
], names=['rprobx_prob', None])
)
)
def test_RPROBCX(self):
rprobcx = vbt.RPROBCX.run(mask, prob=[0., 0.5, 1.], seed=seed)
pd.testing.assert_frame_equal(
rprobcx.new_entries,
pd.DataFrame(np.array([
[True, False, False, True, False, False, True, False, False],
[False, True, False, False, True, False, False, True, False],
[False, False, True, False, False, True, False, False, True],
[False, False, False, True, False, False, True, False, False],
[False, False, False, False, True, False, False, True, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.0, 'a'),
(0.0, 'b'),
(0.0, 'c'),
(0.5, 'a'),
(0.5, 'b'),
(0.5, 'c'),
(1.0, 'a'),
(1.0, 'b'),
(1.0, 'c')
], names=['rprobcx_prob', None])
)
)
pd.testing.assert_frame_equal(
rprobcx.exits,
pd.DataFrame(np.array([
[False, False, False, False, False, False, False, False, False],
[False, False, False, False, False, False, True, False, False],
[False, False, False, True, False, False, False, True, False],
[False, False, False, False, True, True, False, False, True],
[False, False, False, False, False, False, True, False, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.0, 'a'),
(0.0, 'b'),
(0.0, 'c'),
(0.5, 'a'),
(0.5, 'b'),
(0.5, 'c'),
(1.0, 'a'),
(1.0, 'b'),
(1.0, 'c')
], names=['rprobcx_prob', None])
)
)
def test_RPROBNX(self):
rprobnx = vbt.RPROBNX.run(entry_prob=1., exit_prob=1., input_shape=(5,), seed=seed)
pd.testing.assert_series_equal(
rprobnx.entries,
pd.Series(np.array([True, False, True, False, True]), name=(1.0, 1.0))
)
pd.testing.assert_series_equal(
rprobnx.exits,
pd.Series(np.array([False, True, False, True, False]), name=(1.0, 1.0))
)
rprobnx = vbt.RPROBNX.run(
entry_prob=np.asarray([1., 0., 1., 0., 1.]),
exit_prob=np.asarray([0., 1., 0., 1., 0.]),
input_shape=(5,), seed=seed)
pd.testing.assert_series_equal(
rprobnx.entries,
pd.Series(np.array([True, False, True, False, True]), name=('array_0', 'array_0'))
)
pd.testing.assert_series_equal(
rprobnx.exits,
pd.Series(np.array([False, True, False, True, False]), name=('array_0', 'array_0'))
)
rprobnx = vbt.RPROBNX.run(entry_prob=[0.5, 1.], exit_prob=[1., 0.5], input_shape=(5,), seed=seed)
pd.testing.assert_frame_equal(
rprobnx.entries,
pd.DataFrame(np.array([
[True, True],
[False, False],
[False, True],
[False, False],
[True, False]
]), columns=pd.MultiIndex.from_tuples(
[(0.5, 1.0), (1.0, 0.5)],
names=['rprobnx_entry_prob', 'rprobnx_exit_prob'])
)
)
pd.testing.assert_frame_equal(
rprobnx.exits,
pd.DataFrame(np.array([
[False, False],
[True, True],
[False, False],
[False, False],
[False, False]
]), columns=pd.MultiIndex.from_tuples(
[(0.5, 1.0), (1.0, 0.5)],
names=['rprobnx_entry_prob', 'rprobnx_exit_prob'])
)
)
def test_STX(self):
stx = vbt.STX.run(mask, ts, 0.1)
pd.testing.assert_frame_equal(
stx.exits,
pd.DataFrame(np.array([
[False, False, False],
[True, False, False],
[False, True, False],
[False, False, False],
[False, False, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, 'a'),
(0.1, 'b'),
(0.1, 'c')
], names=['stx_stop', None])
)
)
stx = vbt.STX.run(mask, ts, np.asarray([0.1, 0.1, -0.1, -0.1, -0.1])[:, None])
pd.testing.assert_frame_equal(
stx.exits,
pd.DataFrame(np.array([
[False, False, False],
[True, False, False],
[False, True, False],
[False, False, True],
[True, False, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
('array_0', 'a'),
('array_0', 'b'),
('array_0', 'c')
], names=['stx_stop', None])
)
)
stx = vbt.STX.run(mask, ts, [0.1, 0.1, -0.1, -0.1], trailing=[False, True, False, True])
pd.testing.assert_frame_equal(
stx.exits,
pd.DataFrame(np.array([
[False, False, False, False, False, False, False, False, False, False, False, False],
[True, False, False, True, False, False, False, False, False, False, False, False],
[False, True, False, False, True, False, False, False, False, False, False, False],
[False, False, False, False, False, False, False, False, True, False, True, True],
[False, False, False, False, False, False, True, False, False, True, False, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, False, 'a'),
(0.1, False, 'b'),
(0.1, False, 'c'),
(0.1, True, 'a'),
(0.1, True, 'b'),
(0.1, True, 'c'),
(-0.1, False, 'a'),
(-0.1, False, 'b'),
(-0.1, False, 'c'),
(-0.1, True, 'a'),
(-0.1, True, 'b'),
(-0.1, True, 'c')
], names=['stx_stop', 'stx_trailing', None])
)
)
def test_STCX(self):
stcx = vbt.STCX.run(mask, ts, [0.1, 0.1, -0.1, -0.1], trailing=[False, True, False, True])
pd.testing.assert_frame_equal(
stcx.new_entries,
pd.DataFrame(np.array([
[True, False, False, True, False, False, True, False, False, True, False, False],
[False, True, False, False, True, False, False, True, False, False, True, False],
[False, False, True, False, False, True, False, False, True, False, False, True],
[True, False, False, True, False, False, False, False, False, False, False, False],
[False, True, False, False, True, False, False, False, False, False, True, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, False, 'a'),
(0.1, False, 'b'),
(0.1, False, 'c'),
(0.1, True, 'a'),
(0.1, True, 'b'),
(0.1, True, 'c'),
(-0.1, False, 'a'),
(-0.1, False, 'b'),
(-0.1, False, 'c'),
(-0.1, True, 'a'),
(-0.1, True, 'b'),
(-0.1, True, 'c')
], names=['stcx_stop', 'stcx_trailing', None])
)
)
pd.testing.assert_frame_equal(
stcx.exits,
pd.DataFrame(np.array([
[False, False, False, False, False, False, False, False, False, False, False, False],
[True, False, False, True, False, False, False, False, False, False, False, False],
[False, True, False, False, True, False, False, False, False, False, False, False],
[False, False, False, False, False, False, False, False, True, True, True, True],
[False, False, False, False, False, False, False, True, False, False, False, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, False, 'a'),
(0.1, False, 'b'),
(0.1, False, 'c'),
(0.1, True, 'a'),
(0.1, True, 'b'),
(0.1, True, 'c'),
(-0.1, False, 'a'),
(-0.1, False, 'b'),
(-0.1, False, 'c'),
(-0.1, True, 'a'),
(-0.1, True, 'b'),
(-0.1, True, 'c')
], names=['stcx_stop', 'stcx_trailing', None])
)
)
def test_OHLCSTX(self):
ohlcstx = vbt.OHLCSTX.run(
mask, price['open'], price['high'], price['low'], price['close'],
sl_stop=0.1
)
pd.testing.assert_frame_equal(
ohlcstx.exits,
pd.DataFrame(np.array([
[False, False, False],
[False, False, False],
[False, False, False],
[False, False, True],
[True, False, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, 'a'),
(0.1, 'b'),
(0.1, 'c')
], names=['ohlcstx_sl_stop', None])
)
)
pd.testing.assert_frame_equal(
ohlcstx.stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan],
[np.nan, np.nan, 10.8],
[9.9, np.nan, np.nan]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, 'a'),
(0.1, 'b'),
(0.1, 'c')
], names=['ohlcstx_sl_stop', None])
)
)
pd.testing.assert_frame_equal(
ohlcstx.stop_type,
pd.DataFrame(np.array([
[-1, -1, -1],
[-1, -1, -1],
[-1, -1, -1],
[-1, -1, 0],
[0, -1, -1]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, 'a'),
(0.1, 'b'),
(0.1, 'c')
], names=['ohlcstx_sl_stop', None])
)
)
ohlcstx = vbt.OHLCSTX.run(
mask, price['open'], price['high'], price['low'], price['close'],
sl_stop=[0.1, 0.1, np.nan], sl_trail=[False, True, False], tp_stop=[np.nan, np.nan, 0.1]
)
pd.testing.assert_frame_equal(
ohlcstx.exits,
pd.DataFrame(np.array([
[False, False, False, False, False, False, False, False, False],
[False, False, False, False, False, False, True, False, False],
[False, False, False, False, False, False, False, True, False],
[False, False, True, False, True, True, False, False, False],
[True, False, False, True, False, False, False, False, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, False, np.nan, 'a'),
(0.1, False, np.nan, 'b'),
(0.1, False, np.nan, 'c'),
(0.1, True, np.nan, 'a'),
(0.1, True, np.nan, 'b'),
(0.1, True, np.nan, 'c'),
(np.nan, False, 0.1, 'a'),
(np.nan, False, 0.1, 'b'),
(np.nan, False, 0.1, 'c')
], names=['ohlcstx_sl_stop', 'ohlcstx_sl_trail', 'ohlcstx_tp_stop', None])
)
)
pd.testing.assert_frame_equal(
ohlcstx.stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, 11., np.nan, np.nan],
[np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, 12.1, np.nan],
[np.nan, np.nan, 10.8, np.nan, 11.7, 10.8, np.nan, np.nan, np.nan],
[9.9, np.nan, np.nan, 9.9, np.nan, np.nan, np.nan, np.nan, np.nan]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, False, np.nan, 'a'),
(0.1, False, np.nan, 'b'),
(0.1, False, np.nan, 'c'),
(0.1, True, np.nan, 'a'),
(0.1, True, np.nan, 'b'),
(0.1, True, np.nan, 'c'),
(np.nan, False, 0.1, 'a'),
(np.nan, False, 0.1, 'b'),
(np.nan, False, 0.1, 'c')
], names=['ohlcstx_sl_stop', 'ohlcstx_sl_trail', 'ohlcstx_tp_stop', None])
)
)
pd.testing.assert_frame_equal(
ohlcstx.stop_type,
pd.DataFrame(np.array([
[-1, -1, -1, -1, -1, -1, -1, -1, -1],
[-1, -1, -1, -1, -1, -1, 2, -1, -1],
[-1, -1, -1, -1, -1, -1, -1, 2, -1],
[-1, -1, 0, -1, 1, 1, -1, -1, -1],
[0, -1, -1, 1, -1, -1, -1, -1, -1]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, False, np.nan, 'a'),
(0.1, False, np.nan, 'b'),
(0.1, False, np.nan, 'c'),
(0.1, True, np.nan, 'a'),
(0.1, True, np.nan, 'b'),
(0.1, True, np.nan, 'c'),
(np.nan, False, 0.1, 'a'),
(np.nan, False, 0.1, 'b'),
(np.nan, False, 0.1, 'c')
], names=['ohlcstx_sl_stop', 'ohlcstx_sl_trail', 'ohlcstx_tp_stop', None])
)
)
np.testing.assert_array_equal(
vbt.OHLCSTX.run(
mask, price['open'], price['high'], price['low'], price['close'],
sl_stop=[0.1, np.nan], sl_trail=False, tp_stop=[np.nan, 0.1], reverse=False
).exits.values,
vbt.OHLCSTX.run(
mask, price['open'], price['high'], price['low'], price['close'],
sl_stop=[np.nan, 0.1], sl_trail=False, tp_stop=[0.1, np.nan], reverse=True
).exits.values
)
def test_OHLCSTCX(self):
ohlcstcx = vbt.OHLCSTCX.run(
mask, price['open'], price['high'], price['low'], price['close'],
sl_stop=[0.1, 0.1, np.nan], sl_trail=[False, True, False], tp_stop=[np.nan, np.nan, 0.1]
)
pd.testing.assert_frame_equal(
ohlcstcx.exits,
pd.DataFrame(np.array([
[False, False, False, False, False, False, False, False, False],
[False, False, False, False, False, False, True, False, False],
[False, False, False, False, False, False, False, True, False],
[False, False, True, True, True, True, False, False, False],
[True, True, False, False, False, False, False, False, False]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, False, np.nan, 'a'),
(0.1, False, np.nan, 'b'),
(0.1, False, np.nan, 'c'),
(0.1, True, np.nan, 'a'),
(0.1, True, np.nan, 'b'),
(0.1, True, np.nan, 'c'),
(np.nan, False, 0.1, 'a'),
(np.nan, False, 0.1, 'b'),
(np.nan, False, 0.1, 'c')
], names=['ohlcstcx_sl_stop', 'ohlcstcx_sl_trail', 'ohlcstcx_tp_stop', None])
)
)
pd.testing.assert_frame_equal(
ohlcstcx.stop_price,
pd.DataFrame(np.array([
[np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan],
[np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, 11., np.nan, np.nan],
[np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, 12.1, np.nan],
[np.nan, np.nan, 10.8, 11.7, 11.7, 10.8, np.nan, np.nan, np.nan],
[9.0, 9.9, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, False, np.nan, 'a'),
(0.1, False, np.nan, 'b'),
(0.1, False, np.nan, 'c'),
(0.1, True, np.nan, 'a'),
(0.1, True, np.nan, 'b'),
(0.1, True, np.nan, 'c'),
(np.nan, False, 0.1, 'a'),
(np.nan, False, 0.1, 'b'),
(np.nan, False, 0.1, 'c')
], names=['ohlcstcx_sl_stop', 'ohlcstcx_sl_trail', 'ohlcstcx_tp_stop', None])
)
)
pd.testing.assert_frame_equal(
ohlcstcx.stop_type,
pd.DataFrame(np.array([
[-1, -1, -1, -1, -1, -1, -1, -1, -1],
[-1, -1, -1, -1, -1, -1, 2, -1, -1],
[-1, -1, -1, -1, -1, -1, -1, 2, -1],
[-1, -1, 0, 1, 1, 1, -1, -1, -1],
[0, 0, -1, -1, -1, -1, -1, -1, -1]
]), index=mask.index, columns=pd.MultiIndex.from_tuples([
(0.1, False, np.nan, 'a'),
(0.1, False, np.nan, 'b'),
(0.1, False, np.nan, 'c'),
(0.1, True, np.nan, 'a'),
(0.1, True, np.nan, 'b'),
(0.1, True, np.nan, 'c'),
(np.nan, False, 0.1, 'a'),
(np.nan, False, 0.1, 'b'),
(np.nan, False, 0.1, 'c')
], names=['ohlcstcx_sl_stop', 'ohlcstcx_sl_trail', 'ohlcstcx_tp_stop', None])
)
)