Module panama.ml.tunable.forecasting

Classes

class BaseTunableForecaster

A base class for tunable machine learning models.

This class defines a set of abstract methods that must be implemented by any concrete subclass.

Attributes

None.

Expand source code

class BaseTunableForecaster(BaseTunableModel):
    def __init__(self):
        super().__init__()

    def _check_y_X(
        self, y: Union[pd.Series, pd.DataFrame], X: Union[pd.Series, pd.DataFrame] = None
    ) -> Tuple[pd.Series, Union[pd.Series, pd.DataFrame]]:
        if not isinstance(y, (pd.Series, pd.DataFrame)):
            raise TypeError("y must be of type pd.Series or pd.DataFrame with time compatible index")
        if not isinstance(y.index, (pd.PeriodIndex, pd.DatetimeIndex)):
            raise TypeError("y index must be of type pd.PeriodIndex or pd.DatetimeIndex with time compatible index")
        if not isinstance(X, (pd.Series, pd.DataFrame)):
            raise TypeError("X must be of type pd.Series or pd.DataFrame with time compatible index")
        if not isinstance(X.index, (pd.PeriodIndex, pd.DatetimeIndex)):
            raise TypeError("X index must be of type pd.PeriodIndex or pd.DatetimeIndex with time compatible index")
        if any(y.index != X.index):
            raise ValueError("Indexes of y and X must be identical")

    @staticmethod
    def _adjust_non_convertible_freqs(s: Union[pd.DataFrame, pd.Series]) -> Union[pd.DataFrame, pd.Series]:
        if s is not None:
            if s.index.freq == "MS":
                s.index = s.index.to_period()
        return s

    @staticmethod
    def _reset_to_original_freq(s: Union[pd.DataFrame, pd.Series]) -> Union[pd.DataFrame, pd.Series]:
        s.index = s.index.to_timestamp()
        return s

    @staticmethod
    def _build_future_from_start_end(
        start: Union[date, datetime, str], end: Union[date, datetime, str], freq: str
    ) -> ForecastingHorizon:
        return ForecastingHorizon(pd.date_range(start, end, freq=freq), is_relative=False)

    @staticmethod
    def _build_future_from_steps(steps: int) -> ForecastingHorizon:
        return ForecastingHorizon(np.arange(1, steps + 1), is_relative=True)

    @staticmethod
    def _build_future_from_exog(X) -> ForecastingHorizon:
        return ForecastingHorizon(X.index, is_relative=False)

    def _set_freq(self, freq):
        self.freq = freq

Ancestors

• BaseTunableModel
• abc.ABC

Subclasses

• TunableExponentialSmoothingForecaster
• TunableProphetForecaster
• TunableSARIMAXForecaster

Inherited members

• BaseTunableModel:
  • fit
  • get_model
  • get_name
  • get_params
  • predict
  • set_params

class TunableExponentialSmoothingForecaster

A base class for tunable machine learning models.

This class defines a set of abstract methods that must be implemented by any concrete subclass.

Attributes

None.

Expand source code

class TunableExponentialSmoothingForecaster(BaseTunableForecaster):
    DEFAULT_SEARCH_DICT = {
        "trend": {"type": "choice", "values": ["add", "mul"]},
        "seasonal": {"type": "choice", "values": ["add", "mul"]},
        "damped_trend": {"type": "choice", "values": [True, False]},
    }

    def __init__(self):
        self.name = "exponential_smoothing"
        self.model = ExponentialSmoothing(random_state=42)

    def fit(self, y: pd.Series, X: Union[pd.Series, pd.DataFrame] = None) -> None:
        self._check_y_X(y, X)
        y = self._adjust_non_convertible_freqs(y)
        X = self._adjust_non_convertible_freqs(X)
        self._set_freq(y.index.freq)
        self.model.fit(y, X)

    def predict(self, future: Union[Tuple, int, pd.Series, pd.DataFrame]) -> Union[pd.DataFrame, pd.Series]:
        if isinstance(future, tuple):
            start, end = future
            fh = self._build_future_from_start_end(start, end, self.freq)
            preds = self.model.predict(fh=fh)
        elif isinstance(future, int):
            fh = self._build_future_from_steps(future)
            preds = self.model.predict(fh=fh)
        elif isinstance(future, (pd.Series, pd.DataFrame)):
            future = self._adjust_non_convertible_freqs(future)
            fh = self._build_future_from_exog(future)
            preds = self.model.predict(fh=fh, X=future)
        else:
            raise ValueError("future must be of types: Tuple, int, pd.Series, pd.DataFrame")
        if isinstance(preds.index, pd.PeriodIndex):
            preds = self._reset_to_original_freq(preds)
        return preds

    def get_params(self, deep: bool = True) -> Dict:
        return self.model.get_params(deep)

    def set_params(self, params: Dict) -> None:
        self.model.set_params(**params)

    def get_fittedvalues(self) -> pd.DataFrame:
        return self.model._fitted_forecaster.fittedvalues

