diff --git a/pts/dataset/__init__.py b/pts/dataset/__init__.py index a16a9cf..a9737a4 100644 --- a/pts/dataset/__init__.py +++ b/pts/dataset/__init__.py @@ -5,4 +5,11 @@ from .loader import TrainDataLoader, InferenceDataLoader from .process import ProcessStartField, ProcessDataEntry from .utils import to_pandas from .stat import DatasetStatistics, ScaleHistogram, calculate_dataset_statistics -from .artificial import constant_dataset +from .artificial import ( + ArtificialDataset, + ConstantDataset, + ComplexSeasonalTimeSeries, + RecipeDataset, + constant_dataset, + default_synthetic, +) diff --git a/pts/dataset/artificial.py b/pts/dataset/artificial.py index a252997..6167ac4 100644 --- a/pts/dataset/artificial.py +++ b/pts/dataset/artificial.py @@ -1,14 +1,34 @@ +import math +import random from typing import Callable, List, NamedTuple, Optional, Tuple, Union +import numpy as np +import pandas as pd + from .common import ( MetaData, CategoricalFeatureInfo, BasicFeatureInfo, FieldName, Dataset, + TrainDatasets, + DataEntry, ) from .list_dataset import ListDataset from .stat import DatasetStatistics, calculate_dataset_statistics +from .recipe import ( + BinaryHolidays, + BinaryMarkovChain, + Constant, + ForEachCat, + Lag, + LinearTrend, + RandomCat, + RandomGaussian, + Stack, + generate, + take_as_list, +) class DatasetInfo(NamedTuple): @@ -25,6 +45,698 @@ class DatasetInfo(NamedTuple): test_statistics: DatasetStatistics +class ArtificialDataset: + """ + Parent class of a dataset that can be generated from code. + """ + + def __init__(self, freq) -> None: + self.freq = freq + + @property + def metadata(self) -> MetaData: + pass + + @property + def train(self) -> List[DataEntry]: + pass + + @property + def test(self) -> List[DataEntry]: + pass + + # todo return the same type as dataset repo for better usability + def generate(self) -> TrainDatasets: + return TrainDatasets( + metadata=self.metadata, + train=ListDataset(self.train, self.freq), + test=ListDataset(self.test, self.freq), + ) + + +class ConstantDataset(ArtificialDataset): + def __init__( + self, + num_timeseries: int = 10, + num_steps: int = 30, + freq: str = "1H", + start: str = "2000-01-01 00:00:00", + is_nan: bool = False, # Generates constant dataset of 0s with explicit NaN missing values + is_random_constant: bool = False, # Inserts random constant value for each time series + is_different_scales: bool = False, # Generates constants on various scales + is_piecewise: bool = False, # Determines whether the time series in the test + # and train set should have different constant values + is_noise: bool = False, # Determines whether to add Gaussian noise to the constant dataset + is_long: bool = False, # Determines whether some time series will have very long lengths + is_short: bool = False, # Determines whether some time series will have very short lengths + is_trend: bool = False, # Determines whether to add linear trends + num_missing_middle: int = 0, # Number of missing values in the middle of the time series + is_promotions: bool = False, # Determines whether to add promotions to the target time series + # and to store in metadata + holidays: Optional[ + List[pd.Timestamp] + ] = None, # Determines whether to add holidays to the target time series + # and to store in metadata + ) -> None: + super(ConstantDataset, self).__init__(freq) + self.num_timeseries = num_timeseries + self.num_steps = num_steps + self.num_training_steps = self.num_steps // 10 * 8 + self.prediction_length = self.num_steps - self.num_training_steps + self.start = start + self.is_nan = is_nan + self.is_random_constant = is_random_constant + self.is_different_scales = is_different_scales + self.is_piecewise = is_piecewise + self.is_noise = is_noise + self.is_long = is_long + self.is_short = is_short + self.is_trend = is_trend + self.num_missing_middle = num_missing_middle + self.is_promotions = is_promotions + self.holidays = holidays + + @property + def metadata(self) -> MetaData: + metadata = MetaData( + freq=self.freq, + feat_static_cat=[ + { + "name": "feat_static_cat_000", + "cardinality": str(self.num_timeseries), + } + ], + feat_static_real=[{"name": "feat_static_real_000"}], + prediction_length=self.prediction_length, + ) + if self.is_promotions or self.holidays: + metadata = MetaData( + freq=self.freq, + feat_static_cat=[ + { + "name": "feat_static_cat_000", + "cardinality": str(self.num_timeseries), + } + ], + feat_static_real=[{"name": "feat_static_real_000"}], + feat_dynamic_real=[BasicFeatureInfo(name=FieldName.FEAT_DYNAMIC_REAL)], + prediction_length=self.prediction_length, + ) + return metadata + + def determine_constant( + self, index: int, constant: Optional[float] = None, seed: int = 1 + ) -> Optional[float]: + if self.is_random_constant: + my_random = random.Random(seed) + constant = (index + 1) * my_random.random() + elif self.is_different_scales: + if index == 0: + constant = 1e-8 + elif constant is not None: + constant *= 100 + else: + constant = float(index) + return constant + + def compute_data_from_recipe( + self, + num_steps: int, + constant: Optional[float] = None, + one_to_zero: float = 0.1, + zero_to_one: float = 0.1, + scale_features: float = 200, + ) -> TrainDatasets: + recipe = [] + recipe_type = Constant(constant) + if self.is_noise: + recipe_type += RandomGaussian() # Use default stddev = 1.0 + if self.is_trend: + recipe_type += LinearTrend() + if self.is_promotions: + recipe.append( + ("binary_causal", BinaryMarkovChain(one_to_zero, zero_to_one)) + ) + recipe.append((FieldName.FEAT_DYNAMIC_REAL, Stack(["binary_causal"]))) + recipe_type += scale_features * Lag("binary_causal", lag=0) + if self.holidays: + timestamp = self.init_date() + # Compute dates array + dates = [] + for i in range(num_steps): + dates.append(timestamp) + timestamp += 1 + recipe.append(("binary_holidays", BinaryHolidays(dates, self.holidays))) + recipe.append((FieldName.FEAT_DYNAMIC_REAL, Stack(["binary_holidays"]))) + recipe_type += scale_features * Lag("binary_holidays", lag=0) + recipe.append((FieldName.TARGET, recipe_type)) + max_train_length = num_steps - self.prediction_length + data = RecipeDataset( + recipe=recipe, + metadata=self.metadata, + max_train_length=max_train_length, + prediction_length=self.prediction_length, + num_timeseries=1, # Add 1 time series at a time in the loop for different constant valus per time series + ) + generated = data.generate() + return generated + + def piecewise_constant(self, index: int, num_steps: int) -> List: + target = [] + for j in range(num_steps): + if j < self.num_training_steps: + constant = self.determine_constant(index=index) + else: + constant = self.determine_constant(index=index, seed=2) + target.append(constant) + return target + + def get_num_steps( + self, + index: int, + num_steps_max: int = 10000, + long_freq: int = 4, + num_steps_min: int = 2, + short_freq: int = 4, + ) -> int: + num_steps = self.num_steps + if self.is_long and index % long_freq == 0: + num_steps = num_steps_max + elif self.is_short and index % short_freq == 0: + num_steps = num_steps_min + return num_steps + + def init_date(self) -> pd.Timestamp: + week_dict = { + 0: "MON", + 1: "TUE", + 2: "WED", + 3: "THU", + 4: "FRI", + 5: "SAT", + 6: "SUN", + } + timestamp = pd.Timestamp(self.start) + freq_week_start = self.freq + if freq_week_start == "W": + freq_week_start = f"W-{week_dict[timestamp.weekday()]}" + return pd.Timestamp(self.start, freq=freq_week_start) + + @staticmethod + def insert_nans_and_zeros(ts_len: int) -> List: + target = [] + for j in range(ts_len): + # Place NaNs at even indices. Use convention no NaNs before start date. + if j != 0 and j % 2 == 0: + target.append(np.nan) + # Place zeros at odd indices + else: + target.append(0.0) + return target + + def insert_missing_vals_middle( + self, ts_len: int, constant: Optional[float] + ) -> List: + target = [] + lower_bound = (self.num_training_steps - self.num_missing_middle) // 2 + upper_bound = (self.num_training_steps + self.num_missing_middle) // 2 + num_missing_endpts = math.floor(0.1 * self.num_missing_middle) + for j in range(ts_len): + if ( + (0 < j < lower_bound and j % (2 * num_missing_endpts) == 0) + or (lower_bound <= j < upper_bound) + or (j >= upper_bound and j % (2 * num_missing_endpts) == 0) + ): + val = np.nan + else: + val = constant + target.append(val) + return target + + def generate_ts(self, num_ts_steps: int, is_train: bool = False) -> List[DataEntry]: + res = [] + constant = None + for i in range(self.num_timeseries): + if self.is_nan: + target = self.insert_nans_and_zeros(num_ts_steps) + elif self.is_piecewise: + target = self.piecewise_constant(i, num_ts_steps) + else: + constant = self.determine_constant(i, constant) + if self.num_missing_middle > 0: + target = self.insert_missing_vals_middle(num_ts_steps, constant) + elif ( + self.is_noise + or self.is_trend + or self.is_promotions + or self.holidays + ): + + num_steps = self.get_num_steps(i) + generated = self.compute_data_from_recipe(num_steps, constant) + if is_train: + time_series = generated.train + else: + assert generated.test is not None + time_series = generated.test + # returns np array convert to list for consistency + target = list(time_series)[0][FieldName.TARGET].tolist() + else: + target = [constant] * num_ts_steps + ts_data = dict( + start=self.start, + target=target, + item_id=str(i), + feat_static_cat=[i], + feat_static_real=[i], + ) + if self.is_promotions or self.holidays: + ts_data[FieldName.FEAT_DYNAMIC_REAL] = list(time_series)[0][ + FieldName.FEAT_DYNAMIC_REAL + ].tolist() + res.append(ts_data) + return res + + @property + def train(self) -> List[DataEntry]: + return self.generate_ts(num_ts_steps=self.num_training_steps, is_train=True) + + @property + def test(self) -> List[DataEntry]: + return self.generate_ts(num_ts_steps=self.num_steps) + + +class ComplexSeasonalTimeSeries(ArtificialDataset): + """ + Generate sinus time series that ramp up and reach a certain amplitude, and + level and have additional spikes on each sunday. + + + TODO: This could be converted to a RecipeDataset to avoid code duplication. + """ + + def __init__( + self, + num_series: int = 100, + prediction_length: int = 20, + freq_str: str = "D", + length_low: int = 30, + length_high: int = 200, + min_val: float = -10000, + max_val: float = 10000, + is_integer: bool = False, + proportion_missing_values: float = 0, + is_noise: bool = True, + is_scale: bool = True, + percentage_unique_timestamps: float = 0.07, + is_out_of_bounds_date: bool = False, + seasonality: Optional[int] = None, + clip_values: bool = False, + ) -> None: + """ + :param num_series: number of time series generated in the train and + test set + :param prediction_length: + :param freq_str: + :param length_low: minimum length of a time-series, must be larger than + prediction_length + :param length_high: maximum length of a time-series + :param min_val: min value of a time-series + :param max_val: max value of a time-series + :param is_integer: whether the dataset has integers or not + :param proportion_missing_values: + :param is_noise: whether to add noise + :param is_scale: whether to add scale + :param percentage_unique_timestamps: percentage of random start dates bounded between 0 and 1 + :param is_out_of_bounds_date: determines whether to use very old start dates and start dates far in the future + :param seasonality: Seasonality of the generated data. If not given uses default seasonality for frequency + :param clip_values: if True the values will be clipped to [min_val, max_val], otherwise linearly scales them + """ + assert length_low > prediction_length + super(ComplexSeasonalTimeSeries, self).__init__(freq_str) + self.num_series = num_series + self.prediction_length = prediction_length + self.length_low = length_low + self.length_high = length_high + self.freq_str = freq_str + self.min_val = min_val + self.max_val = max_val + self.is_integer = is_integer + self.proportion_missing_values = proportion_missing_values + self.is_noise = is_noise + self.is_scale = is_scale + self.percentage_unique_timestamps = percentage_unique_timestamps + self.is_out_of_bounds_date = is_out_of_bounds_date + self.seasonality = seasonality + self.clip_values = clip_values + + @property + def metadata(self) -> MetaData: + return MetaData(freq=self.freq, prediction_length=self.prediction_length) + + def _get_period(self) -> int: + if self.seasonality is not None: + return self.seasonality + if self.freq_str == "M": + return 24 + elif self.freq_str == "W": + return 52 + elif self.freq_str == "D": + return 14 + elif self.freq_str == "H": + return 24 + elif self.freq_str == "min": + return 60 + else: + raise RuntimeError() + + def _get_start(self, index: int, my_random: random.Random) -> str: + if ( + self.is_out_of_bounds_date and index == 0 + ): # Add edge case of dates out of normal bounds past date + start_y, start_m, start_d = ( + 1690, + 2, + 7, + ) # Pandas doesn't allot before 1650 + start_h, start_min = 18, 36 + elif ( + self.is_out_of_bounds_date and index == self.num_series - 1 + ): # Add edge case of dates out of normal bounds future date + start_y, start_m, start_d = ( + 2030, + 6, + 3, + ) # Pandas doesn't allot before 1650 + start_h, start_min = 18, 36 + # assume that only 100 * percentage_unique_timestamps of timestamps are unique + elif my_random.random() < self.percentage_unique_timestamps: + start_y = my_random.randint(2000, 2018) + start_m = my_random.randint(1, 12) + start_d = my_random.randint(1, 28) + start_h = my_random.randint(0, 23) + start_min = my_random.randint(0, 59) + else: + start_y, start_m, start_d = 2013, 11, 28 + start_h, start_min = 18, 36 + + if self.freq_str == "M": + return "%04.d-%02.d" % (start_y, start_m) + elif self.freq_str in ["W", "D"]: + return "%04.d-%02.d-%02.d" % (start_y, start_m, start_d) + elif self.freq_str == "H": + return "%04.d-%02.d-%02.d %02.d:00:00" % ( + start_y, + start_m, + start_d, + start_h, + ) + else: + return "%04.d-%02.d-%02.d %02.d:%02.d:00" % ( + start_y, + start_m, + start_d, + start_h, + start_min, + ) + + def _special_time_point_indicator(self, index) -> bool: + if self.freq_str == "M": + return index.month == 1 + elif self.freq_str == "W": + return index.month % 2 == 0 + elif self.freq_str == "D": + return index.dayofweek == 0 + elif self.freq_str == "H": + return index.hour == 0 + elif self.freq_str == "min": + return index.minute % 30 == 0 + else: + raise RuntimeError(f'Bad freq_str value "{index}"') + + @property + def train(self) -> List[DataEntry]: + return [ + dict( + start=ts[FieldName.START], + target=ts[FieldName.TARGET][: -self.prediction_length], + item_id=ts[FieldName.ITEM_ID], + ) + for ts in self.make_timeseries() + ] + + @property + def test(self) -> List[DataEntry]: + return self.make_timeseries() + + def make_timeseries(self, seed: int = 1) -> List[DataEntry]: + res = [] + # Fix seed so that the training set is the same + # as the test set from 0:self.prediction_length for the two independent calls + + def sigmoid(x: np.ndarray) -> np.ndarray: + return 1.0 / (1.0 + np.exp(-x)) + + # Ensure same start dates in test and training set + my_random = random.Random(seed) + state = np.random.RandomState(seed) + for i in range(self.num_series): + val_range = self.max_val - self.min_val + length = state.randint(low=self.length_low, high=self.length_high) + start = self._get_start(i, my_random) + envelope = sigmoid((np.arange(length) - 20.0) / 10.0) + level = 0.3 * val_range * (state.random_sample() - 0.5) + phi = 2 * np.pi * state.random_sample() + period = self._get_period() + w = 2 * np.pi / period + t = np.arange(length) + idx = pd.date_range(start=start, freq=self.freq_str, periods=length) + special_tp_indicator = self._special_time_point_indicator(idx) + sunday_effect = state.random_sample() * special_tp_indicator + v = np.sin(w * t + phi) + sunday_effect + + if self.is_scale: + scale = 0.1 * val_range * state.random_sample() + v *= scale + v += level + if self.is_noise: + noise_range = 0.02 * val_range * state.random_sample() + noise = noise_range * state.normal(size=length) + v += noise + v = envelope * v + if self.clip_values: + np.clip(v, a_min=self.min_val, a_max=self.max_val, out=v) + else: + """ + Rather than mapping [v_min, v_max] to [self.min_val, self.max_val] which would lead to + all the time series having the same min and max, we want to keep the same interval length + (v_max - v_min). We thus shift the interval [v_min, v_max] in [self.min_val, self.max_val] + and clip it if needed. + """ + v_min, v_max = v.min(), v.max() + p_min, p_max = ( + max(self.min_val, v_min), + min(self.max_val, v_max), + ) + shifted_min = np.clip( + p_min + (p_max - v_max), a_min=self.min_val, a_max=self.max_val, + ) + shifted_max = np.clip( + p_max + (p_min - v_min), a_min=self.min_val, a_max=self.max_val, + ) + v = shifted_min + (shifted_max - shifted_min) * (v - v_min) / ( + v_max - v_min + ) + + if self.is_integer: + np.clip( + v, a_min=np.ceil(self.min_val), a_max=np.floor(self.max_val), out=v, + ) + v = np.round(v).astype(int) + v = list(v.tolist()) + if self.proportion_missing_values > 0: + assert ( + self.proportion_missing_values < 1.0 + ), "Please chose a number 0 < x < 1.0" + idx = np.arange(len(v)) + state.shuffle(idx) + num_missing_values = ( + int(len(v) * self.proportion_missing_values) + 1 + ) # Add one in case this gets zero + missing_idx = idx[:num_missing_values] + for j in missing_idx: + # Using convention that there are no missing values before the start date. + if j != 0: + v[j] = None if state.rand() < 0.5 else "NaN" + res.append( + dict( + start=pd.Timestamp(start, freq=self.freq_str), + target=np.array(v), + item_id=i, + ) + ) + return res + + +class RecipeDataset(ArtificialDataset): + """Synthetic data set generated by providing a recipe. + + A recipe is either a (non-deterministic) function + + f(length: int, global_state: dict) -> dict + + or list of (field, function) tuples of the form + + (field: str, f(data: dict, length: int, global_state: dict) -> dict) + + which is processed sequentially, with data initially set to {}, + and each entry updating data[field] to the output of the function + call. + """ + + def __init__( + self, + recipe: Union[Callable, List[Tuple[str, Callable]]], + metadata: MetaData, + max_train_length: int, + prediction_length: int, + num_timeseries: int, + trim_length_fun=lambda x, **kwargs: 0, + data_start=pd.Timestamp("2014-01-01"), + ) -> None: + """ + + :param recipe: The recipe to generate from (see class docstring) + :param metadata: The metadata to be included in the dataset + :param max_train_length: The maximum length of a training time series. + :param prediction_length: The length of the prediction range + :param num_timeseries: Number of time series to generate + :param trim_length_fun: Callable f(x: int) -> int returning the + (shortened) training length + :param data_start: Start date for the data set + """ + super().__init__(freq=metadata.freq) + + self.recipe = recipe + self._metadata = metadata + self.max_train_length = max_train_length + self.prediction_length = prediction_length + self.trim_length_fun = trim_length_fun + self.num_timeseries = num_timeseries + self.data_start = pd.Timestamp(data_start, freq=self._metadata.freq) + + @property + def metadata(self) -> MetaData: + return self._metadata + + def dataset_info(self, train_ds: Dataset, test_ds: Dataset) -> DatasetInfo: + return DatasetInfo( + name=f"RecipeDataset({repr(self.recipe)})", + metadata=self.metadata, + prediction_length=self.prediction_length, + train_statistics=calculate_dataset_statistics(train_ds), + test_statistics=calculate_dataset_statistics(test_ds), + ) + + @staticmethod + def trim_ts_item_end(x: DataEntry, length: int) -> DataEntry: + """Trim a TimeSeriesItem into a training range, by removing + the last prediction_length time points from the target and dynamic + features.""" + y = dict( + item_id=x[FieldName.ITEM_ID], + start=x[FieldName.START], + target=x[FieldName.TARGET][:-length], + ) + + if FieldName.FEAT_DYNAMIC_CAT in x: + y[FieldName.FEAT_DYNAMIC_CAT] = x[FieldName.FEAT_DYNAMIC_CAT][:, :-length] + if FieldName.FEAT_DYNAMIC_REAL in x: + y[FieldName.FEAT_DYNAMIC_REAL] = x[FieldName.FEAT_DYNAMIC_REAL][:, :-length] + return y + + @staticmethod + def trim_ts_item_front(x: DataEntry, length: int) -> DataEntry: + """Trim a TimeSeriesItem into a training range, by removing + the first offset_front time points from the target and dynamic + features.""" + assert length <= len(x[FieldName.TARGET]) + + y = dict( + item_id=x[FieldName.ITEM_ID], + start=x[FieldName.START] + length * x[FieldName.START].freq, + target=x[FieldName.TARGET][length:], + ) + + if FieldName.FEAT_DYNAMIC_CAT in x: + y[FieldName.FEAT_DYNAMIC_CAT] = x[FieldName.FEAT_DYNAMIC_CAT][:, length:] + if FieldName.FEAT_DYNAMIC_REAL in x: + y[FieldName.FEAT_DYNAMIC_REAL] = x[FieldName.FEAT_DYNAMIC_REAL][:, length:] + return y + + def generate(self) -> TrainDatasets: + metadata = self.metadata + data_it = generate( + length=self.max_train_length + self.prediction_length, + recipe=self.recipe, + start=self.data_start, + ) + full_length_data = take_as_list(data_it, self.num_timeseries) + + test_data = [ + RecipeDataset.trim_ts_item_front( + x, self.trim_length_fun(x, train_length=self.max_train_length) + ) + for x in full_length_data + ] + train_data = [ + RecipeDataset.trim_ts_item_end(x, self.prediction_length) for x in test_data + ] + return TrainDatasets( + metadata=metadata, + train=ListDataset(train_data, metadata.freq), + test=ListDataset(test_data, metadata.freq), + ) + + +def default_synthetic() -> Tuple[DatasetInfo, Dataset, Dataset]: + + recipe = [ + (FieldName.TARGET, LinearTrend() + RandomGaussian()), + (FieldName.FEAT_STATIC_CAT, RandomCat([10])), + ( + FieldName.FEAT_STATIC_REAL, + ForEachCat(RandomGaussian(1, (10,)), FieldName.FEAT_STATIC_CAT) + + RandomGaussian(0.1, (10,)), + ), + ] + + data = RecipeDataset( + recipe=recipe, + metadata=MetaData( + freq="D", + feat_static_real=[BasicFeatureInfo(name=FieldName.FEAT_STATIC_REAL)], + feat_static_cat=[ + CategoricalFeatureInfo(name=FieldName.FEAT_STATIC_CAT, cardinality=10) + ], + feat_dynamic_real=[BasicFeatureInfo(name=FieldName.FEAT_DYNAMIC_REAL)], + ), + max_train_length=20, + prediction_length=10, + num_timeseries=10, + trim_length_fun=lambda x, **kwargs: np.minimum( + int(np.random.geometric(1 / (kwargs["train_length"] / 2))), + kwargs["train_length"], + ), + ) + + generated = data.generate() + assert generated.test is not None + info = data.dataset_info(generated.train, generated.test) + + return info, generated.train, generated.test + + def constant_dataset() -> Tuple[DatasetInfo, Dataset, Dataset]: metadata = MetaData( freq="1H", diff --git a/pts/dataset/recipe.py b/pts/dataset/recipe.py new file mode 100644 index 0000000..943c3de --- /dev/null +++ b/pts/dataset/recipe.py @@ -0,0 +1,604 @@ +# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). +# You may not use this file except in compliance with the License. +# A copy of the License is located at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# or in the "license" file accompanying this file. This file is distributed +# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +# express or implied. See the License for the specific language governing +# permissions and limitations under the License. + +# Standard library imports +import functools +import itertools +import operator +from typing import ( + Any, + Callable, + Dict, + Iterator, + List, + Optional, + Sequence, + Tuple, + Union, +) + +# Third-party imports +import numpy as np +import pandas as pd + +# First-party imports +from .common import DataEntry + +ValueOrCallable = Union[Any, Callable] +Recipe = List[Tuple[str, Callable]] +Env = Dict[str, Any] + + +def resolve(val_or_callable: ValueOrCallable, context: Env, *args, **kwargs): + if callable(val_or_callable): + return val_or_callable(context, *args, **kwargs) + elif isinstance(val_or_callable, str): + return context[val_or_callable] + else: + return val_or_callable + + +def generate( + length: int, + recipe: Union[Callable, Recipe], + start: pd.Timestamp, + global_state: Optional[dict] = None, + seed: int = 0, + item_id_prefix: str = "", +) -> Iterator[DataEntry]: + np.random.seed(seed) + + if global_state is None: + global_state = {} + + if isinstance(recipe, list): + for x in itertools.count(): + data: DataEntry = {} + for k, f in recipe: + data[k] = resolve( + f, data, length=length, field_name=k, global_state=global_state, + ) + yield dict(**data, item_id=item_id_prefix + str(x), start=start) + else: + assert callable(recipe) + for x in itertools.count(): + data = recipe(length=length, global_state=global_state) + yield dict(**data, item_id=item_id_prefix + str(x), start=start) + + +def evaluate( + funcs: Recipe, length: int, *args, global_state: dict = None, **kwargs +) -> Env: + if global_state is None: + global_state = {} + + if "length" in kwargs: + del kwargs["length"] + if "field_name" in kwargs: + del kwargs["field_name"] + if "global_state" in kwargs: + del kwargs["global_state"] + + data: DataEntry = {} + for k, f in funcs: + try: + data[k] = resolve( + f, + data, + length=length, + field_name=k, + global_state=global_state, + *args, + **kwargs + ) + except ValueError as e: + raise ValueError('Error while evaluating key "{}"'.format(k), e) + + return data + + +def make_func( + length: int, funcs: Recipe, global_state=None +) -> Callable[[int, Env], DataEntry]: + if global_state is None: + global_state = {} + + def f(length=length, global_state=global_state, *args, **kwargs): + data = {} + for k, f in funcs: + data[k] = resolve( + f, + data, + length=length, + field_name=k, + global_state=global_state, + *args, + **kwargs + ) + return data + + return f + + +def take_as_list(iterator, num): + return list(itertools.islice(iterator, num)) + + +class Debug: + def __init__(self, print_global=False) -> None: + self.print_global = print_global + + def __call__(self, x: Env, global_state, **kwargs): + print(x) + if self.print_global: + print(global_state) + return 0 + + +class Lifted: + def __add__(self, other): + return LiftedAdd(self, other) + + def __radd__(self, other): + return LiftedAdd(other, self) + + def __sub__(self, other): + return LiftedSub(self, other) + + def __rsub__(self, other): + return LiftedSub(other, self) + + def __mul__(self, other): + return LiftedMul(self, other, operator.mul) + + def __rmul__(self, other): + return LiftedMul(other, self, operator.mul) + + def __truediv__(self, other): + return LiftedTruediv(self, other, operator.truediv) + + def __rtruediv__(self, other): + return LiftedTruediv(other, self, operator.truediv) + + def __call__( + self, x: Env, length: int, field_name: str, global_state: Dict, *args, **kwargs + ): + pass + + +class LiftedBinaryOp(Lifted): + def __init__(self, left, right, op) -> None: + self.left = left + self.right = right + self.op = op + + def __call__(self, *args, **kwargs): + left = resolve(self.left, *args, **kwargs) + right = resolve(self.right, *args, **kwargs) + return self.op(left, right) + + +class LiftedAdd(LiftedBinaryOp): + def __init__(self, left, right) -> None: + super().__init__(left, right, operator.add) + + +class LiftedSub(LiftedBinaryOp): + def __init__(self, left, right) -> None: + super().__init__(left, right, operator.sub) + + +class LiftedMul(LiftedBinaryOp): + def __init__(self, left, right) -> None: + super().__init__(left, right, operator.mul) + + +class LiftedTruediv(LiftedBinaryOp): + def __init__(self, left, right) -> None: + super().__init__(left, right, operator.truediv) + + +class RandomGaussian(Lifted): + def __init__( + self, stddev: ValueOrCallable = 1.0, shape: Sequence[int] = (0,) + ) -> None: + self.stddev = stddev + self.shape = shape + + def __call__(self, x: Env, length: int, *args, **kwargs): + stddev = resolve(self.stddev, x, length, *args, **kwargs) + s = np.array(self.shape) + s[s == 0] = length + return stddev * np.random.randn(*s) + + +# Binary recipe that returns 1 if date is in holidays list and 0 otherwise +class BinaryHolidays(Lifted): + # TODO: holidays is type List[datetime.date] + def __init__(self, dates: List[pd.Timestamp], holidays: List[Any]) -> None: + self.dates = dates + self.holidays = holidays + + def __call__(self, *args, **kwargs): + length = len(self.dates) + out = np.ones(length) + for i, date in enumerate(self.dates): + # Convert to string to check if inside of holidays datatime.date + if date.date() in self.holidays: + out[i] = 1.0 + else: + out[i] = 0.0 + return out + + +class RandomBinary(Lifted): + def __init__(self, prob: ValueOrCallable = 0.1) -> None: + self.prob = prob + + def __call__(self, x: Env, length: int, *args, **kwargs): + prob = resolve(self.prob, x, length, *args, **kwargs) + return 1.0 * (np.random.rand(length) < prob) + + +class RandomSymmetricDirichlet(Lifted): + def __init__( + self, alpha: ValueOrCallable = 1.0, shape: Sequence[int] = (0,) + ) -> None: + self.alpha = alpha + self.shape = shape + + def __call__(self, x, length, *args, **kwargs): + alpha = resolve(self.alpha, x, length, *args, **kwargs) + s = np.array(self.shape) + s[s == 0] = length + return np.random.dirichlet(alpha * np.ones(s)) + + +class BinaryMarkovChain(Lifted): + def __init__( + self, one_to_zero: ValueOrCallable, zero_to_one: ValueOrCallable + ) -> None: + self.one_to_zero = one_to_zero + self.zero_to_one = zero_to_one + + def __call__(self, x: Env, length: int, *args, **kwargs): + probs = np.zeros(2) + probs[0] = resolve(self.zero_to_one, x, length, *args, **kwargs) + probs[1] = resolve(self.one_to_zero, x, length, *args, **kwargs) + out = np.ones(length, dtype=np.int) # initial state is 1 + uu = np.random.rand(length) + for i in range(1, length): + if uu[i] < probs[out[i - 1]]: + out[i] = 1 - out[i - 1] + else: + out[i] = out[i - 1] + return out + + +class Constant(Lifted): + def __init__(self, constant) -> None: + self.constant = constant + + def __call__(self, *args, **kwargs): + return self.constant + + +class ConstantVec(Lifted): + def __init__(self, constant: ValueOrCallable) -> None: + self.constant = constant + + def __call__(self, x: Env, length: int, *args, **kwargs): + constant = resolve(self.constant, x, length, *args, **kwargs) + return constant * np.ones(length) + + +class NormalizeMax(Lifted): + def __init__(self, input) -> None: + self.input = input + + def __call__(self, x: Env, *args, **kwargs): + inp = resolve(self.input, x, *args, kwargs) + return inp / np.max(inp) + + +class OnesLike(Lifted): + def __init__(self, other) -> None: + self.other = other + + def __call__(self, x, length, *args, **kwargs): + other = resolve(self.other, x, length, **kwargs) + return np.ones_like(other) + + +class LinearTrend(Lifted): + def __init__(self, slope: ValueOrCallable = 1.0) -> None: + self.slope = slope + + def __call__(self, x, length, *args, **kwargs): + slope = resolve(self.slope, x, length, *args, **kwargs) + return slope * np.arange(length) / length + + +class RandomCat: + def __init__( + self, + cardinalities: List[int], + prob_fun: Callable = RandomSymmetricDirichlet(alpha=1.0, shape=(0,)), + ) -> None: + self.cardinalities = cardinalities + self.prob_fun = prob_fun + + def __call__(self, x, field_name, global_state, **kwargs): + if field_name not in global_state: + probs = [self.prob_fun(x, length=c) for c in self.cardinalities] + global_state[field_name] = probs + probs = global_state[field_name] + cats = np.array( + [ + np.random.choice(np.arange(len(probs[i])), p=probs[i]) + for i in range(len(probs)) + ] + ) + return cats + + +class Lag(Lifted): + def __init__( + self, input: ValueOrCallable, lag: ValueOrCallable = 0, pad_const: int = 0, + ) -> None: + self.input = input + self.lag = lag + self.pad_const = pad_const + + def __call__(self, x, *args, **kwargs): + feat = resolve(self.input, x, *args, **kwargs) + lag = resolve(self.lag, x, *args, **kwargs) + + if lag > 0: + lagged_feat = np.concatenate((self.pad_const * np.ones(lag), feat[:-lag])) + elif lag < 0: + lagged_feat = np.concatenate((feat[-lag:], self.pad_const * np.ones(-lag))) + + else: + lagged_feat = feat + return lagged_feat + + +class ForEachCat(Lifted): + def __init__(self, fun, cat_field="cat", cat_idx=0) -> None: + self.fun = fun + self.cat_field = cat_field + self.cat_idx = cat_idx + + def __call__( + self, x: Env, length: int, field_name: str, global_state: Dict, *args, **kwargs + ): + c = x[self.cat_field][self.cat_idx] + if field_name not in global_state: + global_state[field_name] = np.empty( + len(global_state[self.cat_field][self.cat_idx]), dtype=np.object, + ) + if global_state[field_name][c] is None: + global_state[field_name][c] = self.fun( + x, length=length, field_name=field_name, *args, **kwargs + ) + return global_state[field_name][c] + + +class Eval(Lifted): + def __init__(self, expr: str) -> None: + self.expr = expr + + def __call__(self, x: Env, length: int, *args, **kwargs): + return eval(self.expr, globals(), dict(x=x, length=length, **kwargs)) + + +class SmoothSeasonality(Lifted): + def __init__(self, period: ValueOrCallable, phase: ValueOrCallable) -> None: + self.period = period + self.phase = phase + + def __call__(self, x: Env, length: int, *args, **kwargs): + period = resolve(self.period, x, length, *args, **kwargs) + phase = resolve(self.phase, x, length, *args, **kwargs) + return (np.sin(2.0 / period * np.pi * (np.arange(length) + phase)) + 1) / 2.0 + + +class Add(Lifted): + def __init__(self, inputs: List[ValueOrCallable]) -> None: + self.inputs = inputs + + def __call__(self, x: Env, length: int, *args, **kwargs): + return sum([resolve(k, x, length, *args, **kwargs) for k in self.inputs]) + + +class Mul(Lifted): + def __init__(self, inputs) -> None: + self.inputs = inputs + + def __call__(self, x: Env, length: int, *args, **kwargs): + return functools.reduce( + operator.mul, [resolve(k, x, length, *args, **kwargs) for k in self.inputs], + ) + + +class NanWhere(Lifted): + def __init__(self, source: ValueOrCallable, nan_indicator: ValueOrCallable) -> None: + self.source = source + self.nan_indicator = nan_indicator + + def __call__(self, x: Env, length: int, *args, **kwargs): + source = resolve(self.source, x, length, *args, **kwargs) + nan_indicator = resolve(self.nan_indicator, x, length, *args, **kwargs) + out = source.copy() + out[nan_indicator == 1] = np.nan + return out + + +class OneMinus(Lifted): + def __init__(self, source: ValueOrCallable) -> None: + self.source = source + + def __call__(self, x: Env, length: int, *args, **kwargs): + value = resolve(self.source, x, length, *args, **kwargs) + return 1 - value + + +class Concatenate(Lifted): + def __init__(self, inputs: List[ValueOrCallable], axis: int = 0) -> None: + self.inputs = inputs + self.axis = axis + + def __call__(self, x: Env, length: int, *args, **kwargs): + inputs = [resolve(z, x, length, **kwargs) for z in self.inputs] + return np.concatenate(inputs, self.axis) + + +class Stack(Lifted): + def __init__(self, inputs: List[ValueOrCallable]) -> None: + self.inputs = inputs + + def __call__(self, x: Env, length: int, *args, **kwargs): + inputs = [resolve(z, x, length, **kwargs) for z in self.inputs] + return np.stack(inputs, axis=0) + + +class StackPrefix(Lifted): + def __init__(self, prefix: str) -> None: + self.prefix = prefix + + def __call__(self, x: Env, length: int, *args, **kwargs): + inputs = [v for k, v in x.items() if k.startswith(self.prefix)] + return np.stack(inputs, axis=0) + + +class Ref(Lifted): + def __init__(self, field_name: str) -> None: + self.field_name = field_name + + def __call__(self, x: Env, length: int, *args, **kwargs): + return x[self.field_name] + + +class RandomUniform(Lifted): + def __init__( + self, low: ValueOrCallable = 0.0, high: ValueOrCallable = 1.0, shape=(0,), + ) -> None: + self.low = low + self.high = high + self.shape = shape + + def __call__(self, x: Env, length: int, *args, **kwargs): + low = resolve(self.low, x, length, *args, **kwargs) + high = resolve(self.high, x, length, *args, **kwargs) + s = np.array(self.shape) + s[s == 0] = length + return np.random.uniform(low, high, s) + + +class RandomInteger(Lifted): + def __init__( + self, + low: ValueOrCallable, + high: ValueOrCallable, + shape: Optional[Sequence[int]] = (0,), + ) -> None: + self.low = low + self.high = high + self.shape = shape + + def __call__(self, x: Env, length: int, *args, **kwargs): + low = resolve(self.low, x, length, *args, **kwargs) + high = resolve(self.high, x, length, *args, **kwargs) + if self.shape is not None: + s = np.array(self.shape) + s[s == 0] = length + return np.random.randint(low, high, s) + else: + return np.random.randint(low, high) + + +class RandomChangepoints(Lifted): + def __init__(self, max_num_changepoints: ValueOrCallable) -> None: + self.max_num_changepoints = max_num_changepoints + + def __call__(self, x: Env, length: int, *args, **kwargs): + max_num_changepoints = resolve( + self.max_num_changepoints, x, length, *args, **kwargs + ) + num_changepoints = np.random.randint(0, max_num_changepoints + 1) + change_idx = np.sort( + np.random.randint(low=1, high=length - 1, size=(num_changepoints,)) + ) + change_ranges = np.concatenate([change_idx, [length]]) + out = np.zeros(length, dtype=np.int) + for i in range(0, num_changepoints): + out[change_ranges[i] : change_ranges[i + 1]] = i + 1 + return out + + +class Repeated(Lifted): + def __init__(self, pattern: ValueOrCallable) -> None: + self.pattern = pattern + + def __call__(self, x: Env, length: int, *args, **kwargs): + pattern = resolve(self.pattern, x, length, **kwargs) + repeats = length // len(pattern) + 1 + out = np.tile(pattern, (repeats,)) + return out[:length] + + +class Convolve(Lifted): + def __init__(self, input: ValueOrCallable, filter: ValueOrCallable) -> None: + self.filter = filter + self.input = input + + def __call__(self, x: Env, length: int, *args, **kwargs): + fil = resolve(self.filter, x, length, **kwargs) + inp = resolve(self.input, x, length, **kwargs) + out = np.convolve(inp, fil, mode="same") + return out + + +class Dilated(Lifted): + def __init__(self, source: Callable, dilation: int) -> None: + self.source = source + self.dilation = dilation + + def __call__(self, x: Env, length: int, *args, **kwargs): + inner = self.source(x, length // self.dilation + 1, **kwargs) + out = np.repeat(inner, self.dilation) + return out[:length] + + +class Choose(Lifted): + def __init__(self, options: ValueOrCallable, selector: ValueOrCallable) -> None: + self.options = options + self.selector = selector + + def __call__(self, x, length, **kwargs): + options = resolve(self.options, x, length, **kwargs) + selector = resolve(self.selector, x, length, **kwargs) + e = np.eye(options.shape[0]) + out = np.sum(e[selector] * options.T, axis=1) + return out + + +class EvalRecipe(Lifted): + def __init__(self, recipe: Recipe, op: ValueOrCallable) -> None: + self.recipe = recipe + self.op = op + + def __call__(self, x: Env, *args, **kwargs): + xx = evaluate(self.recipe, *args, **kwargs) + return resolve(self.op, xx, *args, **kwargs)