Source code for tsaug.trend

"""
Add trend
"""
from typing import Tuple, Optional, Callable

import numpy as np
from scipy.interpolate import CubicSpline
from .dimensionalize import dimensionalize
from .augmentor import _Augmentor


[docs]@dimensionalize def trend( X: np.ndarray, Y: Optional[np.ndarray] = None, anchors: Optional[np.ndarray] = None, ) -> Tuple[np.ndarray, Optional[np.ndarray]]: """Add trend to time series. Given m anchors, the trend of a series is generated by cublic spline interpolation over all anchors that are uniformly distributed. Parameters ---------- X : numpy.ndarray Time series to be augmented. Matrix with shape (n,), (N, n) or (N, n, c), where n is the length of each series, N is the number of series, and c is the number of channels. Y : numpy.ndarray, optional Binary labels of time series, where 0 represents a normal point and 1 represents an anomalous points. Matrix with shape (n,), (N, n) or (N, n, cl), where n is the length of each series, N is the number of series, and cl is the number of classes (i.e. types of anomaly). Default: None. anchors : numpy.ndarray Anchor points from which trend is interpolated. An array with shape (m,), (N, m), or (N, m, c), where m is the number of anchor points of each series, N is the number of series, and c is the number of channels. Default: None, i.e. no trend added. Returns ------- tuple (numpy.ndarray, numpy.ndarray) Augmented time series and augmented labels (if argument `Y` exists). """ N = ( 0 ) # type: int # NOTE: this is a horrible hack to type hint for Python 3.5 n = 0 # type: int c = 0 # type: int N, n, c = X.shape if anchors is None: anchors = np.zeros((N, 2, c)) if anchors.ndim == 1: anchors = np.tile(anchors, (N, 1, c)) elif anchors.ndim == 2: anchors = np.tile(anchors, (1, 1, c)) if ( (anchors.ndim != 3) or (anchors.shape[0] != N) or (anchors.shape[2] != c) ): raise ValueError("Wrong shape of anchors.") m = anchors.shape[1] # type: int interpFuncs = CubicSpline( np.linspace(0, n, m), anchors, axis=1 ) # type: Callable X_aug = interpFuncs(np.arange(n)) + X # type: np.ndarray if Y is None: Y_aug = None # type: Optional[np.ndarray] else: Y_aug = Y.copy() return X_aug, Y_aug
[docs]@dimensionalize def random_trend( X: np.ndarray, Y: Optional[np.ndarray] = None, num_anchors: int = 5, min_anchor: float = 0.0, max_anchor: float = 100.0, random_seed: Optional[int] = None, ) -> Tuple[np.ndarray, Optional[np.ndarray]]: """Add random trend to time series. Given m anchors, the trend of a series is generated by cublic spline interpolation over all anchors that are uniformly distributed. Anchors are randomly sampled between given bounds. Parameters ---------- X : numpy.ndarray Time series to be augmented. Matrix with shape (n,), (N, n) or (N, n, c), where n is the length of each series, N is the number of series, and c is the number of channels. Y : numpy.ndarray, optional Binary labels of time series, where 0 represents a normal point and 1 represents an anomalous points. Matrix with shape (n,), (N, n) or (N, n, cl), where n is the length of each series, N is the number of series, and cl is the number of classes (i.e. types of anomaly). Default: None. num_anchors : int, optional Number of anchors to sample for each series. Default: 5. min_anchor : float, optional Minimal value of anchors. Default: 0.0. max_anchor : float, optional Maximal value of anchors. Default: 100.0. random_seed : int, optional Random seed used to initialize the pseudo-random number generator. Default: None. Returns ------- tuple (numpy.ndarray, numpy.ndarray) Augmented time series and augmented labels (if argument `Y` exists). """ N = ( 0 ) # type: int # NOTE: this is a horrible hack to type hint for Python 3.5 n = 0 # type: int c = 0 # type: int N, n, c = X.shape rand = np.random.RandomState(random_seed) anchors = np.cumsum( rand.normal(size=(N, num_anchors, c)), axis=1 ) # type: np.ndarray anchors = (anchors - anchors.min(axis=1).reshape((N, 1, c))) / ( anchors.max(axis=1) - anchors.min(axis=1) ).reshape((N, 1, c)) anchors = anchors * (max_anchor - min_anchor) + min_anchor return trend(X, Y, anchors)
[docs]class Trend(_Augmentor): """ Augmentor that adds trend to time series. Given m anchors, the trend of a series is generated by cublic spline interpolation over all anchors that are uniformly distributed. Parameters ---------- anchors : numpy.ndarray Anchor points from which trend is interpolated. An array with shape (m,), (N, m), or (N, m, c), where m is the number of anchor points of each series, N is the number of series, and c is the number of channels. Default: None, i.e. no trend added. """ def __init__(self, anchors: Optional[np.ndarray] = None) -> None: super().__init__( augmentor_func=trend, is_random=False, anchors=anchors ) @property def anchors(self) -> Optional[np.ndarray]: return self._params["anchors"] @anchors.setter def anchors(self, anchors: Optional[np.ndarray]) -> None: self._params["anchors"] = anchors
[docs]class RandomTrend(_Augmentor): """ Augmentor to add random trend to time series. Given m anchors, the trend of a series is generated by cublic spline interpolation over all anchors that are uniformly distributed. Anchors are randomly sampled between given bounds. Parameters ---------- num_anchors : int, optional Number of anchors to sample for each series. Default: 5. min_anchor : float, optional Minimal value of anchors. Default: 0.0. max_anchor : float, optional Maximal value of anchors. Default: 100.0. random_seed : int, optional Random seed used to initialize the pseudo-random number generator. Default: None. """ def __init__( self, num_anchors: int = 5, min_anchor: float = 0.0, max_anchor: float = 100.0, random_seed: Optional[int] = None, ) -> None: super().__init__( augmentor_func=random_trend, is_random=True, num_anchors=num_anchors, min_anchor=min_anchor, max_anchor=max_anchor, random_seed=random_seed, ) @property def num_anchors(self) -> int: return self._params["num_anchors"] @num_anchors.setter def num_anchors(self, num_anchors: int) -> None: self._params["num_anchors"] = num_anchors @property def min_anchor(self) -> float: return self._params["min_anchor"] @min_anchor.setter def min_anchor(self, min_anchor: float) -> None: self._params["min_anchor"] = min_anchor @property def max_anchor(self) -> float: return self._params["max_anchor"] @max_anchor.setter def max_anchor(self, max_anchor: float) -> None: self._params["max_anchor"] = max_anchor @property def random_seed(self) -> Optional[int]: return self._params["random_seed"] @random_seed.setter def random_seed(self, random_seed: Optional[int]) -> None: self._params["random_seed"] = random_seed