--- title: MVP (aka TSBERT) - Self-Supervised Pretraining of Time Series Models keywords: fastai sidebar: home_sidebar summary: "Masked Value Predictor callback used to predict time series step values after a binary mask has been applied." description: "Masked Value Predictor callback used to predict time series step values after a binary mask has been applied." nb_path: "nbs/063_callback.MVP.ipynb" ---
{% raw %}
{% endraw %} {% raw %}
{% endraw %} {% raw %}
{% endraw %} {% raw %}

create_subsequence_mask[source]

create_subsequence_mask(o, r=0.15, lm=3, stateful=True, sync=False)

{% endraw %} {% raw %}

create_variable_mask[source]

create_variable_mask(o, r=0.15)

{% endraw %} {% raw %}

create_future_mask[source]

create_future_mask(o, r=0.15, sync=False)

{% endraw %} {% raw %}

natural_mask[source]

natural_mask(o)

Applies natural missingness in a batch to non-nan values in the next sample

{% endraw %} {% raw %}
{% endraw %} {% raw %} 
t = torch.rand(16, 3, 100)
mask = create_subsequence_mask(t, sync=False)
test_eq(mask.shape, t.shape)
mask = create_subsequence_mask(t, sync=True)
test_eq(mask.shape, t.shape)
mask = create_variable_mask(t)
test_eq(mask.shape, t.shape)
mask = create_future_mask(t)
test_eq(mask.shape, t.shape)

/Users/nacho/opt/anaconda3/envs/py38/lib/python3.8/site-packages/torch/_tensor.py:579: UserWarning: floor_divide is deprecated, and will be removed in a future version of pytorch. It currently rounds toward 0 (like the 'trunc' function NOT 'floor'). This results in incorrect rounding for negative values.
To keep the current behavior, use torch.div(a, b, rounding_mode='trunc'), or for actual floor division, use torch.div(a, b, rounding_mode='floor'). (Triggered internally at  ../aten/src/ATen/native/BinaryOps.cpp:467.)
  return torch.floor_divide(other, self)
{% endraw %} {% raw %} 
o = torch.randn(2, 3, 4)
o[o>.5] = np.nan
test_eq(torch.isnan(natural_mask(o)).sum(), 0)
{% endraw %} {% raw %} 
t = torch.rand(16, 30, 100)
mask = create_subsequence_mask(t, r=.15) # default settings
test_eq(mask.dtype, torch.bool)
plt.figure(figsize=(10, 3))
plt.pcolormesh(mask[0], cmap='cool')
plt.title(f'sample 0 subsequence mask (sync=False) - default mean: {mask[0].float().mean().item():.3f}')
plt.show()
plt.figure(figsize=(10, 3))
plt.pcolormesh(mask[1], cmap='cool')
plt.title(f'sample 1 subsequence mask (sync=False) - default mean: {mask[1].float().mean().item():.3f}')
plt.show()
{% endraw %} {% raw %} 
t = torch.rand(16, 30, 100)
mask = create_subsequence_mask(t, r=.3) # 30% of values masked
test_eq(mask.dtype, torch.bool)
plt.figure(figsize=(10, 3))
plt.pcolormesh(mask[0], cmap='cool')
plt.title(f'sample 0 subsequence mask (r=.3) mean: {mask[0].float().mean().item():.3f}')
plt.show()
{% endraw %} {% raw %} 
t = torch.rand(16, 30, 100)
mask = create_subsequence_mask(t, lm=5) # average length of mask = 5 
test_eq(mask.dtype, torch.bool)
plt.figure(figsize=(10, 3))
plt.pcolormesh(mask[0], cmap='cool')
plt.title(f'sample 0 subsequence mask (lm=5) mean: {mask[0].float().mean().item():.3f}')
plt.show()
{% endraw %} {% raw %} 
t = torch.rand(16, 30, 100)
mask = create_subsequence_mask(t, stateful=False) # individual time steps masked 
test_eq(mask.dtype, torch.bool)
plt.figure(figsize=(10, 3))
plt.pcolormesh(mask[0], cmap='cool')
plt.title(f'per sample subsequence mask (stateful=False) mean: {mask[0].float().mean().item():.3f}')
plt.show()
{% endraw %} {% raw %} 
t = torch.rand(1, 30, 100)
mask = create_subsequence_mask(t, sync=True) # all time steps masked simultaneously
test_eq(mask.dtype, torch.bool)
plt.figure(figsize=(10, 3))
plt.pcolormesh(mask[0], cmap='cool')
plt.title(f'per sample subsequence mask (sync=True) mean: {mask[0].float().mean().item():.3f}')
plt.show()
{% endraw %} {% raw %} 
t = torch.rand(1, 30, 100)
mask = create_variable_mask(t) # masked variables
test_eq(mask.dtype, torch.bool)
plt.figure(figsize=(10, 3))
plt.pcolormesh(mask[0], cmap='cool')
plt.title(f'per sample variable mask mean: {mask[0].float().mean().item():.3f}')
plt.show()
{% endraw %} {% raw %} 
t = torch.rand(1, 30, 100)
mask = create_future_mask(t, r=.15, sync=True) # masked steps
test_eq(mask.dtype, torch.bool)
plt.figure(figsize=(10, 3))
plt.pcolormesh(mask[0], cmap='cool')
plt.title(f'future mask mean: {mask[0].float().mean().item():.3f}')
plt.show()
{% endraw %} {% raw %} 
t = torch.rand(1, 30, 100)
mask = create_future_mask(t, r=.15, sync=False) # masked steps
mask = create_future_mask(t, r=.15, sync=True) # masked steps
test_eq(mask.dtype, torch.bool)
plt.figure(figsize=(10, 3))
plt.pcolormesh(mask[0], cmap='cool')
plt.title(f'future mask mean: {mask[0].float().mean().item():.3f}')
plt.show()
{% endraw %} {% raw %}