Ancestors

• BaseTunableForecaster
• BaseTunableModel
• abc.ABC

Class variables

var DEFAULT_SEARCH_DICT

Methods

def get_fittedvalues(self) ‑> pandas.core.frame.DataFrame

Inherited members

• BaseTunableForecaster:
  • fit
  • get_model
  • get_name
  • get_params
  • predict
  • set_params

class TunableProphetForecaster (uncertainty_samples: int = 0)

A base class for tunable machine learning models.

This class defines a set of abstract methods that must be implemented by any concrete subclass.

Attributes

None.

Expand source code

class TunableProphetForecaster(BaseTunableForecaster):
    DEFAULT_SEARCH_DICT = {
        "changepoint_prior_scale": {"type": "float", "min": 0.001, "max": 5},
        "seasonality_prior_scale": {"type": "float", "min": 0.01, "max": 35},
        "changepoint_range": {"type": "float", "min": 0.6, "max": 0.95},
        "seasonality_mode": {"type": "choice", "values": ["multiplicative", "additive"]},
        "growth": {"type": "choice", "values": ["linear", "flat"]},
        "yearly_seasonality": {"type": "int", "min": 3, "max": 15},
    }

    def __init__(self, uncertainty_samples: int = 0):
        self.name = "prophet_forecaster"
        self.model = Prophet(uncertainty_samples=uncertainty_samples)

    def fit(self, y: pd.Series, X: Union[pd.Series, pd.DataFrame] = None) -> None:
        self._check_y_X(y, X)
        y = self._adjust_non_convertible_freqs(y)
        X = self._adjust_non_convertible_freqs(X)
        self._set_freq(y.index.freq)
        self.model.fit(y, X)

    def predict(self, future: Union[Tuple, int, pd.Series, pd.DataFrame]) -> Union[pd.DataFrame, pd.Series]:
        if isinstance(future, tuple):
            start, end = future
            fh = self._build_future_from_start_end(start, end, self.freq)
            preds = self.model.predict(fh=fh)
        elif isinstance(future, int):
            fh = self._build_future_from_steps(future)
            preds = self.model.predict(fh=fh)
        elif isinstance(future, (pd.Series, pd.DataFrame)):
            future = self._adjust_non_convertible_freqs(future)
            fh = self._build_future_from_exog(future)
            preds = self.model.predict(fh=fh, X=future)
        else:
            raise ValueError("future must be of types: Tuple, int, pd.Series, pd.DataFrame")
        if isinstance(preds.index, pd.PeriodIndex):
            preds = self._reset_to_original_freq(preds)
        return preds

    def get_params(self, deep: bool = True) -> Dict:
        return self.model.get_params(deep)

    def set_params(self, params: Dict) -> None:
        self.model.set_params(**params)

    def get_fittedvalues(self) -> pd.DataFrame:
        return self.model._forecaster.predict()["yhat"]

    def _get_prophet_fh(self, fh: Union[np.array, ForecastingHorizon]) -> pd.DataFrame:
        """Get a prophet compatible fh, in datetime, even if fh was int."""
        if fh.is_relative:
            fh = int(np.max(fh))
            fh = pd.date_range(self.model.cutoff, periods=fh, freq=self.freq)
        else:
            fh = fh.to_pandas()
        return fh

    def predict_components(self, future: Union[Tuple, int, pd.Series, pd.DataFrame]) -> pd.DataFrame:
        if isinstance(future, tuple):
            start, end = future
            fh = self._build_future_from_start_end(start, end, self.freq)
            fh = self._get_prophet_fh(fh)
        elif isinstance(future, int):
            fh = self._build_future_from_steps(future)
            fh = self._get_prophet_fh(fh)
        elif isinstance(future, (pd.Series, pd.DataFrame)):
            future = self._adjust_non_convertible_freqs(future)
            fh = self._build_future_from_exog(future)
        else:
            raise ValueError("future must be of types: Tuple, int, pd.Series, pd.DataFrame")

        fh = self._get_prophet_fh(fh)
        preds = self.model._forecaster.predict(fh=fh)
        if isinstance(preds.index, pd.PeriodIndex):
            preds = self._adjust_non_convertible_freqs(preds)
        return preds

    def plot_components(self, future: Union[Tuple, int, pd.Series, pd.DataFrame]) -> None:
        preds = self.predict_components(future=future)
        self.model._forecaster.plot_components(preds)

