timeseriesx/base/base_time_series.py
import copy
from abc import ABCMeta, abstractmethod
from collections.abc import Iterable
import numpy as np
class BaseTimeSeries(metaclass=ABCMeta):
# ------------------------------ properties ------------------------------ #
@property
def empty(self):
return self._series.empty
@property
def values(self):
raise NotImplementedError()
@property
def first(self):
raise NotImplementedError()
@property
def last(self):
raise NotImplementedError()
# ---------------------------- functionality ----------------------------- #
def map(self, func, **kwargs):
"""
apply a custom function to each value of the series
:param function func: a function mapping a scalar to another scalar
:return: self
:rtype: BaseTimeSeries
"""
self._series = self._series.apply(func)
return self
def aggregate(self, func):
"""
aggregate all values of the series with a custom aggregation function
:param function func: a function mapping a numeric list/array/vector to a scalar
:return: the aggregated value
:rtype: numpy.float/numpy.int
"""
return self._series.agg(func)
def sum(self):
return self.aggregate(np.sum)
def mean(self):
return self.aggregate(np.mean)
def round(self, decimals):
raise NotImplementedError()
def append(self, value):
raise NotImplementedError()
def prepend(self, value):
raise NotImplementedError()
def join(self, other_ts, fit=True):
raise NotImplementedError()
# --------------------------------- cast --------------------------------- #
@abstractmethod
def as_tuples(self):
raise NotImplementedError()
@abstractmethod
def as_dict(self, ordered=False):
raise NotImplementedError()
def as_pd_series(self, include_nan=True):
if include_nan:
return self._series
else:
return self._series[self._series.notnull()]
# ---------------------------- magic methods ----------------------------- #
def __getitem__(self, item):
return NotImplemented()
def __setitem__(self, key, value):
return NotImplemented()
def __len__(self):
return self._series.size
def __eq__(self, other):
return NotImplemented()
def __iter__(self):
self._i = 0
return self
def __next__(self):
if self._i < len(self):
result = self[self._i]
self._i += 1
return result
else:
raise StopIteration
def __add__(self, other):
return self._basic_calc('add', other, fill_value=0)
def __sub__(self, other):
return self._basic_calc('subtract', other, fill_value=0)
def __mul__(self, other):
return self._basic_calc('__mul__', other)
def __truediv__(self, other):
return self._basic_calc('__truediv__', other)
def __floordiv__(self, other):
return self._basic_calc('__floordiv__', other)
def __str__(self):
raise NotImplementedError()
def __repr__(self):
raise NotImplementedError()
# ---------------------------- calculations ------------------------------ #
@abstractmethod
def _basic_calc(self, operation, other, **kwargs):
"""
:param str operation:
:param class:`BaseTimeSeries`/class:`collections.Sequence`/
class:`numpy.ndarray`/class:`numbers.Number` other:
:return: a new time series copy holding the results of the calculation
:rtype: BaseTimeSeries
"""
raise NotImplementedError()