diff --git a/pts/dataset/__init__.py b/pts/dataset/__init__.py index 228b905..3ba88c4 100644 --- a/pts/dataset/__init__.py +++ b/pts/dataset/__init__.py @@ -1,7 +1,11 @@ from pts.dataset.common import DataEntry, FieldName from pts.dataset.list_dataset import ListDataset from pts.dataset.loader import DataLoader, InferenceDataLoader, TrainDataLoader -from pts.dataset.sampler import (BucketInstanceSampler, - ExpectedNumInstanceSampler, InstanceSampler, - TestSplitSampler, UniformSplitSampler) +from pts.dataset.sampler import ( + BucketInstanceSampler, + ExpectedNumInstanceSampler, + InstanceSampler, + TestSplitSampler, + UniformSplitSampler, +) from pts.dataset.utils import to_pandas diff --git a/pts/dataset/list_dataset.py b/pts/dataset/list_dataset.py index 28b5386..a4230d0 100644 --- a/pts/dataset/list_dataset.py +++ b/pts/dataset/list_dataset.py @@ -5,17 +5,16 @@ from .process import ProcessDataEntry class ListDataset(Dataset): - def __init__(self, - data_iter: Iterable[DataEntry], - freq: str, - one_dim_target: bool = True) -> None: + def __init__( + self, data_iter: Iterable[DataEntry], freq: str, one_dim_target: bool = True + ) -> None: process = ProcessDataEntry(freq, one_dim_target) self.list_data = [process(data) for data in data_iter] def __iter__(self): source_name = "list_data" for row_number, data in enumerate(self.list_data, start=1): - data['source'] = SourceContext(source=source_name, row=row_number) + data["source"] = SourceContext(source=source_name, row=row_number) yield data def __len__(self): diff --git a/pts/dataset/loader.py b/pts/dataset/loader.py index f1b02b6..0081642 100644 --- a/pts/dataset/loader.py +++ b/pts/dataset/loader.py @@ -3,6 +3,7 @@ from collections import defaultdict from typing import Any, Dict, Iterable, Iterator, List, Optional # noqa: F401 import numpy as np + # Third-party imports import torch diff --git a/pts/dataset/process.py b/pts/dataset/process.py index 3d5d865..0e84c2a 100644 --- a/pts/dataset/process.py +++ b/pts/dataset/process.py @@ -17,8 +17,7 @@ class ProcessStartField: try: value = ProcessStartField.process(data[self.name], self.freq) except (TypeError, ValueError) as e: - raise Exception( - f'Error "{e}" occurred when reading field "{self.name}"') + raise Exception(f'Error "{e}" occurred when reading field "{self.name}"') data[self.name] = value @@ -31,23 +30,19 @@ class ProcessStartField: # operate on time information (days, hours, minute, second) if isinstance(timestamp.freq, Tick): - return pd.Timestamp(timestamp.floor(timestamp.freq), - timestamp.freq) + return pd.Timestamp(timestamp.floor(timestamp.freq), timestamp.freq) # since we are only interested in the data piece, we normalize the # time information - timestamp = timestamp.replace(hour=0, - minute=0, - second=0, - microsecond=0, - nanosecond=0) + timestamp = timestamp.replace( + hour=0, minute=0, second=0, microsecond=0, nanosecond=0 + ) return timestamp.freq.rollforward(timestamp) class ProcessTimeSeriesField: - def __init__(self, name, is_required: bool, is_static: bool, - is_cat: bool) -> None: + def __init__(self, name, is_required: bool, is_static: bool, is_cat: bool) -> None: self.name = name self.is_required = is_required self.req_ndim = 1 if is_static else 2 @@ -71,8 +66,7 @@ class ProcessTimeSeriesField: elif not self.is_required: return data else: - raise Exception( - f"JSON object is missing a required field `{self.name}`") + raise Exception(f"JSON object is missing a required field `{self.name}`") class ProcessDataEntry: @@ -81,28 +75,24 @@ class ProcessDataEntry: List[Callable[[DataEntry], DataEntry]], [ ProcessStartField("start", freq=freq), - ProcessTimeSeriesField("target", - is_required=True, - is_cat=False, - is_static=one_dim_target), - ProcessTimeSeriesField("feat_dynamic_cat", - is_required=False, - is_cat=True, - is_static=False), + ProcessTimeSeriesField( + "target", is_required=True, is_cat=False, is_static=one_dim_target + ), + ProcessTimeSeriesField( + "feat_dynamic_cat", is_required=False, is_cat=True, is_static=False + ), ProcessTimeSeriesField( "feat_dynamic_real", is_required=False, is_cat=False, is_static=False, ), - ProcessTimeSeriesField("feat_static_cat", - is_required=False, - is_cat=True, - is_static=True), - ProcessTimeSeriesField("feat_static_real", - is_required=False, - is_cat=False, - is_static=True), + ProcessTimeSeriesField( + "feat_static_cat", is_required=False, is_cat=True, is_static=True + ), + ProcessTimeSeriesField( + "feat_static_real", is_required=False, is_cat=False, is_static=True + ), ], ) diff --git a/pts/dataset/sampler.py b/pts/dataset/sampler.py index 0f84c9c..18460a8 100644 --- a/pts/dataset/sampler.py +++ b/pts/dataset/sampler.py @@ -8,8 +8,7 @@ from .stat import ScaleHistogram class InstanceSampler(ABC): @abstractmethod - def __call__(self, ts: np.ndarray, a: int, - b: int) -> Union[np.ndarray, List[int]]: + def __call__(self, ts: np.ndarray, a: int, b: int) -> Union[np.ndarray, List[int]]: pass @@ -21,16 +20,17 @@ class UniformSplitSampler(InstanceSampler): p Probability of selecting a time point """ + def __init__(self, p: float = 1.0 / 20.0) -> None: self.p = p - self.lookup = np.arange(2**13) + self.lookup = np.arange(2 ** 13) def __call__(self, ts: np.ndarray, a: int, b: int) -> np.ndarray: assert a <= b while ts.shape[-1] >= len(self.lookup): self.lookup = np.arange(2 * len(self.lookup)) mask = np.random.uniform(low=0.0, high=1.0, size=b - a + 1) < self.p - return self.lookup[a:a + len(mask)][mask] + return self.lookup[a : a + len(mask)][mask] class TestSplitSampler(InstanceSampler): @@ -38,6 +38,7 @@ class TestSplitSampler(InstanceSampler): Sampler used for prediction. Always selects the last time point for splitting i.e. the forecast point for the time series. """ + def __call__(self, ts: np.ndarray, a: int, b: int) -> np.ndarray: return np.array([b]) @@ -52,11 +53,12 @@ class ExpectedNumInstanceSampler(InstanceSampler): num_instances number of training examples generated per time series on average """ + def __init__(self, num_instances: float) -> None: self.num_instances = num_instances self.avg_length = 0.0 self.n = 0.0 - self.lookup = np.arange(2**13) + self.lookup = np.arange(2 ** 13) def __call__(self, ts: np.ndarray, a: int, b: int) -> np.ndarray: while ts.shape[-1] >= len(self.lookup): @@ -67,7 +69,7 @@ class ExpectedNumInstanceSampler(InstanceSampler): p = self.num_instances / self.avg_length mask = np.random.uniform(low=0.0, high=1.0, size=b - a + 1) < p - indices = self.lookup[a:a + len(mask)][mask] + indices = self.lookup[a : a + len(mask)][mask] return indices @@ -83,16 +85,17 @@ class BucketInstanceSampler(InstanceSampler): The histogram of scale for the time series. Here scale is the mean abs value of the time series. """ + def __init__(self, scale_histogram: ScaleHistogram) -> None: # probability of sampling a bucket i is the inverse of its number of # elements self.scale_histogram = scale_histogram - self.lookup = np.arange(2**13) + self.lookup = np.arange(2 ** 13) def __call__(self, ts: np.ndarray, a: int, b: int) -> np.ndarray: while ts.shape[-1] >= len(self.lookup): self.lookup = np.arange(2 * len(self.lookup)) p = 1.0 / self.scale_histogram.count(ts) mask = np.random.uniform(low=0.0, high=1.0, size=b - a + 1) < p - indices = self.lookup[a:a + len(mask)][mask] + indices = self.lookup[a : a + len(mask)][mask] return indices diff --git a/pts/dataset/transformed_dataset.py b/pts/dataset/transformed_dataset.py index be064ed..cf0538b 100644 --- a/pts/dataset/transformed_dataset.py +++ b/pts/dataset/transformed_dataset.py @@ -18,8 +18,10 @@ class TransformedDataset(Dataset): transformations List of transformations to apply """ - def __init__(self, base_dataset: Dataset, - transformations: List[Transformation]) -> None: + + def __init__( + self, base_dataset: Dataset, transformations: List[Transformation] + ) -> None: self.base_dataset = base_dataset self.transformations = Chain(transformations) diff --git a/pts/exception.py b/pts/exception.py index e3b212b..0116429 100644 --- a/pts/exception.py +++ b/pts/exception.py @@ -1,4 +1,3 @@ - def assert_pts(condition: bool, message: str, *args, **kwargs) -> None: if not condition: raise Exception(message.format(*args, **kwargs)) diff --git a/pts/feature/__init__.py b/pts/feature/__init__.py index ad4b80d..e4bfaee 100644 --- a/pts/feature/__init__.py +++ b/pts/feature/__init__.py @@ -1,16 +1,38 @@ from pts.feature.lag import get_lags_for_frequency -from pts.feature.time_feature import (DayOfMonth, DayOfWeek, DayOfYear, - HourOfDay, MinuteOfHour, MonthOfYear, - TimeFeature, WeekOfYear, - time_features_from_frequency_str) -from pts.feature.transform import (AddAgeFeature, AddConstFeature, - AddObservedValuesIndicator, AddTimeFeatures, - AdhocTransform, AsNumpyArray, - CanonicalInstanceSplitter, Chain, - ConcatFeatures, ExpandDimArray, - FilterTransformation, FlatMapTransformation, - IdentityTransformation, InstanceSplitter, - ListFeatures, MapTransformation, - RemoveFields, RenameFields, SelectFields, - SetField, SimpleTransformation, SwapAxes, - Transformation, VstackFeatures) +from pts.feature.time_feature import ( + DayOfMonth, + DayOfWeek, + DayOfYear, + HourOfDay, + MinuteOfHour, + MonthOfYear, + TimeFeature, + WeekOfYear, + time_features_from_frequency_str, +) +from pts.feature.transform import ( + AddAgeFeature, + AddConstFeature, + AddObservedValuesIndicator, + AddTimeFeatures, + AdhocTransform, + AsNumpyArray, + CanonicalInstanceSplitter, + Chain, + ConcatFeatures, + ExpandDimArray, + FilterTransformation, + FlatMapTransformation, + IdentityTransformation, + InstanceSplitter, + ListFeatures, + MapTransformation, + RemoveFields, + RenameFields, + SelectFields, + SetField, + SimpleTransformation, + SwapAxes, + Transformation, + VstackFeatures, +) diff --git a/pts/feature/lag.py b/pts/feature/lag.py index 2bf88ce..47fc80c 100644 --- a/pts/feature/lag.py +++ b/pts/feature/lag.py @@ -28,9 +28,9 @@ def _make_lags(middle: int, delta: int) -> np.ndarray: return np.arange(middle - delta, middle + delta + 1).tolist() -def get_lags_for_frequency(freq_str: str, - lag_ub: int = 1200, - num_lags: Optional[int] = None) -> List[int]: +def get_lags_for_frequency( + freq_str: str, lag_ub: int = 1200, num_lags: Optional[int] = None +) -> List[int]: """ Generates a list of lags that that are appropriate for the given frequency string. @@ -56,39 +56,29 @@ def get_lags_for_frequency(freq_str: str, # Lags are target values at the same `season` (+/- delta) but in the previous cycle. def _make_lags_for_minute(multiple, num_cycles=3): # We use previous ``num_cycles`` hours to generate lags - return [ - _make_lags(k * 60 // multiple, 2) - for k in range(1, num_cycles + 1) - ] + return [_make_lags(k * 60 // multiple, 2) for k in range(1, num_cycles + 1)] def _make_lags_for_hour(multiple, num_cycles=7): # We use previous ``num_cycles`` days to generate lags - return [ - _make_lags(k * 24 // multiple, 1) - for k in range(1, num_cycles + 1) - ] + return [_make_lags(k * 24 // multiple, 1) for k in range(1, num_cycles + 1)] def _make_lags_for_day(multiple, num_cycles=4): # We use previous ``num_cycles`` weeks to generate lags # We use the last month (in addition to 4 weeks) to generate lag. - return [ - _make_lags(k * 7 // multiple, 1) for k in range(1, num_cycles + 1) - ] + [_make_lags(30 // multiple, 1)] + return [_make_lags(k * 7 // multiple, 1) for k in range(1, num_cycles + 1)] + [ + _make_lags(30 // multiple, 1) + ] def _make_lags_for_week(multiple, num_cycles=3): # We use previous ``num_cycles`` years to generate lags # Additionally, we use previous 4, 8, 12 weeks - return [ - _make_lags(k * 52 // multiple, 1) - for k in range(1, num_cycles + 1) - ] + [[4 // multiple, 8 // multiple, 12 // multiple]] + return [_make_lags(k * 52 // multiple, 1) for k in range(1, num_cycles + 1)] + [ + [4 // multiple, 8 // multiple, 12 // multiple] + ] def _make_lags_for_month(multiple, num_cycles=3): # We use previous ``num_cycles`` years to generate lags - return [ - _make_lags(k * 12 // multiple, 1) - for k in range(1, num_cycles + 1) - ] + return [_make_lags(k * 12 // multiple, 1) for k in range(1, num_cycles + 1)] # multiple, granularity = get_granularity(freq_str) offset = to_offset(freq_str) @@ -98,28 +88,29 @@ def get_lags_for_frequency(freq_str: str, elif offset.name == "W-SUN": lags = _make_lags_for_week(offset.n) elif offset.name == "D": - lags = _make_lags_for_day(offset.n) + _make_lags_for_week( - offset.n / 7.0) + lags = _make_lags_for_day(offset.n) + _make_lags_for_week(offset.n / 7.0) elif offset.name == "B": # todo find good lags for business day lags = [] elif offset.name == "H": - lags = (_make_lags_for_hour(offset.n) + - _make_lags_for_day(offset.n / 24.0) + - _make_lags_for_week(offset.n / (24.0 * 7))) + lags = ( + _make_lags_for_hour(offset.n) + + _make_lags_for_day(offset.n / 24.0) + + _make_lags_for_week(offset.n / (24.0 * 7)) + ) # minutes elif offset.name == "T": - lags = (_make_lags_for_minute(offset.n) + - _make_lags_for_hour(offset.n / 60.0) + - _make_lags_for_day(offset.n / (60.0 * 24)) + - _make_lags_for_week(offset.n / (60.0 * 24 * 7))) + lags = ( + _make_lags_for_minute(offset.n) + + _make_lags_for_hour(offset.n / 60.0) + + _make_lags_for_day(offset.n / (60.0 * 24)) + + _make_lags_for_week(offset.n / (60.0 * 24 * 7)) + ) else: raise Exception("invalid frequency") # flatten lags list and filter - lags = [ - int(lag) for sub_list in lags for lag in sub_list if 7 < lag <= lag_ub - ] + lags = [int(lag) for sub_list in lags for lag in sub_list if 7 < lag <= lag_ub] lags = [1, 2, 3, 4, 5, 6, 7] + sorted(list(set(lags))) return lags[:num_lags] diff --git a/pts/feature/time_feature.py b/pts/feature/time_feature.py index e15b342..967a556 100644 --- a/pts/feature/time_feature.py +++ b/pts/feature/time_feature.py @@ -20,6 +20,7 @@ class MinuteOfHour(TimeFeature): """ Minute of hour encoded as value between [-0.5, 0.5] """ + def __call__(self, index: pd.DatetimeIndex) -> np.ndarray: if self.normalized: return index.minute / 59.0 - 0.5 @@ -31,6 +32,7 @@ class HourOfDay(TimeFeature): """ Hour of day encoded as value between [-0.5, 0.5] """ + def __call__(self, index: pd.DatetimeIndex) -> np.ndarray: if self.normalized: return index.hour / 23.0 - 0.5 @@ -42,6 +44,7 @@ class DayOfWeek(TimeFeature): """ Hour of day encoded as value between [-0.5, 0.5] """ + def __call__(self, index: pd.DatetimeIndex) -> np.ndarray: if self.normalized: return index.dayofweek / 6.0 - 0.5 @@ -53,6 +56,7 @@ class DayOfMonth(TimeFeature): """ Day of month encoded as value between [-0.5, 0.5] """ + def __call__(self, index: pd.DatetimeIndex) -> np.ndarray: if self.normalized: return index.day / 30.0 - 0.5 @@ -64,6 +68,7 @@ class DayOfYear(TimeFeature): """ Day of year encoded as value between [-0.5, 0.5] """ + def __call__(self, index: pd.DatetimeIndex) -> np.ndarray: if self.normalized: return index.dayofyear / 364.0 - 0.5 @@ -75,6 +80,7 @@ class MonthOfYear(TimeFeature): """ Month of year encoded as value between [-0.5, 0.5] """ + def __call__(self, index: pd.DatetimeIndex) -> np.ndarray: if self.normalized: return index.month / 11.0 - 0.5 @@ -86,6 +92,7 @@ class WeekOfYear(TimeFeature): """ Week of year encoded as value between [-0.5, 0.5] """ + def __call__(self, index: pd.DatetimeIndex) -> np.ndarray: if self.normalized: return index.weekofyear / 51.0 - 0.5 @@ -114,9 +121,7 @@ def time_features_from_frequency_str(freq_str: str) -> List[TimeFeature]: elif granularity == "H": feature_classes = [HourOfDay, DayOfWeek, DayOfMonth, DayOfYear] elif granularity in ["min", "T"]: - feature_classes = [ - MinuteOfHour, HourOfDay, DayOfWeek, DayOfMonth, DayOfYear - ] + feature_classes = [MinuteOfHour, HourOfDay, DayOfWeek, DayOfMonth, DayOfYear] else: supported_freq_msg = f""" Unsupported frequency {freq_str} diff --git a/pts/feature/transform.py b/pts/feature/transform.py index 1f69a68..a9e9be1 100644 --- a/pts/feature/transform.py +++ b/pts/feature/transform.py @@ -25,15 +25,17 @@ def shift_timestamp(ts: pd.Timestamp, offset: int) -> pd.Timestamp: raise Exception(ex) -def target_transformation_length(target: np.array, pred_length: int, - is_train: bool) -> int: +def target_transformation_length( + target: np.array, pred_length: int, is_train: bool +) -> int: return target.shape[-1] + (0 if is_train else pred_length) class Transformation(ABC): @abstractmethod - def __call__(self, data_it: Iterator[DataEntry], - is_train: bool) -> Iterator[DataEntry]: + def __call__( + self, data_it: Iterator[DataEntry], is_train: bool + ) -> Iterator[DataEntry]: pass def estimate(self, data_it: Iterator[DataEntry]) -> Iterator[DataEntry]: @@ -44,11 +46,13 @@ class Chain(Transformation): """ Chain multiple transformations together. """ + def __init__(self, trans: List[Transformation]) -> None: self.trans = trans - def __call__(self, data_it: Iterator[DataEntry], - is_train: bool) -> Iterator[DataEntry]: + def __call__( + self, data_it: Iterator[DataEntry], is_train: bool + ) -> Iterator[DataEntry]: tmp = data_it for t in self.trans: tmp = t(tmp, is_train) @@ -59,8 +63,9 @@ class Chain(Transformation): class IdentityTransformation(Transformation): - def __call__(self, data_it: Iterator[DataEntry], - is_train: bool) -> Iterator[DataEntry]: + def __call__( + self, data_it: Iterator[DataEntry], is_train: bool + ) -> Iterator[DataEntry]: return data_it @@ -68,8 +73,8 @@ class MapTransformation(Transformation): """ Base class for Transformations that returns exactly one result per input in the stream. """ - def __call__(self, data_it: Iterator[DataEntry], - is_train: bool) -> Iterator: + + def __call__(self, data_it: Iterator[DataEntry], is_train: bool) -> Iterator: for data_entry in data_it: try: yield self.map_transform(data_entry.copy(), is_train) @@ -85,6 +90,7 @@ class SimpleTransformation(MapTransformation): """ Element wise transformations that are the same in train and test mode """ + def map_transform(self, data: DataEntry, is_train: bool) -> DataEntry: return self.transform(data) @@ -100,6 +106,7 @@ class AdhocTransform(SimpleTransformation): It is OK to use this for experiments and outside of a model pipeline that needs to be serialized. """ + def __init__(self, func: Callable[[DataEntry], DataEntry]) -> None: self.func = func @@ -112,14 +119,13 @@ class FlatMapTransformation(Transformation): Transformations that yield zero or more results per input, but do not combine elements from the input stream. """ - def __call__(self, data_it: Iterator[DataEntry], - is_train: bool) -> Iterator: + + def __call__(self, data_it: Iterator[DataEntry], is_train: bool) -> Iterator: num_idle_transforms = 0 for data_entry in data_it: num_idle_transforms += 1 try: - for result in self.flatmap_transform(data_entry.copy(), - is_train): + for result in self.flatmap_transform(data_entry.copy(), is_train): num_idle_transforms = 0 yield result except Exception as e: @@ -130,11 +136,11 @@ class FlatMapTransformation(Transformation): f"This means the transformation looped over " f"MAX_IDLE_TRANSFORMS={MAX_IDLE_TRANSFORMS} " f"inputs without returning any output.\n" - f"This occurred in the following transformation:\n{self}") + f"This occurred in the following transformation:\n{self}" + ) @abstractmethod - def flatmap_transform(self, data: DataEntry, - is_train: bool) -> Iterator[DataEntry]: + def flatmap_transform(self, data: DataEntry, is_train: bool) -> Iterator[DataEntry]: pass @@ -142,8 +148,7 @@ class FilterTransformation(FlatMapTransformation): def __init__(self, condition: Callable[[DataEntry], bool]) -> None: self.condition = condition - def flatmap_transform(self, data: DataEntry, - is_train: bool) -> Iterator[DataEntry]: + def flatmap_transform(self, data: DataEntry, is_train: bool) -> Iterator[DataEntry]: if self.condition(data): yield data @@ -169,6 +174,7 @@ class SetField(SimpleTransformation): value Value to be set """ + def __init__(self, output_field: str, value: Any) -> None: self.output_field = output_field self.value = value @@ -188,6 +194,7 @@ class SetFieldIfNotPresent(SimpleTransformation): value Value to be set """ + def __init__(self, field: str, value: Any) -> None: self.output_field = field self.value = value @@ -209,10 +216,10 @@ class AsNumpyArray(SimpleTransformation): dtype numpy dtype to use. """ - def __init__(self, - field: str, - expected_ndim: int, - dtype: np.dtype = np.float32) -> None: + + def __init__( + self, field: str, expected_ndim: int, dtype: np.dtype = np.float32 + ) -> None: self.field = field self.expected_ndim = expected_ndim self.dtype = dtype @@ -252,6 +259,7 @@ class ExpandDimArray(SimpleTransformation): axis Axis to expand (see np.expand_dims for details) """ + def __init__(self, field: str, axis: Optional[int] = None) -> None: self.field = field self.axis = axis @@ -275,21 +283,20 @@ class VstackFeatures(SimpleTransformation): drop_inputs If set to true the input fields will be dropped. """ - def __init__(self, - output_field: str, - input_fields: List[str], - drop_inputs: bool = True) -> None: + + def __init__( + self, output_field: str, input_fields: List[str], drop_inputs: bool = True + ) -> None: self.output_field = output_field self.input_fields = input_fields - self.cols_to_drop = ([] if not drop_inputs else [ - fname for fname in self.input_fields if fname != output_field - ]) + self.cols_to_drop = ( + [] + if not drop_inputs + else [fname for fname in self.input_fields if fname != output_field] + ) def transform(self, data: DataEntry) -> DataEntry: - r = [ - data[fname] for fname in self.input_fields - if data[fname] is not None - ] + r = [data[fname] for fname in self.input_fields if data[fname] is not None] output = np.vstack(r) data[self.output_field] = output for fname in self.cols_to_drop: @@ -310,21 +317,20 @@ class ConcatFeatures(SimpleTransformation): drop_inputs If set to true the input fields will be dropped. """ - def __init__(self, - output_field: str, - input_fields: List[str], - drop_inputs: bool = True) -> None: + + def __init__( + self, output_field: str, input_fields: List[str], drop_inputs: bool = True + ) -> None: self.output_field = output_field self.input_fields = input_fields - self.cols_to_drop = ([] if not drop_inputs else [ - fname for fname in self.input_fields if fname != output_field - ]) + self.cols_to_drop = ( + [] + if not drop_inputs + else [fname for fname in self.input_fields if fname != output_field] + ) def transform(self, data: DataEntry) -> DataEntry: - r = [ - data[fname] for fname in self.input_fields - if data[fname] is not None - ] + r = [data[fname] for fname in self.input_fields if data[fname] is not None] output = np.concatenate(r) data[self.output_field] = output for fname in self.cols_to_drop: @@ -342,6 +348,7 @@ class SwapAxes(SimpleTransformation): axes Axes to use """ + def __init__(self, input_fields: List[str], axes: Tuple[int, int]) -> None: self.input_fields = input_fields self.axis1, self.axis2 = axes @@ -359,7 +366,8 @@ class SwapAxes(SimpleTransformation): else: raise ValueError( f"Unexpected field type {type(v).__name__}, expected " - f"np.ndarray or list[np.ndarray]") + f"np.ndarray or list[np.ndarray]" + ) class ListFeatures(SimpleTransformation): @@ -374,15 +382,17 @@ class ListFeatures(SimpleTransformation): drop_inputs If true the input fields will be removed from the result. """ - def __init__(self, - output_field: str, - input_fields: List[str], - drop_inputs: bool = True) -> None: + + def __init__( + self, output_field: str, input_fields: List[str], drop_inputs: bool = True + ) -> None: self.output_field = output_field self.input_fields = input_fields - self.cols_to_drop = ([] if not drop_inputs else [ - fname for fname in self.input_fields if fname != output_field - ]) + self.cols_to_drop = ( + [] + if not drop_inputs + else [fname for fname in self.input_fields if fname != output_field] + ) def transform(self, data: DataEntry) -> DataEntry: data[self.output_field] = [data[fname] for fname in self.input_fields] @@ -411,13 +421,14 @@ class AddObservedValuesIndicator(SimpleTransformation): result. dtype """ + def __init__( - self, - target_field: str, - output_field: str, - dummy_value: int = 0, - convert_nans: bool = True, - dtype: np.dtype = np.float32, + self, + target_field: str, + output_field: str, + dummy_value: int = 0, + convert_nans: bool = True, + dtype: np.dtype = np.float32, ) -> None: self.dummy_value = dummy_value self.target_field = target_field @@ -447,6 +458,7 @@ class RenameFields(SimpleTransformation): mapping Name mapping `input_name -> output_name` """ + def __init__(self, mapping: Dict[str, str]) -> None: self.mapping = mapping values_count = Counter(mapping.values()) @@ -484,13 +496,14 @@ class AddConstFeature(MapTransformation): dtype Numpy dtype to use for resulting array. """ + def __init__( - self, - output_field: str, - target_field: str, - pred_length: int, - const: float = 1.0, - dtype: np.dtype = np.float32, + self, + output_field: str, + target_field: str, + pred_length: int, + const: float = 1.0, + dtype: np.dtype = np.float32, ) -> None: self.pred_length = pred_length self.const = const @@ -499,11 +512,12 @@ class AddConstFeature(MapTransformation): self.target_field = target_field def map_transform(self, data: DataEntry, is_train: bool) -> DataEntry: - length = target_transformation_length(data[self.target_field], - self.pred_length, - is_train=is_train) - data[self.output_field] = self.const * np.ones(shape=(1, length), - dtype=self.dtype) + length = target_transformation_length( + data[self.target_field], self.pred_length, is_train=is_train + ) + data[self.output_field] = self.const * np.ones( + shape=(1, length), dtype=self.dtype + ) return data @@ -525,13 +539,14 @@ class AddTimeFeatures(MapTransformation): pred_length Prediction length """ + def __init__( - self, - start_field: str, - target_field: str, - output_field: str, - time_features: List[TimeFeature], - pred_length: int, + self, + start_field: str, + target_field: str, + output_field: str, + time_features: List[TimeFeature], + pred_length: int, ) -> None: self.date_features = time_features self.pred_length = pred_length @@ -551,26 +566,26 @@ class AddTimeFeatures(MapTransformation): if self._min_time_point is None: self._min_time_point = start self._max_time_point = end - self._min_time_point = min(shift_timestamp(start, -50), - self._min_time_point) - self._max_time_point = max(shift_timestamp(end, 50), - self._max_time_point) - self.full_date_range = pd.date_range(self._min_time_point, - self._max_time_point, - freq=start.freq) + self._min_time_point = min(shift_timestamp(start, -50), self._min_time_point) + self._max_time_point = max(shift_timestamp(end, 50), self._max_time_point) + self.full_date_range = pd.date_range( + self._min_time_point, self._max_time_point, freq=start.freq + ) self._full_range_date_features = np.vstack( - [feat(self.full_date_range) for feat in self.date_features]) - self._date_index = pd.Series(index=self.full_date_range, - data=np.arange(len(self.full_date_range))) + [feat(self.full_date_range) for feat in self.date_features] + ) + self._date_index = pd.Series( + index=self.full_date_range, data=np.arange(len(self.full_date_range)) + ) def map_transform(self, data: DataEntry, is_train: bool) -> DataEntry: start = data[self.start_field] - length = target_transformation_length(data[self.target_field], - self.pred_length, - is_train=is_train) + length = target_transformation_length( + data[self.target_field], self.pred_length, is_train=is_train + ) self._update_cache(start, length) i0 = self._date_index[start] - features = self._full_range_date_features[..., i0:i0 + length] + features = self._full_range_date_features[..., i0 : i0 + length] data[self.output_field] = features return data @@ -594,13 +609,14 @@ class AddAgeFeature(MapTransformation): If set to true the age feature grows logarithmically otherwise linearly over time. dtype """ + def __init__( - self, - target_field: str, - output_field: str, - pred_length: int, - log_scale: bool = True, - dtype: np.dtype = np.float32, + self, + target_field: str, + output_field: str, + pred_length: int, + log_scale: bool = True, + dtype: np.dtype = np.float32, ) -> None: self.pred_length = pred_length self.target_field = target_field @@ -610,9 +626,9 @@ class AddAgeFeature(MapTransformation): self.dtype = dtype def map_transform(self, data: DataEntry, is_train: bool) -> DataEntry: - length = target_transformation_length(data[self.target_field], - self.pred_length, - is_train=is_train) + length = target_transformation_length( + data[self.target_field], self.pred_length, is_train=is_train + ) if self.log_scale: age = np.log10(2.0 + np.arange(length, dtype=self.dtype)) @@ -669,18 +685,19 @@ class InstanceSplitter(FlatMapTransformation): cold-start. In such case, is_pad_out contains an indicator whether data is padded or not. """ + def __init__( - self, - target_field: str, - is_pad_field: str, - start_field: str, - forecast_start_field: str, - train_sampler: InstanceSampler, - past_length: int, - future_length: int, - batch_first: bool = True, - time_series_fields: Optional[List[str]] = None, - pick_incomplete: bool = True, + self, + target_field: str, + is_pad_field: str, + start_field: str, + forecast_start_field: str, + train_sampler: InstanceSampler, + past_length: int, + future_length: int, + batch_first: bool = True, + time_series_fields: Optional[List[str]] = None, + pick_incomplete: bool = True, ) -> None: assert future_length > 0 @@ -702,8 +719,7 @@ class InstanceSplitter(FlatMapTransformation): def _future(self, col_name): return f"future_{col_name}" - def flatmap_transform(self, data: DataEntry, - is_train: bool) -> Iterator[DataEntry]: + def flatmap_transform(self, data: DataEntry, is_train: bool) -> Iterator[DataEntry]: pl = self.future_length slice_cols = self.ts_fields + [self.target_field] target = data[self.target_field] @@ -720,11 +736,12 @@ class InstanceSplitter(FlatMapTransformation): else: if self.pick_incomplete: sampling_indices = self.train_sampler( - target, 0, len_target - self.future_length) + target, 0, len_target - self.future_length + ) else: sampling_indices = self.train_sampler( - target, self.past_length, - len_target - self.future_length) + target, self.past_length, len_target - self.future_length + ) else: sampling_indices = [len_target] for i in sampling_indices: @@ -735,17 +752,18 @@ class InstanceSplitter(FlatMapTransformation): for ts_field in slice_cols: if i > self.past_length: # truncate to past_length - past_piece = d[ts_field][..., i - self.past_length:i] + past_piece = d[ts_field][..., i - self.past_length : i] elif i < self.past_length: - pad_block = np.zeros(d[ts_field].shape[:-1] + - (pad_length, ), - dtype=d[ts_field].dtype) + pad_block = np.zeros( + d[ts_field].shape[:-1] + (pad_length,), dtype=d[ts_field].dtype + ) past_piece = np.concatenate( - [pad_block, d[ts_field][..., :i]], axis=-1) + [pad_block, d[ts_field][..., :i]], axis=-1 + ) else: past_piece = d[ts_field][..., :i] d[self._past(ts_field)] = past_piece - d[self._future(ts_field)] = d[ts_field][..., i:i + pl] + d[self._future(ts_field)] = d[ts_field][..., i : i + pl] del d[ts_field] pad_indicator = np.zeros(self.past_length) if pad_length > 0: @@ -753,14 +771,11 @@ class InstanceSplitter(FlatMapTransformation): if self.batch_first: for ts_field in slice_cols: - d[self._past(ts_field)] = d[self._past( - ts_field)].transpose() - d[self._future(ts_field)] = d[self._future( - ts_field)].transpose() + d[self._past(ts_field)] = d[self._past(ts_field)].transpose() + d[self._future(ts_field)] = d[self._future(ts_field)].transpose() d[self._past(self.is_pad_field)] = pad_indicator - d[self.forecast_start_field] = shift_timestamp( - d[self.start_field], i) + d[self.forecast_start_field] = shift_timestamp(d[self.start_field], i) yield d @@ -813,20 +828,21 @@ class CanonicalInstanceSplitter(FlatMapTransformation): length of the prediction range, must be set if use_prediction_features is True """ + def __init__( - self, - target_field: str, - is_pad_field: str, - start_field: str, - forecast_start_field: str, - instance_sampler: InstanceSampler, - instance_length: int, - batch_first: bool = True, - time_series_fields: List[str] = [], - allow_target_padding: bool = False, - pad_value: float = 0.0, - use_prediction_features: bool = False, - prediction_length: Optional[int] = None, + self, + target_field: str, + is_pad_field: str, + start_field: str, + forecast_start_field: str, + instance_sampler: InstanceSampler, + instance_length: int, + batch_first: bool = True, + time_series_fields: List[str] = [], + allow_target_padding: bool = False, + pad_value: float = 0.0, + use_prediction_features: bool = False, + prediction_length: Optional[int] = None, ) -> None: self.instance_sampler = instance_sampler self.instance_length = instance_length @@ -852,8 +868,7 @@ class CanonicalInstanceSplitter(FlatMapTransformation): def _future(self, col_name): return f"future_{col_name}" - def flatmap_transform(self, data: DataEntry, - is_train: bool) -> Iterator[DataEntry]: + def flatmap_transform(self, data: DataEntry, is_train: bool) -> Iterator[DataEntry]: ts_fields = self.dynamic_feature_fields + [self.target_field] ts_target = data[self.target_field] @@ -863,10 +878,14 @@ class CanonicalInstanceSplitter(FlatMapTransformation): if len_target < self.instance_length: sampling_indices = ( # Returning [] for all time series will cause this to be in loop forever! - [len_target] if self.allow_target_padding else []) + [len_target] + if self.allow_target_padding + else [] + ) else: sampling_indices = self.instance_sampler( - ts_target, self.instance_length, len_target) + ts_target, self.instance_length, len_target + ) else: sampling_indices = [len_target] @@ -876,8 +895,9 @@ class CanonicalInstanceSplitter(FlatMapTransformation): pad_length = max(self.instance_length - i, 0) # update start field - d[self.start_field] = shift_timestamp(data[self.start_field], - i - self.instance_length) + d[self.start_field] = shift_timestamp( + data[self.start_field], i - self.instance_length + ) # set is_pad field is_pad = np.zeros(self.instance_length) @@ -890,26 +910,28 @@ class CanonicalInstanceSplitter(FlatMapTransformation): full_ts = data[ts_field] if pad_length > 0: pad_pre = self.pad_value * np.ones( - shape=full_ts.shape[:-1] + (pad_length, )) - past_ts = np.concatenate([pad_pre, full_ts[..., :i]], - axis=-1) + shape=full_ts.shape[:-1] + (pad_length,) + ) + past_ts = np.concatenate([pad_pre, full_ts[..., :i]], axis=-1) else: - past_ts = full_ts[..., (i - self.instance_length):i] + past_ts = full_ts[..., (i - self.instance_length) : i] past_ts = past_ts.transpose() if self.batch_first else past_ts d[self._past(ts_field)] = past_ts if self.use_prediction_features and not is_train: if not ts_field == self.target_field: - future_ts = full_ts[..., i:i + self.prediction_length] - future_ts = (future_ts.transpose() - if self.batch_first else future_ts) + future_ts = full_ts[..., i : i + self.prediction_length] + future_ts = ( + future_ts.transpose() if self.batch_first else future_ts + ) d[self._future(ts_field)] = future_ts del d[ts_field] d[self.forecast_start_field] = shift_timestamp( - d[self.start_field], self.instance_length) + d[self.start_field], self.instance_length + ) yield d @@ -922,6 +944,7 @@ class SelectFields(MapTransformation): input_fields List of fields to keep. """ + def __init__(self, input_fields: List[str]) -> None: self.input_fields = input_fields diff --git a/pts/model/deepar/deepar_estimator.py b/pts/model/deepar/deepar_estimator.py index b27fa53..7aa1606 100644 --- a/pts/model/deepar/deepar_estimator.py +++ b/pts/model/deepar/deepar_estimator.py @@ -3,13 +3,17 @@ from typing import List, Optional import numpy as np from pts import Trainer -from pts.feature import (TimeFeature, get_lags_for_frequency, - time_features_from_frequency_str) +from pts.feature import ( + TimeFeature, + get_lags_for_frequency, + time_features_from_frequency_str, +) from pts.model import PTSEstimator class DeepAREstimator(PTSEstimator): - def __init__(self, + def __init__( + self, freq: str, prediction_length: int, trainer: Trainer = Trainer(), @@ -28,10 +32,9 @@ class DeepAREstimator(PTSEstimator): lags_seq: Optional[List[int]] = None, time_features: Optional[List[TimeFeature]] = None, num_parallel_samples: int = 100, - dtype : np.dtype = np.float32, + dtype: np.dtype = np.float32, ) -> None: super().__init__(trainer=trainer) - self.freq = freq self.context_length = ( @@ -47,9 +50,7 @@ class DeepAREstimator(PTSEstimator): self.use_feat_dynamic_real = use_feat_dynamic_real self.use_feat_static_cat = use_feat_static_cat self.use_feat_static_real = use_feat_static_real - self.cardinality = ( - cardinality if cardinality and use_feat_static_cat else [1] - ) + self.cardinality = cardinality if cardinality and use_feat_static_cat else [1] self.embedding_dimension = ( embedding_dimension if embedding_dimension is not None @@ -57,9 +58,7 @@ class DeepAREstimator(PTSEstimator): ) self.scaling = scaling self.lags_seq = ( - lags_seq - if lags_seq is not None - else get_lags_for_frequency(freq_str=freq) + lags_seq if lags_seq is not None else get_lags_for_frequency(freq_str=freq) ) self.time_features = ( time_features diff --git a/pts/model/estimator.py b/pts/model/estimator.py index 83169a7..07eccbf 100644 --- a/pts/model/estimator.py +++ b/pts/model/estimator.py @@ -33,6 +33,7 @@ class DummyEstimator(Estimator): **kwargs Keyword arguments to pass to the predictor constructor. """ + def __init__(self, predictor_cls: type, **kwargs) -> None: self.predictor = predictor_cls(**kwargs) @@ -78,8 +79,9 @@ class PTSEstimator(Estimator): pass @abstractmethod - def create_predictor(self, transformation: Transformation, - trained_network: nn.Module) -> Predictor: + def create_predictor( + self, transformation: Transformation, trained_network: nn.Module + ) -> Predictor: """ Create and return a predictor object. diff --git a/pts/model/forecast.py b/pts/model/forecast.py index f37808e..62778ef 100644 --- a/pts/model/forecast.py +++ b/pts/model/forecast.py @@ -15,7 +15,7 @@ class OutputType(str, Enum): quantiles = "quantiles" -class Config(): +class Config: output_types: Set[OutputType] = {"quantiles", "mean"} quantiles: List[str] = ["0.1", "0.5", "0.9"] @@ -102,25 +102,25 @@ class SampleForecast(Forecast): additional information that the forecaster may provide e.g. estimated parameters, number of iterations ran etc. """ + @validated() def __init__( - self, - samples: Union[torch.Tensor, np.ndarray], - start_date, - freq, - item_id: Optional[str] = None, - info: Optional[Dict] = None, + self, + samples: Union[torch.Tensor, np.ndarray], + start_date, + freq, + item_id: Optional[str] = None, + info: Optional[Dict] = None, ): assert isinstance( - samples, - (np.ndarray, torch.Tensor - )), "samples should be either a numpy array or an torch tensor" + samples, (np.ndarray, torch.Tensor) + ), "samples should be either a numpy array or an torch tensor" assert ( len(np.shape(samples)) == 2 or len(np.shape(samples)) == 3 ), "samples should be a 2-dimensional or 3-dimensional array. Dimensions found: {}".format( - len(np.shape(samples))) - self.samples = (samples if - (isinstance(samples, np.ndarray)) else samples.numpy()) + len(np.shape(samples)) + ) + self.samples = samples if (isinstance(samples, np.ndarray)) else samples.numpy() self._sorted_samples_value = None self._mean = None self._dim = None @@ -128,8 +128,8 @@ class SampleForecast(Forecast): self.info = info assert isinstance( - start_date, - pd.Timestamp), "start_date should be a pandas Timestamp object" + start_date, pd.Timestamp + ), "start_date should be a pandas Timestamp object" self.start_date = start_date assert isinstance(freq, str), "freq should be a string" @@ -184,14 +184,17 @@ class SampleForecast(Forecast): target_dim = self.samples.shape[2] assert dim < target_dim, ( f"must set 0 <= dim < target_dim, but got dim={dim}," - f" target_dim={target_dim}") + f" target_dim={target_dim}" + ) samples = self.samples[:, :, dim] - return SampleForecast(samples=samples, - start_date=self.start_date, - freq=self.freq, - item_id=self.item_id, - info=self.info) + return SampleForecast( + samples=samples, + start_date=self.start_date, + freq=self.freq, + item_id=self.item_id, + info=self.info, + ) def dim(self) -> int: if self._dim is not None: @@ -215,13 +218,15 @@ class SampleForecast(Forecast): return result def __repr__(self): - return ", ".join([ - f"SampleForecast({self.samples!r})", - f"{self.start_date!r}", - f"{self.freq!r}", - f"item_id={self.item_id!r}", - f"info={self.info!r})", - ]) + return ", ".join( + [ + f"SampleForecast({self.samples!r})", + f"{self.start_date!r}", + f"{self.freq!r}", + f"item_id={self.item_id!r}", + f"info={self.info!r})", + ] + ) class QuantileForecast(Forecast): @@ -244,14 +249,15 @@ class QuantileForecast(Forecast): additional information that the forecaster may provide e.g. estimated parameters, number of iterations ran etc. """ + def __init__( - self, - forecast_arrays: np.ndarray, - start_date: pd.Timestamp, - freq: str, - forecast_keys: List[str], - item_id: Optional[str] = None, - info: Optional[Dict] = None, + self, + forecast_arrays: np.ndarray, + start_date: pd.Timestamp, + freq: str, + forecast_keys: List[str], + item_id: Optional[str] = None, + info: Optional[Dict] = None, ): self.forecast_array = forecast_arrays self.start_date = pd.Timestamp(start_date, freq=freq) @@ -269,11 +275,11 @@ class QuantileForecast(Forecast): shape = self.forecast_array.shape assert shape[0] == len(self.forecast_keys), ( f"The forecast_array (shape={shape} should have the same " - f"length as the forecast_keys (len={len(self.forecast_keys)}).") + f"length as the forecast_keys (len={len(self.forecast_keys)})." + ) self.prediction_length = shape[-1] self._forecast_dict = { - k: self.forecast_array[i] - for i, k in enumerate(self.forecast_keys) + k: self.forecast_array[i] for i, k in enumerate(self.forecast_keys) } self._nan_out = np.array([np.nan] * self.prediction_length) @@ -294,22 +300,26 @@ class QuantileForecast(Forecast): if self._dim is not None: return self._dim else: - if (len(self.forecast_array.shape) == 2 - ): # 1D target. shape: (num_samples, prediction_length) + if ( + len(self.forecast_array.shape) == 2 + ): # 1D target. shape: (num_samples, prediction_length) return 1 else: return self.forecast_array.shape[ - 1] # 2D target. shape: (num_samples, target_dim, prediction_length) + 1 + ] # 2D target. shape: (num_samples, target_dim, prediction_length) def __repr__(self): - return ", ".join([ - f"QuantileForecast({self.forecast_array!r})", - f"start_date={self.start_date!r}", - f"freq={self.freq!r}", - f"forecast_keys={self.forecast_keys!r}", - f"item_id={self.item_id!r}", - f"info={self.info!r})", - ]) + return ", ".join( + [ + f"QuantileForecast({self.forecast_array!r})", + f"start_date={self.start_date!r}", + f"freq={self.freq!r}", + f"forecast_keys={self.forecast_keys!r}", + f"item_id={self.item_id!r}", + f"info={self.info!r})", + ] + ) # class DistributionForecast(Forecast): diff --git a/pts/model/quantile.py b/pts/model/quantile.py index 9411ff4..cf787a3 100644 --- a/pts/model/quantile.py +++ b/pts/model/quantile.py @@ -21,8 +21,7 @@ class Quantile(NamedTuple): @classmethod def checked(cls, value: float, name: str) -> "Quantile": if not 0 <= value <= 1: - raise Exception( - f"quantile value should be in [0, 1] but found {value}") + raise Exception(f"quantile value should be in [0, 1] but found {value}") return Quantile(value, name) diff --git a/pts/modules/distribution_output.py b/pts/modules/distribution_output.py index 5db192c..4c01ebd 100644 --- a/pts/modules/distribution_output.py +++ b/pts/modules/distribution_output.py @@ -1,8 +1,7 @@ -from typing import Callable, Dict, Optional, Tuple from abc import ABC, abstractmethod +from typing import Callable, Dict, Optional, Tuple import numpy as np - import torch import torch.nn as nn @@ -30,5 +29,6 @@ class ArgProj(nn.Module): return self.domain_map(*params_unbounded) + class Output(ABC): - \ No newline at end of file + pass diff --git a/pts/trainer.py b/pts/trainer.py index 5bc58ca..426fa63 100644 --- a/pts/trainer.py +++ b/pts/trainer.py @@ -9,20 +9,23 @@ from .dataset import TrainDataLoader class Trainer: - def __init__(self, - epochs: int = 100, - batch_size: int = 32, - num_batches_per_epoch: int = 50, - learning_rate: float = 1e-3, - device: Optional[torch.device] = None) -> None: + def __init__( + self, + epochs: int = 100, + batch_size: int = 32, + num_batches_per_epoch: int = 50, + learning_rate: float = 1e-3, + device: Optional[torch.device] = None, + ) -> None: self.epochs = epochs self.batch_size = batch_size self.num_batches_per_epoch = num_batches_per_epoch self.learning_rate = learning_rate self.device = device - def __call__(self, net: nn.Module, input_names: List[str], - train_iter: TrainDataLoader) -> None: + def __call__( + self, net: nn.Module, input_names: List[str], train_iter: TrainDataLoader + ) -> None: net.to(self.device) @@ -45,6 +48,6 @@ class Trainer: loss.backward() optimizer.step() - + # mark epoch end time and log time cost of current epoch toc = time.time()