create_mask[source]

create_mask(o, r=0.15, lm=3, stateful=True, sync=False, subsequence_mask=True, variable_mask=False, future_mask=False)

{% endraw %} {% raw %}
{% endraw %} {% raw %}

class MVP[source]

MVP(r:float=0.15, subsequence_mask:bool=True, lm:float=3.0, stateful:bool=True, sync:bool=False, variable_mask:bool=False, future_mask:bool=False, custom_mask:Optional=None, nan_to_num:int=0, window_size:Optional[tuple]=None, dropout:float=0.1, crit:callable=None, weights_path:Optional[str]=None, target_dir:str='./data/MVP', fname:str='model', save_best:bool=True, verbose:bool=False) :: Callback

Basic class handling tweaks of the training loop by changing a Learner in various events

{% endraw %} {% raw %}
{% endraw %}

Experiments

{% raw %} 
from fastai.data.transforms import *
from tsai.data.all import *
from tsai.models.utils import *
from tsai.models.layers import *
from tsai.learner import *
from tsai.models.TSTPlus import *
from tsai.models.InceptionTimePlus import *

[W ParallelNative.cpp:212] Warning: Cannot set number of intraop threads after parallel work has started or after set_num_threads call when using native parallel backend (function set_num_threads)
[W ParallelNative.cpp:212] Warning: Cannot set number of intraop threads after parallel work has started or after set_num_threads call when using native parallel backend (function set_num_threads)
{% endraw %} {% raw %} 
dsid = 'MoteStrain'
X, y, splits = get_UCR_data(dsid, split_data=False)
check_data(X, y, splits, False)
X[X<-1] = np.nan # This is to test the model works well even if nan values are passed through the dataloaders.

X      - shape: [1272 samples x 1 features x 84 timesteps]  type: memmap  dtype:float32  isnan: 0
y      - shape: (1272,)  type: memmap  dtype:<U1  n_classes: 2 (636 samples per class) ['1', '2']  isnan: False
splits - n_splits: 2 shape: [20, 1252]  overlap: False
{% endraw %} {% raw %} 
tfms  = [None, [Categorize()]]
batch_tfms = [TSStandardize(by_var=True)]
unlabeled_dls = get_ts_dls(X, splits=splits, tfms=tfms, batch_tfms=batch_tfms)
learn = ts_learner(unlabeled_dls, InceptionTimePlus, cbs=[MVP(fname=f'{dsid}', window_size=(.5, 1))]) # trained on variable window size
learn.fit_one_cycle(1, 3e-3)

