piecewise_trend

class PiecewiseTrend(TimeRelatedTransformer):
    """Transformation to create a PiecewiseTrend variable from a pandas series.
        This transformation needs to be fitted by a referenced dates to be computed for any new dates.

    Args:
        slopes (list[float]): The slope indicates the steepness of a line.
            Ex: [-239.57, 73.59, 55.64, 104.45, -17.41]
        intercepts (list[float]): Indicates the location in the y-axis at each start of time range.
            Ex : [-2736.58, 7576.09, -6994.35, -6302.76, -4248.78]
        times_ranges (list[tuple[int, int]]): List of times ranges for each linear function of the PiecewiseTrend.
            Ex: [(0,4)] represents first to 5th period from the start_date.
        start_date (Optional[np.datetime64], optional): start_date date to identify the date when the sequence starts.
            Ex: np.datetime64("2018-01-01"). Optional, will be computed on fitted dates if not specified.
        granularity (Optional[np.timedelta64], optional): time period between two observations.
            Ex: np.timedelta64(7,'D') for weekly. Optional, will be computed on fitted dates if not specified.

    Attributes:
        slopes (list[float]): The slope indicates the steepness of a line.
            Ex: [-239.57, 73.59, 55.64, 104.45, -17.41]
        intercepts (list[float]): Indicates the location in the y-axis at each start of time range.
            Ex : [-2736.58, 7576.09, -6994.35, -6302.76, -4248.78]
        times_ranges (list[tuple[int, int]]): List of times ranges for each linear function of the PiecewiseTrend.
            Ex: [(0,4)] represents first to 5th period from the start_date.
        start_date (Optional[np.datetime64], optional): start_date date to identify the date when the sequence starts.
            Ex: np.datetime64("2018-01-01"). Optional, will be computed on fitted dates if not specified.
        granularity (Optional[np.timedelta64], optional): time period between two observations.
            Ex: np.timedelta64(7,'D') for weekly. Optional, will be computed on fitted dates if not specified.

    """

    def __init__(
        self,
        slopes: list[float],
        intercepts: list[float],
        times_ranges: list[tuple[int, int]],
        start_date: Optional[np.datetime64] = None,
        granularity: Optional[np.timedelta64] = None,
    ) -> None:

        self.slopes: list[float] = slopes
        self.intercepts: list[float] = intercepts
        self.times_ranges: list[tuple[int, int]] = times_ranges
        self.start_date: Optional[np.datetime64] = start_date if start_date else None
        self.granularity: Optional[np.timedelta64] = granularity if granularity else None

    @property
    def parameters(self) -> Dict[str, float]:
        """Returns instant transformation parameters

        Returns:
            (Dict[str, float]): Instant transformation parameters
        """
        return self.__dict__

    # ------- METHODS -------

    def fit(self, dates: npt.NDArray[np.datetime64]):
        """Fit transformation on dates by computing start_date and granularity

        Args:
            dates (npt.NDArray[np.datetime64]): dates to fit on
        """
        seasonal_scenario = get_seasonal_scenario(dates)
        self.start_date = seasonal_scenario["start_date"]
        self.granularity = seasonal_scenario["granularity"]

        return super().fit(dates)

    def transform(self, dates: npt.NDArray[np.datetime64]) -> npt.NDArray[np.float64]:
        """Method to compute PiecewiseTrend value for dates

        Args:
            dates (npt.NDArray[np.datetime64]): dates to transform
            times_ranges (list[tuple[int, int]]): List of times ranges for each linear function of the PiecewiseTrend.
                Ex: [(0,4)] represents first to 5th period after the start_date.
            intercepts (list[float]): Indicates the location in the y-axis at each start of time range.
            slopes (list[float]): The slope indicates the steepness of a line.
            start_date (numpy.datetime64): start_date date to identify the date when the sequence starts.
                Ex: np.datetime64("2018-01-01").
            granularity (np.timedelta64): time period between two observations. Ex: np.timedelta64(7,'D') for weekly.

        Returns:
            (npt.NDArray[np.float64]): computed value for each date
        """
        super().transform(dates)

        if not self.start_date or not self.granularity:
            raise PiecewiseTrendError("start_date and granularity must be filled")

        return self._transformer(
            dates, self.times_ranges, self.intercepts, self.slopes, self.start_date, self.granularity
        )

    # ------- TRANSFORMERS -------

    @staticmethod
    def _transformer(
        dates: npt.NDArray[np.datetime64],
        times_ranges: list[tuple[int, int]],
        intercepts: list[float],
        slopes: list[float],
        start_date: np.datetime64,
        granularity: np.timedelta64,
    ) -> npt.NDArray[np.float64]:
        """Method to compute PiecewiseTrend value for dates

        Args:
            dates (npt.NDArray[np.datetime64]): dates to transform
            times_ranges (list[tuple[int, int]]): List of times ranges for each linear function of the PiecewiseTrend.
                Ex: [(0,4)] represents first to 5th period after the start_date.
            intercepts (list[float]): Indicates the location in the y-axis at each start of time range.
            slopes (list[float]): The slope indicates the steepness of a line.
            start_date (numpy.datetime64): start_date date to identify the date when the sequence starts.
                Ex: np.datetime64("2018-01-01").
            granularity (np.timedelta64): time period between two observations. Ex: np.timedelta64(7,'D') for weekly.

        Returns:
            (npt.NDArray[np.float64]): computed value for each date
        """

        return np.array(
            [predict_value(date, times_ranges, intercepts, slopes, start_date, granularity) for date in dates]
        )

    # ------- CHECKERS -------

    def check_params(self):
        """Check if parameters respect their application scope."""
        if not (len(self.intercepts) == len(self.slopes) and len(self.intercepts) == len(self.times_ranges)):
            raise ParameterScopeException(
                "The parameter 'intercepts', 'slopes' and 'times_ranges' are not of the same size"
                + f"Size of 'intercepts':{len(self.intercepts)}, 'slopes': {len(self.slopes)}, "
                + "'times_ranges': {len(self.times_ranges)}."
            )

class PiecewiseTrendError(Exception):
    """Exception raised if start_date and granularity are missing."""

    pass