Time Series Data Augmentation

Functions used to transform TSTensors (Data Augmentation)

from tsai.data.core import TSCategorize
from tsai.data.external import get_UCR_data
from tsai.data.preprocessing import TSStandardize

dsid = 'NATOPS'
X, y, splits = get_UCR_data(dsid, return_split=False)
tfms = [None, TSCategorize()]
batch_tfms = TSStandardize()
dls = get_ts_dls(X, y, tfms=tfms, splits=splits, batch_tfms=batch_tfms, bs=128)
xb, yb = next(iter(dls.train))

---

source

TSIdentity

def TSIdentity(
    magnitude:NoneType=None, kwargs:VAR_KEYWORD
):

Applies the identity tfm to a TSTensor batch

test_eq(TSIdentity()(xb, split_idx=0).shape, xb.shape)

---

source

TSShuffle_HLs

def TSShuffle_HLs(
    magnitude:float=1.0, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Randomly shuffles HIs/LOs of an OHLC TSTensor batch

test_eq(TSShuffle_HLs()(xb, split_idx=0).shape, xb.shape)

---

source

TSShuffleSteps

def TSShuffleSteps(
    magnitude:float=1.0, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Randomly shuffles consecutive sequence datapoints in batch

t = TSTensor(torch.arange(11).float())
tt_ = []
for _ in range(1000):
    tt = TSShuffleSteps()(t, split_idx=0)
    test_eq(len(set(tt.tolist())), len(t))
    test_ne(tt, t)
    tt_.extend([t for i,t in enumerate(tt) if t!=i])
x, y = np.unique(tt_, return_counts=True) # This is to visualize distribution which should be equal for all and half for first and last items
plt.bar(x, y);

---

source

TSGaussianNoise

def TSGaussianNoise(
    magnitude:float=0.5, additive:bool=True, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies additive or multiplicative gaussian noise

test_eq(TSGaussianNoise(.1, additive=True)(xb, split_idx=0).shape, xb.shape)
test_eq(TSGaussianNoise(.1, additive=False)(xb, split_idx=0).shape, xb.shape)

---

source

TSMagMulNoise

def TSMagMulNoise(
    magnitude:int=1, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies multiplicative noise on the y-axis for each step of a TSTensor batch

---

source

TSMagAddNoise

def TSMagAddNoise(
    magnitude:int=1, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies additive noise on the y-axis for each step of a TSTensor batch

test_eq(TSMagAddNoise()(xb, split_idx=0).shape, xb.shape)
test_eq(TSMagMulNoise()(xb, split_idx=0).shape, xb.shape)
test_ne(TSMagAddNoise()(xb, split_idx=0), xb)
test_ne(TSMagMulNoise()(xb, split_idx=0), xb)

---

source

random_cum_linear_generator

def random_cum_linear_generator(
    o, magnitude:float=0.1
):

Call self as a function.

---

source

random_cum_noise_generator

def random_cum_noise_generator(
    o, magnitude:float=0.1, noise:NoneType=None
):

Call self as a function.

---

source

random_cum_curve_generator

def random_cum_curve_generator(
    o, magnitude:float=0.1, order:int=4, noise:NoneType=None
):

Call self as a function.

---

source

random_curve_generator

def random_curve_generator(
    o, magnitude:float=0.1, order:int=4, noise:NoneType=None
):

Call self as a function.

---

source

TSTimeNoise

def TSTimeNoise(
    magnitude:float=0.1, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies noise to each step in the x-axis of a TSTensor batch based on smooth random curve

test_eq(TSTimeNoise()(xb, split_idx=0).shape, xb.shape)
test_ne(TSTimeNoise()(xb, split_idx=0), xb)

---

source

TSMagWarp

def TSMagWarp(
    magnitude:float=0.02, ord:int=4, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies warping to the y-axis of a TSTensor batch based on a smooth random curve

test_eq(TSMagWarp()(xb, split_idx=0).shape, xb.shape)
test_ne(TSMagWarp()(xb, split_idx=0), xb)

---

source

TSTimeWarp

def TSTimeWarp(
    magnitude:float=0.1, ord:int=6, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies time warping to the x-axis of a TSTensor batch based on a smooth random curve

test_eq(TSTimeWarp()(xb, split_idx=0).shape, xb.shape)
test_ne(TSTimeWarp()(xb, split_idx=0), xb)

---

source

TSWindowWarp

def TSWindowWarp(
    magnitude:float=0.1, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies window slicing to the x-axis of a TSTensor batch based on a random linear curve based on https://halshs.archives-ouvertes.fr/halshs-01357973/document

test_eq(TSWindowWarp()(xb, split_idx=0).shape, xb.shape)

---

source

TSMagScalePerVar

def TSMagScalePerVar(
    magnitude:float=0.5, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies per_var scaling to the y-axis of a TSTensor batch based on a scalar

---

source

TSMagScale

def TSMagScale(
    magnitude:float=0.5, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies scaling to the y-axis of a TSTensor batch based on a scalar

test_eq(TSMagScale()(xb, split_idx=0).shape, xb.shape)
test_eq(TSMagScalePerVar()(xb, split_idx=0).shape, xb.shape)
test_ne(TSMagScale()(xb, split_idx=0), xb)
test_ne(TSMagScalePerVar()(xb, split_idx=0), xb)

---

source

test_interpolate

def test_interpolate(
    mode:str='linear'
):

Call self as a function.

# Run the test
test_interpolate('linear')

True

test_interpolate('nearest')

True

---

source

TSRandomResizedCrop

def TSRandomResizedCrop(
    magnitude:float=0.1, size:NoneType=None, scale:NoneType=None, ex:NoneType=None, mode:str='nearest',
    kwargs:VAR_KEYWORD
):

Randomly amplifies a sequence focusing on a random section of the steps

if test_interpolate('nearest'):
    test_eq(TSRandomResizedCrop(.5)(xb, split_idx=0).shape, xb.shape)
    test_ne(TSRandomResizedCrop(size=.8, scale=(.5, 1))(xb, split_idx=0).shape, xb.shape)
    test_ne(TSRandomResizedCrop(size=20, scale=(.5, 1))(xb, split_idx=0).shape, xb.shape)

---

source

TSWindowSlicing

def TSWindowSlicing(
    magnitude:float=0.1, ex:NoneType=None, mode:str='nearest', kwargs:VAR_KEYWORD
):

Randomly extracts an resize a ts slice based on https://halshs.archives-ouvertes.fr/halshs-01357973/document

if test_interpolate('nearest'):
    test_eq(TSWindowSlicing()(xb, split_idx=0).shape, xb.shape)
    test_ne(TSWindowSlicing()(xb, split_idx=0), xb)

---

source

TSRandomZoomOut

def TSRandomZoomOut(
    magnitude:float=0.1, ex:NoneType=None, mode:str='nearest', kwargs:VAR_KEYWORD
):

Randomly compresses a sequence on the x-axis

if test_interpolate('nearest'):
    test_eq(TSRandomZoomOut(.5)(xb, split_idx=0).shape, xb.shape)#

---

source

TSRandomTimeScale

def TSRandomTimeScale(
    magnitude:float=0.1, ex:NoneType=None, mode:str='nearest', kwargs:VAR_KEYWORD
):

Randomly amplifies/ compresses a sequence on the x-axis keeping the same length

if test_interpolate('nearest'):
    test_eq(TSRandomTimeScale(.5)(xb, split_idx=0).shape, xb.shape)

---

source

TSRandomTimeStep

def TSRandomTimeStep(
    magnitude:float=0.02, ex:NoneType=None, mode:str='nearest', kwargs:VAR_KEYWORD
):

Compresses a sequence on the x-axis by randomly selecting sequence steps and interpolating to previous size

if test_interpolate('nearest'):
    test_eq(TSRandomTimeStep()(xb, split_idx=0).shape, xb.shape)

---

source

TSResampleSteps

def TSResampleSteps(
    step_pct:float=1.0, same_seq_len:bool=True, magnitude:NoneType=None, kwargs:VAR_KEYWORD
):

Transform that randomly selects and sorts sequence steps (with replacement) maintaining the sequence length

test_eq(TSResampleSteps(step_pct=.9, same_seq_len=False)(xb, split_idx=0).shape[-1], round(.9*xb.shape[-1]))
test_eq(TSResampleSteps(step_pct=.9, same_seq_len=True)(xb, split_idx=0).shape[-1], xb.shape[-1])

---

source

TSBlur

def TSBlur(
    magnitude:float=1.0, ex:NoneType=None, filt_len:NoneType=None, kwargs:VAR_KEYWORD
):

Blurs a sequence applying a filter of type [1, 0, 1]

test_eq(TSBlur(filt_len=7)(xb, split_idx=0).shape, xb.shape)
test_ne(TSBlur()(xb, split_idx=0), xb)

---

source

TSSmooth

def TSSmooth(
    magnitude:float=1.0, ex:NoneType=None, filt_len:NoneType=None, kwargs:VAR_KEYWORD
):

Smoothens a sequence applying a filter of type [1, 5, 1]

test_eq(TSSmooth(filt_len=7)(xb, split_idx=0).shape, xb.shape)
test_ne(TSSmooth()(xb, split_idx=0), xb)

---

source

TSFreqDenoise

def TSFreqDenoise(
    magnitude:float=0.1, ex:NoneType=None, wavelet:str='db4', level:int=2, thr:NoneType=None, thr_mode:str='hard',
    pad_mode:str='per', kwargs:VAR_KEYWORD
):

Denoises a sequence applying a wavelet decomposition method

---

source

maddest

def maddest(
    d, axis:NoneType=None
):

Call self as a function.

try: import pywt
except ImportError: pass

if 'pywt' in dir():
    test_eq(TSFreqDenoise()(xb, split_idx=0).shape, xb.shape)
    test_ne(TSFreqDenoise()(xb, split_idx=0), xb)

---

source

TSRandomFreqNoise

def TSRandomFreqNoise(
    magnitude:float=0.1, ex:NoneType=None, wavelet:str='db4', level:int=2, mode:str='constant', kwargs:VAR_KEYWORD
):

Applys random noise using a wavelet decomposition method

if 'pywt' in dir():
    test_eq(TSRandomFreqNoise()(xb, split_idx=0).shape, xb.shape)

---

source

TSRandomResizedLookBack

def TSRandomResizedLookBack(
    magnitude:float=0.1, mode:str='nearest', kwargs:VAR_KEYWORD
):

Selects a random number of sequence steps starting from the end and return an output of the same shape

if test_interpolate('nearest'):
    for i in range(100):
        o = TSRandomResizedLookBack()(xb, split_idx=0)
        test_eq(o.shape[-1], xb.shape[-1])

---

source

TSRandomLookBackOut

def TSRandomLookBackOut(
    magnitude:float=0.1, kwargs:VAR_KEYWORD
):

Selects a random number of sequence steps starting from the end and set them to zero

for i in range(100):
    o = TSRandomLookBackOut()(xb, split_idx=0)
    test_eq(o.shape[-1], xb.shape[-1])

---

source

TSVarOut

def TSVarOut(
    magnitude:float=0.05, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Set the value of a random number of variables to zero

test_eq(TSVarOut()(xb, split_idx=0).shape, xb.shape)

---

source

TSCutOut

def TSCutOut(
    magnitude:float=0.05, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Sets a random section of the sequence to zero

test_eq(TSCutOut()(xb, split_idx=0).shape, xb.shape)

---

source

TSTimeStepOut

def TSTimeStepOut(
    magnitude:float=0.05, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Sets random sequence steps to zero

test_eq(TSTimeStepOut()(xb, split_idx=0).shape, xb.shape)

---

source

TSRandomCropPad

def TSRandomCropPad(
    magnitude:float=0.05, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Crops a section of the sequence of a random length

test_eq(TSRandomCropPad()(xb, split_idx=0).shape, xb.shape)

---

source

TSMaskOut

def TSMaskOut(
    magnitude:float=0.1, compensate:bool=False, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies a random mask

test_eq(TSMaskOut()(xb, split_idx=0).shape, xb.shape)
test_ne(TSMaskOut()(xb, split_idx=0), xb)

---

source

TSInputDropout

def TSInputDropout(
    magnitude:float=0.0, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies input dropout with required_grad=False

test_eq(TSInputDropout(.1)(xb, split_idx=0).shape, xb.shape)
test_ne(TSInputDropout(.1)(xb, split_idx=0), xb)

---

source

TSTranslateX

def TSTranslateX(
    magnitude:float=0.1, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Moves a selected sequence window a random number of steps

test_eq(TSTranslateX()(xb, split_idx=0).shape, xb.shape)

---

source

TSRandomShift

def TSRandomShift(
    magnitude:float=0.02, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Shifts and splits a sequence

test_eq(TSRandomShift()(xb, split_idx=0).shape, xb.shape)

---

source

TSHorizontalFlip

def TSHorizontalFlip(
    magnitude:float=1.0, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Flips the sequence along the x-axis

test_eq(TSHorizontalFlip()(xb, split_idx=0).shape, xb.shape)
test_ne(TSHorizontalFlip()(xb, split_idx=0), xb)

---

source

TSRandomTrend

def TSRandomTrend(
    magnitude:float=0.1, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Randomly rotates the sequence along the z-axis

test_eq(TSRandomTrend()(xb, split_idx=0).shape, xb.shape)

---

source

TSVerticalFlip

def TSVerticalFlip(
    magnitude:float=1.0, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Applies a negative value to the time sequence

test_eq(TSVerticalFlip()(xb, split_idx=0).shape, xb.shape)
test_ne(TSVerticalFlip()(xb, split_idx=0), xb)

---

source

TSResize

def TSResize(
    magnitude:float=-0.5, size:NoneType=None, ex:NoneType=None, mode:str='nearest', kwargs:VAR_KEYWORD
):

Resizes the sequence length of a time series

if test_interpolate('nearest'):
    for sz in np.linspace(.2, 2, 10): test_eq(TSResize(sz)(xb, split_idx=0).shape[-1], int(round(xb.shape[-1]*(1+sz))))
    test_ne(TSResize(1)(xb, split_idx=0), xb)

---

source

TSRandomSize

def TSRandomSize(
    magnitude:float=0.1, ex:NoneType=None, mode:str='nearest', kwargs:VAR_KEYWORD
):

Randomly resizes the sequence length of a time series

if test_interpolate('nearest'):
    seq_len_ = []
    for i in range(100):
        o = TSRandomSize(.5)(xb, split_idx=0)
        seq_len_.append(o.shape[-1])
    test_lt(min(seq_len_), xb.shape[-1])
    test_gt(max(seq_len_), xb.shape[-1])

---

source

TSRandomLowRes

def TSRandomLowRes(
    magnitude:float=0.5, ex:NoneType=None, mode:str='nearest', kwargs:VAR_KEYWORD
):

Randomly resizes the sequence length of a time series to a lower resolution

---

source

TSDownUpScale

def TSDownUpScale(
    magnitude:float=0.5, ex:NoneType=None, mode:str='nearest', kwargs:VAR_KEYWORD
):

Downscales a time series and upscales it again to previous sequence length

if test_interpolate('nearest'):
    test_eq(TSDownUpScale()(xb, split_idx=0).shape, xb.shape)

---

source

TSRandomDownUpScale

def TSRandomDownUpScale(
    magnitude:float=0.5, ex:NoneType=None, mode:str='nearest', kwargs:VAR_KEYWORD
):

Randomly downscales a time series and upscales it again to previous sequence length

if test_interpolate('nearest'):
    test_eq(TSRandomDownUpScale()(xb, split_idx=0).shape, xb.shape)
    test_ne(TSDownUpScale()(xb, split_idx=0), xb)
    test_eq(TSDownUpScale()(xb, split_idx=1), xb)

---

source

TSRandomConv

def TSRandomConv(
    magnitude:float=0.05, ex:NoneType=None, ks:list=[1, 3, 5, 7], kwargs:VAR_KEYWORD
):

Applies a convolution with a random kernel and random weights with required_grad=False

for i in range(5):
    o = TSRandomConv(magnitude=0.05, ex=None, ks=[1, 3, 5, 7])(xb, split_idx=0)
    test_eq(o.shape, xb.shape)

---

source

TSRandom2Value

def TSRandom2Value(
    magnitude:float=0.1, sel_vars:NoneType=None, sel_steps:NoneType=None, static:bool=False, value:float=nan,
    kwargs:VAR_KEYWORD
):

Randomly sets selected variables of type TSTensor to predefined value (default: np.nan)

t = TSTensor(torch.ones(2, 3, 10))
TSRandom2Value(magnitude=0.5, sel_vars=None, sel_steps=None, static=False, value=0)(t, split_idx=0).data

tensor([[[0., 1., 0., 1., 1., 1., 0., 1., 1., 1.],
         [0., 0., 0., 0., 0., 0., 0., 0., 1., 0.],
         [0., 1., 1., 0., 0., 0., 1., 0., 0., 1.]],

        [[1., 1., 1., 0., 0., 1., 1., 1., 0., 1.],
         [0., 1., 1., 0., 1., 0., 0., 0., 0., 0.],
         [1., 1., 0., 1., 0., 0., 1., 0., 0., 1.]]])

t = TSTensor(torch.ones(2, 3, 10))
TSRandom2Value(magnitude=0.5, sel_vars=[1], sel_steps=slice(-5, None), static=False, value=0)(t, split_idx=0).data

tensor([[[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 1., 1., 1., 1., 1., 0., 0., 0., 1.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.]],

        [[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 1., 1., 1., 1., 0., 0., 0., 0., 1.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.]]])

t = TSTensor(torch.ones(2, 3, 10))
TSRandom2Value(magnitude=0.5, sel_vars=[1], sel_steps=None, static=True, value=0)(t, split_idx=0).data

tensor([[[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.]],

        [[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.]]])

t = TSTensor(torch.ones(2, 3, 10))
TSRandom2Value(magnitude=1, sel_vars=1, sel_steps=None, static=False, value=0)(t, split_idx=0).data

tensor([[[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [0., 0., 0., 0., 0., 0., 0., 0., 0., 0.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.]],

        [[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [0., 0., 0., 0., 0., 0., 0., 0., 0., 0.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.]]])

t = TSTensor(torch.ones(2, 3, 10))
TSRandom2Value(magnitude=1, sel_vars=[1,2], sel_steps=None, static=False, value=0)(t, split_idx=0).data

tensor([[[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [0., 0., 0., 0., 0., 0., 0., 0., 0., 0.],
         [0., 0., 0., 0., 0., 0., 0., 0., 0., 0.]],

        [[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [0., 0., 0., 0., 0., 0., 0., 0., 0., 0.],
         [0., 0., 0., 0., 0., 0., 0., 0., 0., 0.]]])

t = TSTensor(torch.ones(2, 3, 10))
TSRandom2Value(magnitude=1, sel_vars=1, sel_steps=[1,3,5], static=False, value=0)(t, split_idx=0).data

tensor([[[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 0., 1., 0., 1., 0., 1., 1., 1., 1.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.]],

        [[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 0., 1., 0., 1., 0., 1., 1., 1., 1.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1.]]])

t = TSTensor(torch.ones(2, 3, 10))
TSRandom2Value(magnitude=1, sel_vars=[1,2], sel_steps=[1,3,5], static=False, value=0)(t, split_idx=0).data

tensor([[[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 0., 1., 0., 1., 0., 1., 1., 1., 1.],
         [1., 0., 1., 0., 1., 0., 1., 1., 1., 1.]],

        [[1., 1., 1., 1., 1., 1., 1., 1., 1., 1.],
         [1., 0., 1., 0., 1., 0., 1., 1., 1., 1.],
         [1., 0., 1., 0., 1., 0., 1., 1., 1., 1.]]])

t = TSTensor(torch.ones(2,3,4))
TSRandom2Value(magnitude=.5, sel_vars=[0,2])(t, split_idx=0).data

tensor([[[1., 1., 1., 1.],
         [1., 1., 1., 1.],
         [1., 1., 1., nan]],

        [[nan, nan, 1., 1.],
         [1., 1., 1., 1.],
         [1., nan, nan, 1.]]])

t = TSTensor(torch.ones(2,3,4))
TSRandom2Value(magnitude=.5, sel_steps=slice(2, None))(t, split_idx=0).data

tensor([[[1., 1., nan, 1.],
         [1., 1., nan, nan],
         [1., 1., 1., nan]],

        [[1., 1., nan, 1.],
         [1., 1., 1., 1.],
         [1., 1., 1., nan]]])

t = TSTensor(torch.ones(2,3,100))
test_gt(np.isnan(TSRandom2Value(magnitude=.5)(t, split_idx=0)).sum().item(), 0)
t = TSTensor(torch.ones(2,3,100))
test_gt(np.isnan(TSRandom2Value(magnitude=.5, sel_vars=[0,2])(t, split_idx=0)[:, [0,2]]).sum().item(), 0)
t = TSTensor(torch.ones(2,3,100))
test_eq(np.isnan(TSRandom2Value(magnitude=.5, sel_vars=[0,2])(t, split_idx=0)[:, 1]).sum().item(), 0)

/var/folders/yw/1vck7tm93_z1z0bftrw65hbw0000gn/T/ipykernel_22669/3870966226.py:2: DeprecationWarning: __array_wrap__ must accept context and return_scalar arguments (positionally) in the future. (Deprecated NumPy 2.0)
  test_gt(np.isnan(TSRandom2Value(magnitude=.5)(t, split_idx=0)).sum().item(), 0)
/var/folders/yw/1vck7tm93_z1z0bftrw65hbw0000gn/T/ipykernel_22669/3870966226.py:4: DeprecationWarning: __array_wrap__ must accept context and return_scalar arguments (positionally) in the future. (Deprecated NumPy 2.0)
  test_gt(np.isnan(TSRandom2Value(magnitude=.5, sel_vars=[0,2])(t, split_idx=0)[:, [0,2]]).sum().item(), 0)
/var/folders/yw/1vck7tm93_z1z0bftrw65hbw0000gn/T/ipykernel_22669/3870966226.py:6: DeprecationWarning: __array_wrap__ must accept context and return_scalar arguments (positionally) in the future. (Deprecated NumPy 2.0)
  test_eq(np.isnan(TSRandom2Value(magnitude=.5, sel_vars=[0,2])(t, split_idx=0)[:, 1]).sum().item(), 0)

---

source

TSMask2Value

def TSMask2Value(
    mask_fn, value:float=nan, sel_vars:NoneType=None, kwargs:VAR_KEYWORD
):

Randomly sets selected variables of type TSTensor to predefined value (default: np.nan)

t = TSTensor(torch.ones(2,3,100))
def _mask_fn(o, r=.15, value=np.nan):
    return torch.rand_like(o) > (1-r)
test_gt(np.isnan(TSMask2Value(_mask_fn)(t, split_idx=0)).sum().item(), 0)

/var/folders/yw/1vck7tm93_z1z0bftrw65hbw0000gn/T/ipykernel_22669/2303918966.py:4: DeprecationWarning: __array_wrap__ must accept context and return_scalar arguments (positionally) in the future. (Deprecated NumPy 2.0)
  test_gt(np.isnan(TSMask2Value(_mask_fn)(t, split_idx=0)).sum().item(), 0)

---

source

TSSelfDropout

def TSSelfDropout(
    p:float=1.0, # Probability of applying Transform
    nm:str=None, before_call:Callable=None, # Optional batchwise preprocessing function
    kwargs:VAR_KEYWORD
):

Applies dropout to a tensor with nan values by rotating axis=0 inplace

---

source

self_mask

def self_mask(
    o
):

Call self as a function.

t = TSTensor(torch.ones(2,3,100))
mask = torch.rand_like(t) > .7
t[mask] = np.nan
nan_perc = np.isnan(t).float().mean().item()
t2 = TSSelfDropout()(t, split_idx=0)
test_gt(torch.isnan(t2).float().mean().item(), nan_perc)
nan_perc, torch.isnan(t2).float().mean().item()

/var/folders/yw/1vck7tm93_z1z0bftrw65hbw0000gn/T/ipykernel_22669/2944010346.py:4: DeprecationWarning: __array_wrap__ must accept context and return_scalar arguments (positionally) in the future. (Deprecated NumPy 2.0)
  nan_perc = np.isnan(t).float().mean().item()

(0.3100000023841858, 0.5166666507720947)

---

source

RandAugment

def RandAugment(
    tfms:list, N:int=1, M:int=3, kwargs:VAR_KEYWORD
):

A transform that before_call its state at each __call__

test_ne(RandAugment(TSMagAddNoise, N=5, M=10)(xb, split_idx=0), xb)

---

source

TestTfm

def TestTfm(
    tfm, magnitude:float=1.0, ex:NoneType=None, kwargs:VAR_KEYWORD
):

Utility class to test the output of selected tfms during training

---

source

get_tfm_name

def get_tfm_name(
    tfm
):

Call self as a function.

test_eq(get_tfm_name(partial(TSMagScale()))==get_tfm_name((partial(TSMagScale()), 0.1, .05))==get_tfm_name(TSMagScale())==get_tfm_name((TSMagScale(), 0.1, .05)), True)

all_TS_randaugs_names = [get_tfm_name(t) for t in all_TS_randaugs]