/Users/nacho/opt/anaconda3/envs/py38/lib/python3.8/site-packages/torch/nn/init.py:388: UserWarning: Initializing zero-element tensors is a no-op
  warnings.warn("Initializing zero-element tensors is a no-op")
epoch train_loss valid_loss time
0 1.111950 1.174239 00:13 
best epoch:   0  val_loss: 1.174239 - pretrained weights_path='data/MVP/MoteStrain.pth'
{% endraw %} {% raw %} 
learn = ts_learner(unlabeled_dls, InceptionTimePlus, cbs=[MVP(weights_path=f'data/MVP/{dsid}.pth')])
learn.fit_one_cycle(1, 3e-3)

weights from data/MVP/MoteStrain.pth successfully transferred!
epoch train_loss valid_loss time
0 0.534770 1.184634 00:16 
best epoch:   0  val_loss: 1.184634 - pretrained weights_path='data/MVP/model.pth'
{% endraw %} {% raw %} 
learn.MVP.show_preds(sharey=True) # these preds are highly inaccurate as the model's been trained for just 1 epoch for testing purposes
{% endraw %} {% raw %} 
tfms  = [None, [Categorize()]]
batch_tfms = [TSStandardize(by_var=True), Nan2Value()]
labeled_dls = get_ts_dls(X, y, splits=splits, tfms=tfms, batch_tfms=batch_tfms, bs=64)
learn = ts_learner(labeled_dls, InceptionTimePlus, pretrained=True, weights_path=f'data/MVP/{dsid}.pth', metrics=accuracy)
learn.fit_one_cycle(1)

weights from data/MVP/MoteStrain.pth successfully transferred!
epoch train_loss valid_loss accuracy time
0 0.834448 0.736260 0.460863 00:17
{% endraw %} {% raw %} 
tfms  = [None, [Categorize()]]
batch_tfms = [TSStandardize(by_var=True), Nan2Value()]
unlabeled_dls = get_ts_dls(X, splits=splits, tfms=tfms, batch_tfms=batch_tfms, bs=64)
fname = f'{dsid}_test'
mvp = MVP(subsequence_mask=True, sync='random', variable_mask=True, future_mask=True, fname=fname)
learn = ts_learner(unlabeled_dls, InceptionTimePlus, metrics=accuracy, cbs=mvp) # Metrics will not be used!

/var/folders/42/4hhwknbd5kzcbq48tmy_gbp00000gn/T/ipykernel_26448/547766861.py:40: UserWarning: Only future_mask will be used
  warnings.warn("Only future_mask will be used")
/Users/nacho/opt/anaconda3/envs/py38/lib/python3.8/site-packages/torch/nn/init.py:388: UserWarning: Initializing zero-element tensors is a no-op
  warnings.warn("Initializing zero-element tensors is a no-op")
{% endraw %} {% raw %} 
tfms  = [None, [Categorize()]]
batch_tfms = [TSStandardize(by_var=True)]
unlabeled_dls = get_ts_dls(X, splits=splits, tfms=tfms, batch_tfms=batch_tfms, bs=64)
fname = f'{dsid}_test'
mvp = MVP(subsequence_mask=True, sync='random', variable_mask=True, future_mask=True, custom_mask=partial(create_future_mask, r=.15),
                fname=fname)
learn = ts_learner(unlabeled_dls, InceptionTimePlus, metrics=accuracy, cbs=mvp) # Metrics will not be used!

/var/folders/42/4hhwknbd5kzcbq48tmy_gbp00000gn/T/ipykernel_26448/547766861.py:38: UserWarning: Only custom_mask will be used
  warnings.warn("Only custom_mask will be used")
{% endraw %}