mirror of
https://github.com/wassname/pytorch-ts.git
synced 2026-08-04 13:13:57 +08:00
artificial dataset
This commit is contained in:
Dr. Kashif Rasul
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user