wassname/pytorch-ts
1
0
Fork 0
mirror of https://github.com/wassname/pytorch-ts.git synced 2026-07-19 11:27:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

ran black

Browse Source
This commit is contained in:
Kashif Rasul
2019-10-30 09:42:02 +01:00
parent 4ad01ea2e3
commit 5772c04ffe
17 changed files with 382 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pts/dataset/__init__.py
+7 -3
View File
@@ -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
pts/dataset/list_dataset.py
+4 -5
View File
@@ -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):
pts/dataset/loader.py
+1
View File
@@ -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
pts/dataset/process.py
+19 -29
View File
@@ -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
),
],
)
pts/dataset/sampler.py
+11 -8
View File
@@ -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
pts/dataset/transformed_dataset.py
+4 -2
View File
@@ -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)
pts/exception.py
-1
View File
@@ -1,4 +1,3 @@
def assert_pts(condition: bool, message: str, *args, **kwargs) -> None:
if not condition:
raise Exception(message.format(*args, **kwargs))
pts/feature/__init__.py
+37 -15
View File
@@ -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,
)
pts/feature/lag.py
+25 -34
View File
@@ -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]
pts/feature/time_feature.py
+8 -3
View File
@@ -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}
pts/feature/transform.py
+177 -154
View File
@@ -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
pts/model/deepar/deepar_estimator.py
+10 -11
View File
@@ -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
pts/model/estimator.py
+4 -2
View File
@@ -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.
pts/model/forecast.py
+59 -49
View File
@@ -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):
pts/model/quantile.py
+1 -2
View File
@@ -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)
pts/modules/distribution_output.py
+3 -3
View File
@@ -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):
pass
pts/trainer.py
+12 -9
View File
@@ -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()