Dev 005 (#17)

Adding simple generators for time-series: trends, seasons, changepoints

Author: siebert-julien

README.md

@@ -17,10 +17,10 @@ Import badgers as any other library and start using it:
 
 ```python
 from sklearn.datasets import make_blobs
-from badgers.generators.tabular_data.noise import GaussianNoiseGenerator
+from badgers.generators.tabular_data.noise import GlobalGaussianNoiseGenerator
 
 X, y = make_blobs()
-trf = GaussianNoiseGenerator(noise_std=0.5)
+trf = GlobalGaussianNoiseGenerator(noise_std=0.5)
 Xt, yt = trf.generate(X,y)
 ```
 

badgers/generators/time_series/changepoints.py

@@ -0,0 +1,60 @@
+import abc
+from typing import Tuple
+
+from numpy.random import default_rng
+
+from badgers.core.base import GeneratorMixin
+
+
+class ChangePointGenerator(GeneratorMixin):
+    """
+    Base class for generators that generate changepoints in time-series data
+    """
+
+    def __init__(self, random_generator=default_rng(seed=0), n_changepoints: int = 10):
+        """
+        :param random_generator: a random number generator
+        :param n_outliers: the number of outliers to generate
+        """
+        self.random_generator = random_generator
+        self.n_changepoints = n_changepoints
+        self.changepoints = None
+
+    @abc.abstractmethod
+    def generate(self, X, y, **params) -> Tuple:
+        pass
+
+
+class RandomChangeInMeanGenerator(ChangePointGenerator):
+    """
+    Generate randomly change in mean changepoints
+    """
+
+    def __init__(self, random_generator=default_rng(seed=0), n_changepoints: int = 10, min_change: float = -5,
+                 max_change: float = 5):
+        super().__init__(random_generator=random_generator, n_changepoints=n_changepoints)
+        self.min_change = min_change
+        self.max_change = max_change
+
+    def generate(self, X, y, **params) -> Tuple:
+        """
+
+        :param X:
+        :param y:
+        :param params:
+        :return:
+        """
+        # Generate change points
+        self.changepoints = list(
+            zip(
+                self.random_generator.integers(int(0.05 * len(X)), int(0.95 * len(X)), size=self.n_changepoints),
+                self.random_generator.uniform(self.min_change, self.max_change, size=self.n_changepoints)
+            )
+        )
+
+        Xt = X.copy()
+
+        for idx, change in self.changepoints:
+            Xt[idx:] += change
+
+        return Xt, y

badgers/generators/time_series/noise.py

@@ -23,7 +23,43 @@ def generate(self, X, y, **params) -> Tuple:
         pass
 
 
-class GaussianNoiseGenerator(NoiseGenerator):
+class LocalGaussianNoiseGenerator(NoiseGenerator):
+
+    def __init__(self, random_generator=default_rng(seed=0), n_patterns: int = 10, patterns_width: int = 10,
+                 noise_std: float = 0.1):
+        super().__init__(random_generator=random_generator)
+        self.n_patterns = n_patterns
+        self.patterns_width = patterns_width
+        self.noise_std = noise_std
+
+    def generate(self, X, y, **params) -> Tuple:
+        # TODO input validation!
+        if X.ndim < 2:
+            raise ValueError(
+                "Expected 2D array. "
+                "Reshape your data either using array.reshape(-1, 1) if "
+                "your data has a single feature or array.reshape(1, -1) "
+                "if it contains a single sample."
+            )
+        # generate extreme values indices and values
+        self.patterns_indices_ = [(x, x + self.patterns_width) for x in
+                                  self.random_generator.choice(X.shape[0] - self.patterns_width,
+                                                               size=self.n_patterns,
+                                                               replace=False, p=None)]
+
+        scaler = StandardScaler()
+        # fit, transform
+        scaler.fit(X)
+        Xt = scaler.transform(X)
+
+        for (start, end) in self.patterns_indices_:
+            Xt[start:end, :] += self.random_generator.normal(loc=0, scale=self.noise_std, size=(self.patterns_width, Xt.shape[1]))
+
+        # inverse standardization
+        return scaler.inverse_transform(Xt), y
+
+
+class GlobalGaussianNoiseGenerator(NoiseGenerator):
     def __init__(self, random_generator=default_rng(seed=0), noise_std: float = 0.1):
         """
 
@@ -49,5 +85,5 @@ def generate(self, X, y, **params):
         Xt = scaler.transform(X)
         # add noise
         Xt = Xt + self.random_generator.normal(loc=0, scale=self.noise_std, size=Xt.shape)
-        # inverse pca
+        # inverse standardization
         return scaler.inverse_transform(Xt), y

badgers/generators/time_series/seasons.py

@@ -0,0 +1,46 @@
+import abc
+from typing import Tuple
+
+import numpy as np
+from numpy.random import default_rng
+
+from badgers.core.base import GeneratorMixin
+
+
+class SeasonsGenerator(GeneratorMixin):
+    """
+    Base class for transformers that generate seasons in time-series data
+    """
+
+    def __init__(self, random_generator=default_rng(seed=0)):
+        """
+        :param random_generator: a random number generator
+        """
+        self.random_generator = random_generator
+
+    @abc.abstractmethod
+    def generate(self, X, y, **params) -> Tuple:
+        pass
+
+
+class GlobalAdditiveSinusoidalSeasonGenerator(SeasonsGenerator):
+    """
+    Add a sinusoidal season to the input time-series data
+    """
+
+    def __init__(self, random_generator=default_rng(seed=0), period: int = 10):
+        super().__init__(random_generator=random_generator)
+        self.period = period
+
+    def generate(self, X, y, **params) -> Tuple:
+        """
+
+        :param X:
+        :param y:
+        :param params:
+        :return:
+        """
+        t = np.arange(len(X))
+        season = np.sin(t*2*np.pi/self.period)
+        Xt = X + season
+        return Xt, y

badgers/generators/time_series/trends.py

@@ -0,0 +1,45 @@
+import abc
+from typing import Tuple
+
+import numpy as np
+from numpy.random import default_rng
+
+from badgers.core.base import GeneratorMixin
+
+
+class TrendsGenerator(GeneratorMixin):
+    """
+    Base class for transformers that generate trends in time-series data
+    """
+
+    def __init__(self, random_generator=default_rng(seed=0)):
+        """
+        :param random_generator: a random number generator
+        """
+        self.random_generator = random_generator
+
+    @abc.abstractmethod
+    def generate(self, X, y, **params) -> Tuple:
+        pass
+
+
+class GlobalAdditiveLinearTrendGenerator(TrendsGenerator):
+    """
+    Add a linear trend to the input time-series data
+    """
+
+    def __init__(self, random_generator=default_rng(seed=0), slope: float = 0.1):
+        super().__init__(random_generator=random_generator)
+        self.slope = slope
+
+    def generate(self, X, y, **params) -> Tuple:
+        """
+
+        :param X:
+        :param y:
+        :param params:
+        :return:
+        """
+        trend = self.slope * np.linspace(0, 1, len(X))
+        Xt = X + trend
+        return Xt, y

docs/getting-started.md

@@ -10,10 +10,10 @@ Import badgers as any other library and start using it:
 
 ```python
 from sklearn.datasets import make_blobs
-from badgers.generators.tabular_data.noise import GaussianNoiseGenerator
+from badgers.generators.tabular_data.noise import GlobalGaussianNoiseGenerator
 
 X, y = make_blobs()
-trf = GaussianNoiseGenerator(noise_std=0.5)
+trf = GlobalGaussianNoiseGenerator(noise_std=0.5)
 Xt, yt = trf.generate(X, y)
 ```