diff --git a/pts/__init__.py b/pts/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pts/dataset/__init__.py b/pts/dataset/__init__.py new file mode 100644 index 0000000..b995b7c --- /dev/null +++ b/pts/dataset/__init__.py @@ -0,0 +1 @@ +from pts.dataset.list_dataset import ListDataset diff --git a/pts/dataset/common.py b/pts/dataset/common.py new file mode 100644 index 0000000..1cc464d --- /dev/null +++ b/pts/dataset/common.py @@ -0,0 +1,21 @@ +from abc import ABC, abstractmethod + +from typing import Any, Dict, Sized, Iterable, NamedTuple + + +DataEntry = Dict[str, Any] + + +class SourceContext(NamedTuple): + source: str + row: int + + +class Dataset(Sized, Iterable[DataEntry], ABC): + @abstractmethod + def __iter__(self) -> Iterable[DataEntry]: + pass + + @abstractmethod + def __len__(self): + pass diff --git a/pts/dataset/list_dataset.py b/pts/dataset/list_dataset.py new file mode 100644 index 0000000..543a2d9 --- /dev/null +++ b/pts/dataset/list_dataset.py @@ -0,0 +1,21 @@ +from typing import Iterable + +from .common import Dataset, DataEntry, SourceContext +from .process import ProcessDataEntry + + +class ListDataset(Dataset): + 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) + yield data + + def __len__(self): + return len(self.list_data) diff --git a/pts/dataset/process.py b/pts/dataset/process.py new file mode 100644 index 0000000..6f8a7a3 --- /dev/null +++ b/pts/dataset/process.py @@ -0,0 +1,101 @@ +from functools import lru_cache +from typing import Callable, List, cast + +import numpy as np +import pandas as pd +from pandas.tseries.frequencies import to_offset + +from .common import DataEntry + + +class ProcessStartField: + def __init__(self, name: str, freq: str) -> None: + self.name = name + self.freq = freq + + def __call__(self, data: DataEntry) -> DataEntry: + 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}"') + + data[self.name] = value + + return data + + @staticmethod + @lru_cache(maxsize=10000) + def process(string: str, freq: str) -> pd.Timestamp: + timestamp = pd.Timestamp(string, freq=freq) + # 'W-SUN' is the standardized freqstr for W + if timestamp.freq.name in ("M", "W-SUN"): + offset = to_offset(freq) + timestamp = timestamp.replace( + hour=0, minute=0, second=0, microsecond=0, nanosecond=0 + ) + return pd.Timestamp(offset.rollback(timestamp), freq=offset.freqstr) + if timestamp.freq == "B": + # does not floor on business day as it is not allowed + return timestamp + return pd.Timestamp(timestamp.floor(timestamp.freq), freq=timestamp.freq) + + +class ProcessTimeSeriesField: + 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 + self.dtype = np.int32 if is_cat else np.float32 + + def __call__(self, data: DataEntry) -> DataEntry: + value = data.get(self.name, None) + + if value is not None: + value = np.asarray(value, dtype=self.dtype) + dim_diff = self.req_ndim - value.ndim + if dim_diff == 1: + value = np.expand_dims(a=value, axis=0) + elif dim_diff != 0: + raise Exception( + f"JSON array has bad shape - expected {self.req_ndim} dimensions got {dim_diff}" + ) + + data[self.name] = value + return data + elif not self.is_required: + return data + else: + raise Exception(f"JSON object is missing a required field `{self.name}`") + + +class ProcessDataEntry: + def __init__(self, freq: str, one_dim_target: bool = True) -> None: + self.trans = cast( + 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( + "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 + ), + ], + ) + + def __call__(self, data: DataEntry) -> DataEntry: + for t in self.trans: + data = t(data) + return data