Ancestors

• BaseTunableForecaster
• BaseTunableModel
• abc.ABC

Class variables

var DEFAULT_SEARCH_DICT

Methods

def get_fittedvalues(self) ‑> pandas.core.frame.DataFrame

def plot_components(self, future: Union[Tuple, int, pandas.core.series.Series, pandas.core.frame.DataFrame]) ‑> None

def predict_components(self, future: Union[Tuple, int, pandas.core.series.Series, pandas.core.frame.DataFrame]) ‑> pandas.core.frame.DataFrame

Inherited members

• BaseTunableForecaster:
  • fit
  • get_model
  • get_name
  • get_params
  • predict
  • set_params

class TunableSARIMAXForecaster

A base class for tunable machine learning models.

This class defines a set of abstract methods that must be implemented by any concrete subclass.

Attributes

None.

Expand source code

class TunableSARIMAXForecaster(BaseTunableForecaster):
    DEFAULT_SEARCH_DICT = {
        "trend": {"type": "choice", "values": ["n", "c", "t", "ct"]},
        "order": {"type": "choice", "values": list(itertools.product(range(5), range(2), range(5)))},
        "seasonal_order": {"type": "choice", "values": list(itertools.product(range(5), range(2), range(5), [12]))},
        "use_exact_diffuse": {"type": "choice", "values": [True, False]},
        "simple_differencing": {"type": "choice", "values": [True, False]},
    }

    def __init__(self):
        self.name = "sarimax_forecaster"
        self.model = SARIMAX(random_state=42)

    def fit(self, y: pd.Series, X: Union[pd.Series, pd.DataFrame] = None) -> None:
        self._check_y_X(y, X)
        y = self._adjust_non_convertible_freqs(y)
        X = self._adjust_non_convertible_freqs(X)
        self._set_freq(y.index.freq)
        self.model.fit(y, X)

    def predict(self, future: Union[Tuple, int, pd.Series, pd.DataFrame]) -> Union[pd.DataFrame, pd.Series]:
        if isinstance(future, tuple):
            start, end = future
            fh = self._build_future_from_start_end(start, end, self.freq)
            preds = self.model.predict(fh=fh)
        elif isinstance(future, int):
            fh = self._build_future_from_steps(future)
            preds = self.model.predict(fh=fh)
        elif isinstance(future, (pd.Series, pd.DataFrame)):
            future = self._adjust_non_convertible_freqs(future)
            fh = self._build_future_from_exog(future)
            preds = self.model.predict(fh=fh, X=future)
        else:
            raise ValueError("future must be of types: Tuple, int, pd.Series, pd.DataFrame")
        if isinstance(preds.index, pd.PeriodIndex):
            preds = self._reset_to_original_freq(preds)
        return preds

    def get_params(self, deep: bool = True) -> Dict:
        return self.model.get_params(deep)

    def set_params(self, params: Dict) -> None:
        self.model.set_params(**params)

    def get_fittedvalues(self) -> pd.DataFrame:
        return self.model._fitted_forecaster.fittedvalues

    def plot_components(self):
        return self.model.summary()

Ancestors

• BaseTunableForecaster
• BaseTunableModel
• abc.ABC

Class variables

var DEFAULT_SEARCH_DICT

Methods

def get_fittedvalues(self) ‑> pandas.core.frame.DataFrame

def plot_components(self)

Inherited members

• BaseTunableForecaster:
  • fit
  • get_model
  • get_name
  • get_params
  • predict
  • set_params