preprocessing

skforecast.preprocessing._preprocessing.RollingFeatures

RollingFeatures(
    stats,
    window_sizes,
    min_periods=None,
    features_names=None,
    fillna=None,
    kwargs_stats={"ewm": {"alpha": 0.3}},
)

This class computes rolling features. To avoid data leakage, the last point in the window is excluded from calculations, ('closed': 'left' and 'center': False).

Currently, the following statistics are supported: 'mean', 'std', 'min', 'max', 'sum', 'median', 'ratio_min_max', 'coef_variation', 'ewm'. For 'ewm', the alpha parameter can be set in the kwargs_stats dictionary, default is {'ewm': {'alpha': 0.3}}.

Parameters:

Name	Type	Description	Default
stats	(str, list)	Statistics to compute over the rolling window. Can be a string or a list, and can have repeats. Available statistics are: 'mean', 'std', 'min', 'max', 'sum', 'median', 'ratio_min_max', 'coef_variation', 'ewm'. For 'ewm', the alpha parameter can be set in the kwargs_stats dictionary, default is {'ewm': {'alpha': 0.3}}.	required
window_sizes	(int, list)	Size of the rolling window for each statistic. If an int, all stats share the same window size. If a list, it should have the same length as stats.	required
min_periods	(int, list)	Minimum number of observations in window required to have a value. Same as the min_periods argument of pandas rolling. If None, defaults to window_sizes.	None
features_names	list	Names of the output features. If None, default names will be used in the format 'roll_stat_window_size', for example 'roll_mean_7'.	None
fillna	(str, float)	Fill missing values in transform_batch method. Available methods are: 'mean', 'median', 'ffill', 'bfill', or a float value.	None
kwargs_stats	dict	Dictionary with additional arguments for the statistics. The keys are the statistic names and the values are dictionaries with the arguments for the corresponding statistic. For example, {'ewm': {'alpha': 0.3}}.	{'ewm': {'alpha': 0.3}}

Attributes:

Name	Type	Description
stats	list	Statistics to compute over the rolling window.
n_stats	int	Number of statistics to compute.
window_sizes	list	Size of the rolling window for each statistic.
max_window_size	int	Maximum window size.
min_periods	list	Minimum number of observations in window required to have a value.
features_names	list	Names of the output features.
fillna	(str, float)	Method to fill missing values in transform_batch method.
unique_rolling_windows	dict	Dictionary containing unique rolling window parameters and the corresponding statistics.
kwargs_stats	dict	Dictionary with additional arguments for the statistics.

Methods:

Name	Description
transform_batch	Transform an entire pandas Series using rolling windows and compute the
transform	Transform a numpy array using rolling windows and compute the

Source code in skforecast/preprocessing/_preprocessing.py

def __init__(
    self, 
    stats: str | list[str],
    window_sizes: int | list[int],
    min_periods: int | list[int] | None = None,
    features_names: list[str] | None = None, 
    fillna: str | float | None = None,
    kwargs_stats: dict[str, dict[str, object]] | None = {'ewm': {'alpha': 0.3}}
) -> None:

    self._validate_params(
        stats          = stats,
        window_sizes   = window_sizes,
        min_periods    = min_periods,
        features_names = features_names,
        fillna         = fillna,
        kwargs_stats   = kwargs_stats
    )

    if isinstance(stats, str):
        stats = [stats]
    self.stats = stats
    self.n_stats = len(stats)

    if isinstance(window_sizes, int):
        window_sizes = [window_sizes] * self.n_stats
    self.window_sizes = window_sizes
    self.max_window_size = max(window_sizes)

    if min_periods is None:
        min_periods = self.window_sizes
    elif isinstance(min_periods, int):
        min_periods = [min_periods] * self.n_stats
    self.min_periods = min_periods

    if features_names is None:
        features_names = []
        for stat, window_size in zip(self.stats, self.window_sizes):
            if stat not in kwargs_stats:
                features_names.append(f"roll_{stat}_{window_size}")
            else:
                kwargs_suffix = "_".join([f"{k}_{v}" for k, v in kwargs_stats[stat].items()])
                features_names.append(f"roll_{stat}_{window_size}_{kwargs_suffix}")
    self.features_names = features_names

    self.fillna = fillna
    self.kwargs_stats = kwargs_stats if kwargs_stats is not None else {}

    window_params_list = []
    for i in range(len(self.stats)):
        window_params = (self.window_sizes[i], self.min_periods[i])
        window_params_list.append(window_params)

    # Find unique window parameter combinations
    unique_rolling_windows = {}
    for i, params in enumerate(window_params_list):
        key = f"{params[0]}_{params[1]}"
        if key not in unique_rolling_windows:
            unique_rolling_windows[key] = {
                'params': {
                    'window': params[0], 
                    'min_periods': params[1], 
                    'center': False,
                    'closed': 'left'
                },
                'stats_idx': [], 
                'stats_names': [], 
                'rolling_obj': None
            }
        unique_rolling_windows[key]['stats_idx'].append(i)
        unique_rolling_windows[key]['stats_names'].append(self.features_names[i])

    self.unique_rolling_windows = unique_rolling_windows

Attributes

stats instance-attribute

stats = stats

n_stats instance-attribute

n_stats = len(stats)

window_sizes instance-attribute

window_sizes = window_sizes

max_window_size instance-attribute

max_window_size = max(window_sizes)

min_periods instance-attribute

min_periods = min_periods

features_names instance-attribute

features_names = features_names

fillna instance-attribute

fillna = fillna

kwargs_stats instance-attribute

kwargs_stats = (
    kwargs_stats if kwargs_stats is not None else {}
)

unique_rolling_windows instance-attribute

unique_rolling_windows = unique_rolling_windows

Methods:

_repr_html_

_repr_html_()

HTML representation of the object. The "General Information" section is expanded by default.

Source code in skforecast/preprocessing/_preprocessing.py

def _repr_html_(self) -> str:
    """
    HTML representation of the object.
    The "General Information" section is expanded by default.
    """

    style, unique_id = get_style_repr_html()
    content = f"""
    <div class="container-{unique_id}">
        <p style="font-size: 1.5em; font-weight: bold; margin-block-start: 0.83em; margin-block-end: 0.83em;">{type(self).__name__}</p>
        <details open>
            <summary>General Information</summary>
            <ul>
                <li><strong>Stats:</strong> {self.stats}</li>
                <li><strong>Window size:</strong> {self.window_sizes}</li>
                <li><strong>Maximum window size:</strong> {self.max_window_size}</li>
                <li><strong>Minimum periods:</strong> {self.min_periods}</li>
                <li><strong>Features names:</strong> {self.features_names}</li>
                <li><strong>Fill na strategy:</strong> {self.fillna}</li>
                <li><strong>Kwargs stats:</strong> {self.kwargs_stats}</li>
            </ul>
        </details>
        <p>
            <a href="https://skforecast.org/{__version__}/api/preprocessing.html#skforecast.preprocessing._preprocessing.RollingFeatures">&#128214; <strong>API Reference</strong></a>
            &nbsp;&nbsp;
            <a href="https://skforecast.org/{__version__}/user_guides/window-features-and-custom-features.html">&#128221; <strong>User Guide</strong></a>
        </p>
    </div>
    """

    return style + content

_validate_params

_validate_params(
    stats,
    window_sizes,
    min_periods=None,
    features_names=None,
    fillna=None,
    kwargs_stats=None,
)

Validate the parameters of the RollingFeatures class.

Parameters:

Name	Type	Description	Default
stats	(str, list)	Statistics to compute over the rolling window. Can be a string or a list, and can have repeats. Available statistics are: 'mean', 'std', 'min', 'max', 'sum', 'median', 'ratio_min_max', 'coef_variation', 'ewm'.	required
window_sizes	(int, list)	Size of the rolling window for each statistic. If an int, all stats share the same window size. If a list, it should have the same length as stats.	required
min_periods	(int, list)	Minimum number of observations in window required to have a value. Same as the min_periods argument of pandas rolling. If None, defaults to window_sizes.	None
features_names	list	Names of the output features. If None, default names will be used in the format 'roll_stat_window_size', for example 'roll_mean_7'.	None
fillna	(str, float)	Fill missing values in transform_batch method. Available methods are: 'mean', 'median', 'ffill', 'bfill', or a float value.	None
kwargs_stats	dict	Dictionary with additional arguments for the statistics. The keys are the statistic names and the values are dictionaries with the arguments for the corresponding statistic. For example, {'ewm': {'alpha': 0.3}}.	None

Returns:

Type	Description
None

Source code in skforecast/preprocessing/_preprocessing.py

def _validate_params(
    self, 
    stats: str | list[str], 
    window_sizes: int | list[int],
    min_periods: int | list[int] | None = None,
    features_names: list[str] | None = None, 
    fillna: str | float | None = None,
    kwargs_stats: dict[str, dict[str, object]] | None = None
) -> None:
    """
    Validate the parameters of the RollingFeatures class.

    Parameters
    ----------
    stats : str, list
        Statistics to compute over the rolling window. Can be a `string` or a `list`,
        and can have repeats. Available statistics are: 'mean', 'std', 'min', 'max',
        'sum', 'median', 'ratio_min_max', 'coef_variation', 'ewm'.
    window_sizes : int, list
        Size of the rolling window for each statistic. If an `int`, all stats share 
        the same window size. If a `list`, it should have the same length as stats.
    min_periods : int, list, default None
        Minimum number of observations in window required to have a value. 
        Same as the `min_periods` argument of pandas rolling. If `None`, 
        defaults to `window_sizes`.
    features_names : list, default None
        Names of the output features. If `None`, default names will be used in the 
        format 'roll_stat_window_size', for example 'roll_mean_7'.
    fillna : str, float, default None
        Fill missing values in `transform_batch` method. Available 
        methods are: 'mean', 'median', 'ffill', 'bfill', or a float value.
    kwargs_stats : dict, default None
        Dictionary with additional arguments for the statistics. The keys are the
        statistic names and the values are dictionaries with the arguments for the
        corresponding statistic. For example, {'ewm': {'alpha': 0.3}}.

    Returns
    -------
    None

    """

    # stats
    allowed_stats = [
        'mean', 'std', 'min', 'max', 'sum', 'median', 'ratio_min_max', 
        'coef_variation', 'ewm'
    ]

    if not isinstance(stats, (str, list)):
        raise TypeError(
            f"`stats` must be a string or a list of strings. Got {type(stats)}."
        )        
    if isinstance(stats, str):
        stats = [stats]

    for stat in set(stats):
        if stat not in allowed_stats:
            raise ValueError(
                f"Statistic '{stat}' is not allowed. Allowed stats are: {allowed_stats}."
            )
    n_stats = len(stats)

    # window_sizes
    if not isinstance(window_sizes, (int, list)):
        raise TypeError(
            f"`window_sizes` must be an int or a list of ints. Got {type(window_sizes)}."
        )

    if isinstance(window_sizes, list):
        n_window_sizes = len(window_sizes)
        if n_window_sizes != n_stats:
            raise ValueError(
                f"Length of `window_sizes` list ({n_window_sizes}) "
                f"must match length of `stats` list ({n_stats})."
            )

    # Check duplicates (stats, window_sizes)
    if isinstance(window_sizes, int):
        window_sizes = [window_sizes] * n_stats
    if len(set(zip(stats, window_sizes))) != n_stats:
        raise ValueError(
            f"Duplicate (stat, window_size) pairs are not allowed.\n"
            f"    `stats`        : {stats}\n"
            f"    `window_sizes` : {window_sizes}"
        )

    # min_periods
    if not isinstance(min_periods, (int, list, type(None))):
        raise TypeError(
            f"`min_periods` must be an int, list of ints, or None. Got {type(min_periods)}."
        )

    if min_periods is not None:
        if isinstance(min_periods, int):
            min_periods = [min_periods] * n_stats
        elif isinstance(min_periods, list):
            n_min_periods = len(min_periods)
            if n_min_periods != n_stats:
                raise ValueError(
                    f"Length of `min_periods` list ({n_min_periods}) "
                    f"must match length of `stats` list ({n_stats})."
                )

        for i, min_period in enumerate(min_periods):
            if min_period > window_sizes[i]:
                raise ValueError(
                    "Each `min_period` must be less than or equal to its "
                    "corresponding `window_size`."
                )

    # features_names
    if not isinstance(features_names, (list, type(None))):
        raise TypeError(
            f"`features_names` must be a list of strings or None. Got {type(features_names)}."
        )

    if isinstance(features_names, list):
        n_features_names = len(features_names)
        if n_features_names != n_stats:
            raise ValueError(
                f"Length of `features_names` list ({n_features_names}) "
                f"must match length of `stats` list ({n_stats})."
            )

    # fillna
    if fillna is not None:
        if not isinstance(fillna, (int, float, str)):
            raise TypeError(
                f"`fillna` must be a float, string, or None. Got {type(fillna)}."
            )

        if isinstance(fillna, str):
            allowed_fill_strategy = ['mean', 'median', 'ffill', 'bfill']
            if fillna not in allowed_fill_strategy:
                raise ValueError(
                    f"'{fillna}' is not allowed. Allowed `fillna` "
                    f"values are: {allowed_fill_strategy} or a float value."
                )

    # kwargs_stats
    allowed_kwargs_stats = ['ewm']
    if kwargs_stats is not None:
        if not isinstance(kwargs_stats, dict):
            raise TypeError(
                f"`kwargs_stats` must be a dictionary or None. Got {type(kwargs_stats)}."
            )

        for stat in kwargs_stats.keys():
            if stat not in allowed_kwargs_stats:
                raise ValueError(
                    f"Invalid statistic '{stat}' found in `kwargs_stats`. "
                    f"Allowed statistics with additional arguments are: "
                    f"{allowed_kwargs_stats}. Please ensure all keys in "
                    f"`kwargs_stats` are among the allowed statistics."
                )

_apply_stat_pandas

_apply_stat_pandas(rolling_obj, stat)

Apply the specified statistic to a pandas rolling object.

Parameters:

Name	Type	Description	Default
rolling_obj	pandas Rolling	Rolling object to apply the statistic.	required
stat	str	Statistic to compute.	required

Returns:

Name	Type	Description
stat_series	pandas Series	Series with the computed statistic.

Source code in skforecast/preprocessing/_preprocessing.py

def _apply_stat_pandas(
    self, 
    rolling_obj: pd.core.window.rolling.Rolling, 
    stat: str
) -> pd.Series:
    """
    Apply the specified statistic to a pandas rolling object.

    Parameters
    ----------
    rolling_obj : pandas Rolling
        Rolling object to apply the statistic.
    stat : str
        Statistic to compute.

    Returns
    -------
    stat_series : pandas Series
        Series with the computed statistic.

    """

    if stat == 'mean':
        return rolling_obj.mean()
    elif stat == 'std':
        return rolling_obj.std()
    elif stat == 'min':
        return rolling_obj.min()
    elif stat == 'max':
        return rolling_obj.max()
    elif stat == 'sum':
        return rolling_obj.sum()
    elif stat == 'median':
        return rolling_obj.median()
    elif stat == 'ratio_min_max':
        return rolling_obj.min() / rolling_obj.max()
    elif stat == 'coef_variation':
        return rolling_obj.std() / rolling_obj.mean()
    elif stat == 'ewm':
        kwargs = self.kwargs_stats.get(stat, {})
        return rolling_obj.apply(lambda x: _ewm_jit(x, **kwargs), raw=True)
    else:
        raise ValueError(f"Statistic '{stat}' is not implemented.")

transform_batch

transform_batch(X)

Transform an entire pandas Series using rolling windows and compute the specified statistics.

Parameters:

Name	Type	Description	Default
X	pandas Series	The input data series to transform.	required

Returns:

Name	Type	Description
rolling_features	pandas DataFrame	A DataFrame containing the rolling features.

Source code in skforecast/preprocessing/_preprocessing.py

def transform_batch(
    self, 
    X: pd.Series
) -> pd.DataFrame:
    """
    Transform an entire pandas Series using rolling windows and compute the 
    specified statistics.

    Parameters
    ----------
    X : pandas Series
        The input data series to transform.

    Returns
    -------
    rolling_features : pandas DataFrame
        A DataFrame containing the rolling features.

    """

    for k in self.unique_rolling_windows.keys():
        rolling_obj = X.rolling(**self.unique_rolling_windows[k]['params'])
        self.unique_rolling_windows[k]['rolling_obj'] = rolling_obj

    rolling_features = []
    for i, stat in enumerate(self.stats):
        window_size = self.window_sizes[i]
        min_periods = self.min_periods[i]

        key = f"{window_size}_{min_periods}"
        rolling_obj = self.unique_rolling_windows[key]['rolling_obj']

        stat_series = self._apply_stat_pandas(rolling_obj=rolling_obj, stat=stat)            
        rolling_features.append(stat_series)

    rolling_features = pd.concat(rolling_features, axis=1)
    rolling_features.columns = self.features_names
    rolling_features = rolling_features.iloc[self.max_window_size:]

    if self.fillna is not None:
        if self.fillna == 'mean':
            rolling_features = rolling_features.fillna(rolling_features.mean())
        elif self.fillna == 'median':
            rolling_features = rolling_features.fillna(rolling_features.median())
        elif self.fillna == 'ffill':
            rolling_features = rolling_features.ffill()
        elif self.fillna == 'bfill':
            rolling_features = rolling_features.bfill()
        else:
            rolling_features = rolling_features.fillna(self.fillna)

    return rolling_features

_apply_stat_numpy_jit

_apply_stat_numpy_jit(X_window, stat)

Apply the specified statistic to a numpy array using Numba JIT.

Parameters:

Name	Type	Description	Default
X_window	numpy array	Array with the rolling window.	required
stat	str	Statistic to compute.	required

Returns:

Name	Type	Description
stat_value	float	Value of the computed statistic.

Source code in skforecast/preprocessing/_preprocessing.py

def _apply_stat_numpy_jit(
    self, 
    X_window: np.ndarray, 
    stat: str
) -> float:
    """
    Apply the specified statistic to a numpy array using Numba JIT.

    Parameters
    ----------
    X_window : numpy array
        Array with the rolling window.
    stat : str
        Statistic to compute.

    Returns
    -------
    stat_value : float
        Value of the computed statistic.

    """

    if stat == 'mean':
        return _np_mean_jit(X_window)
    elif stat == 'std':
        return _np_std_jit(X_window)
    elif stat == 'min':
        return _np_min_jit(X_window)
    elif stat == 'max':
        return _np_max_jit(X_window)
    elif stat == 'sum':
        return _np_sum_jit(X_window)
    elif stat == 'median':
        return _np_median_jit(X_window)
    elif stat == 'ratio_min_max':
        return _np_min_max_ratio_jit(X_window)
    elif stat == 'coef_variation':
        return _np_cv_jit(X_window)
    elif stat == 'ewm':
        kwargs = self.kwargs_stats.get(stat, {})
        return _ewm_jit(X_window, **kwargs)
    else:
        raise ValueError(f"Statistic '{stat}' is not implemented.")

transform

transform(X)

Transform a numpy array using rolling windows and compute the specified statistics. The returned array will have the shape (X.shape[1] if exists, n_stats). For example, if X is a flat array, the output will have shape (n_stats,). If X is a 2D array, the output will have shape (X.shape[1], n_stats).

Parameters:

Name	Type	Description	Default
X	numpy ndarray	The input data array to transform.	required

Returns:

Name	Type	Description
rolling_features	numpy ndarray	An array containing the computed statistics.

Source code in skforecast/preprocessing/_preprocessing.py

def transform(
    self, 
    X: np.ndarray
) -> np.ndarray:
    """
    Transform a numpy array using rolling windows and compute the 
    specified statistics. The returned array will have the shape 
    (X.shape[1] if exists, n_stats). For example, if X is a flat
    array, the output will have shape (n_stats,). If X is a 2D array,
    the output will have shape (X.shape[1], n_stats).

    Parameters
    ----------
    X : numpy ndarray
        The input data array to transform.

    Returns
    -------
    rolling_features : numpy ndarray
        An array containing the computed statistics.

    """

    array_ndim = X.ndim
    if array_ndim == 1:
        X = X[:, np.newaxis]

    vectorizable_stats = {'mean', 'std', 'min', 'max', 'sum', 'median'}
    has_vectorizable = bool(set(self.stats) & vectorizable_stats)

    rolling_features = np.full(
        shape=(X.shape[1], self.n_stats), fill_value=np.nan, dtype=float
    )

    # Compute vectorized stats if any are requested
    if has_vectorizable:
        self._transform_vectorized(X, rolling_features)

    # Compute non-vectorizable stats
    for i in range(X.shape[1]):
        for j, stat in enumerate(self.stats):
            if stat in vectorizable_stats:
                continue
            X_window = X[-self.window_sizes[j]:, i]
            X_window = X_window[~np.isnan(X_window)]
            if len(X_window) > 0: 
                rolling_features[i, j] = self._apply_stat_numpy_jit(X_window, stat)
            else:
                rolling_features[i, j] = np.nan

    if array_ndim == 1:
        rolling_features = rolling_features.ravel()

    return rolling_features

_transform_vectorized

_transform_vectorized(X, rolling_features)

Vectorized transform using NumPy axis operations for vectorizable stats. Modifies rolling_features in place for the vectorizable statistics. This method is specifically designed to speed up the computation of statistics in predict_bootstrap method of forecasters.

Parameters:

Name	Type	Description	Default
X	numpy ndarray	Input array of shape (window_length, n_samples).	required
rolling_features	numpy ndarray	Output array of shape (n_samples, n_stats) to fill in.	required

Returns:

Type	Description
None	Modifies rolling_features in place. Some statistics do not follow the numpy behavior exactly: - For 'std', if the window has only one non-NaN value, the result is 0.0 instead of NaN (to match _np_std_jit behavior). - For 'sum', if the window has all NaN values, the result is NaN instead of 0.0 (to match _np_sum_jit behavior).

Source code in skforecast/preprocessing/_preprocessing.py

def _transform_vectorized(
    self,
    X: np.ndarray,
    rolling_features: np.ndarray
) -> np.ndarray:
    """
    Vectorized transform using NumPy axis operations for vectorizable stats.
    Modifies rolling_features in place for the vectorizable statistics.
    This method is specifically designed to speed up the computation of
    statistics in `predict_bootstrap` method of forecasters.

    Parameters
    ----------
    X : numpy ndarray
        Input array of shape (window_length, n_samples).
    rolling_features : numpy ndarray
        Output array of shape (n_samples, n_stats) to fill in.

    Returns
    -------
    None
        Modifies rolling_features in place.
        Some statistics do not follow the numpy behavior exactly:
        - For 'std', if the window has only one non-NaN value, the result is 0.0
          instead of NaN (to match _np_std_jit behavior).
        - For 'sum', if the window has all NaN values, the result is NaN
          instead of 0.0 (to match _np_sum_jit behavior).

    """
    vectorizable_stats = {'mean', 'std', 'min', 'max', 'sum', 'median'}
    for j, stat in enumerate(self.stats):
        if stat not in vectorizable_stats:
            continue
        window = X[-self.window_sizes[j]:, :]
        with warnings.catch_warnings():
            warnings.filterwarnings('ignore', message='Mean of empty slice')
            warnings.filterwarnings('ignore', message='Degrees of freedom <= 0 for slice')
            warnings.filterwarnings('ignore', message='All-NaN slice encountered')
            if stat == 'mean':
                rolling_features[:, j] = np.nanmean(window, axis=0)
            elif stat == 'std':
                result = np.nanstd(window, axis=0, ddof=1)
                # Note: np.nanstd returns nan for single non-NaN values (ddof=1),
                # but it is replaced by 0.0 to match the behavior of the non-vectorized
                # _np_std_jit function
                n_valid = np.sum(~np.isnan(window), axis=0)
                result[n_valid == 1] = 0.0
                rolling_features[:, j] = result
            elif stat == 'min':
                rolling_features[:, j] = np.nanmin(window, axis=0)
            elif stat == 'max':
                rolling_features[:, j] = np.nanmax(window, axis=0)
            elif stat == 'sum':
                result = np.nansum(window, axis=0, dtype=float)
                # Note: np.nansum returns 0 for all-NaN slices, but it is replaced by NaN
                # to match the behavior of the non-vectorized _np_sum_jit function
                all_nan_mask = np.all(np.isnan(window), axis=0)
                result[all_nan_mask] = np.nan
                rolling_features[:, j] = result
            elif stat == 'median':
                rolling_features[:, j] = np.nanmedian(window, axis=0)

skforecast.preprocessing._preprocessing.RollingFeaturesClassification

RollingFeaturesClassification(
    stats,
    window_sizes,
    min_periods=None,
    features_names=None,
    fillna=None,
)

This class computes rolling features for classification problems. To avoid data leakage, the last point in the window is excluded from calculations, ('closed': 'left' and 'center': False).

Currently, the following statistics are supported: 'proportion', 'mode', 'entropy', 'n_changes', 'n_unique'.

Parameters:

Name	Type	Description	Default
stats	(str, list)	Statistics to compute over the rolling window. Can be a string or a list, and can have repeats. Available statistics are: 'proportion', 'mode', 'entropy', 'n_changes', 'n_unique'.	required
window_sizes	(int, list)	Size of the rolling window for each statistic. If an int, all stats share the same window size. If a list, it should have the same length as stats.	required
min_periods	(int, list)	Minimum number of observations in window required to have a value. Same as the min_periods argument of pandas rolling. If None, defaults to window_sizes.	None
features_names	list	Names of the output features. If None, default names will be used in the format 'roll_stat_window_size', for example 'roll_mode_7'. For 'proportion', class-specific names are appended, e.g., 'roll_proportion_7_class_0'.	None
fillna	(str, float)	Fill missing values in transform_batch method. Available methods are: 'mean', 'median', 'ffill', 'bfill', or a float value.	None

Attributes:

Name	Type	Description
stats	list	Statistics to compute over the rolling window.
n_stats	int	Number of statistics to compute.
window_sizes	list	Size of the rolling window for each statistic.
max_window_size	int	Maximum window size.
min_periods	list	Minimum number of observations in window required to have a value.
classes	list	Unique classes found in the data. Inferred during transform_batch.
features_names	list	Names of the output features.
fillna	(str, float)	Method to fill missing values in transform_batch method.
unique_rolling_windows	dict	Dictionary containing unique rolling window parameters and the corresponding statistics.

Methods:

Name	Description
transform_batch	Transform an entire pandas Series using rolling windows and compute the
transform	Transform a numpy array using rolling windows and compute the

Source code in skforecast/preprocessing/_preprocessing.py

def __init__(
    self, 
    stats: str | list[str],
    window_sizes: int | list[int],
    min_periods: int | list[int] | None = None,
    features_names: list[str] | None = None, 
    fillna: str | float | None = None
) -> None:

    self._validate_params(
        stats          = stats,
        window_sizes   = window_sizes,
        min_periods    = min_periods,
        features_names = features_names,
        fillna         = fillna
    )

    if isinstance(stats, str):
        stats = [stats]
    self.stats = stats
    self.n_stats = len(stats)

    if isinstance(window_sizes, int):
        window_sizes = [window_sizes] * self.n_stats
    self.window_sizes = window_sizes
    self.max_window_size = max(window_sizes)

    if min_periods is None:
        min_periods = self.window_sizes
    elif isinstance(min_periods, int):
        min_periods = [min_periods] * self.n_stats
    self.min_periods = min_periods

    self.classes = None
    if features_names is None:
        features_names = []
        for stat, window_size in zip(self.stats, self.window_sizes):
            features_names.append(f"roll_{stat}_{window_size}")
    self.features_names = features_names

    self.fillna = fillna

    window_params_list = []
    for i in range(len(self.stats)):
        window_params = (self.window_sizes[i], self.min_periods[i])
        window_params_list.append(window_params)

    # Find unique window parameter combinations
    unique_rolling_windows = {}
    for i, params in enumerate(window_params_list):
        key = f"{params[0]}_{params[1]}"
        if key not in unique_rolling_windows:
            unique_rolling_windows[key] = {
                'params': {
                    'window': params[0], 
                    'min_periods': params[1], 
                    'center': False,
                    'closed': 'left'
                },
                'stats_idx': [], 
                'stats_names': [], 
                'rolling_obj': None
            }
        unique_rolling_windows[key]['stats_idx'].append(i)
        unique_rolling_windows[key]['stats_names'].append(self.features_names[i])

    self.unique_rolling_windows = unique_rolling_windows

Attributes

stats instance-attribute

stats = stats

n_stats instance-attribute

n_stats = len(stats)

window_sizes instance-attribute

window_sizes = window_sizes

max_window_size instance-attribute

max_window_size = max(window_sizes)

min_periods instance-attribute

min_periods = min_periods

classes instance-attribute

classes = None

features_names instance-attribute

features_names = features_names

fillna instance-attribute

fillna = fillna

unique_rolling_windows instance-attribute

unique_rolling_windows = unique_rolling_windows

Methods:

_repr_html_

_repr_html_()

HTML representation of the object. The "General Information" section is expanded by default.

Source code in skforecast/preprocessing/_preprocessing.py

def _repr_html_(self) -> str:
    """
    HTML representation of the object.
    The "General Information" section is expanded by default.
    """

    style, unique_id = get_style_repr_html()
    content = f"""
    <div class="container-{unique_id}">
        <p style="font-size: 1.5em; font-weight: bold; margin-block-start: 0.83em; margin-block-end: 0.83em;">{type(self).__name__}</p>
        <details open>
            <summary>General Information</summary>
            <ul>
                <li><strong>Stats:</strong> {self.stats}</li>
                <li><strong>Window size:</strong> {self.window_sizes}</li>
                <li><strong>Maximum window size:</strong> {self.max_window_size}</li>
                <li><strong>Minimum periods:</strong> {self.min_periods}</li>
                <li><strong>Classes:</strong> {self.classes}</li>
                <li><strong>Features names:</strong> {self.features_names}</li>
                <li><strong>Fill na strategy:</strong> {self.fillna}</li>
            </ul>
        </details>
        <p>
            <a href="https://skforecast.org/{__version__}/api/preprocessing.html#skforecast.preprocessing._preprocessing.RollingFeaturesClassification">&#128214; <strong>API Reference</strong></a>
            &nbsp;&nbsp;
            <a href="https://skforecast.org/{__version__}/user_guides/autoregressive-classification-forecasting.html">&#128221; <strong>User Guide</strong></a>
        </p>
    </div>
    """

    return style + content

_validate_params

_validate_params(
    stats,
    window_sizes,
    min_periods=None,
    features_names=None,
    fillna=None,
)

Validate the parameters of the RollingFeaturesClassification class.

Parameters:

Name	Type	Description	Default
stats	(str, list)	Statistics to compute over the rolling window. Can be a string or a list, and can have repeats. Available statistics are: 'proportion', 'mode', 'entropy', 'n_changes', 'n_unique'.	required
window_sizes	(int, list)	Size of the rolling window for each statistic. If an int, all stats share the same window size. If a list, it should have the same length as stats.	required
min_periods	(int, list)	Minimum number of observations in window required to have a value. Same as the min_periods argument of pandas rolling. If None, defaults to window_sizes.	None
features_names	list	Names of the output features. If None, default names will be used in the format 'roll_stat_window_size', for example 'roll_mode_7'. For 'proportion', class-specific names are appended, e.g., 'roll_proportion_7_class_0'.	None
fillna	(str, float)	Fill missing values in transform_batch method. Available methods are: 'mean', 'median', 'ffill', 'bfill', or a float value.	None

Returns:

Type	Description
None

Source code in skforecast/preprocessing/_preprocessing.py

def _validate_params(
    self, 
    stats: str | list[str], 
    window_sizes: int | list[int],
    min_periods: int | list[int] | None = None,
    features_names: list[str] | None = None, 
    fillna: str | float | None = None
) -> None:
    """
    Validate the parameters of the RollingFeaturesClassification class.

    Parameters
    ----------
    stats : str, list
        Statistics to compute over the rolling window. Can be a `string` or a `list`,
        and can have repeats. Available statistics are: 'proportion', 'mode', 
        'entropy', 'n_changes', 'n_unique'.
    window_sizes : int, list
        Size of the rolling window for each statistic. If an `int`, all stats share 
        the same window size. If a `list`, it should have the same length as `stats`.
    min_periods : int, list, default None
        Minimum number of observations in window required to have a value. 
        Same as the `min_periods` argument of pandas rolling. If `None`, 
        defaults to `window_sizes`.
    features_names : list, default None
        Names of the output features. If `None`, default names will be used in the 
        format 'roll_stat_window_size', for example 'roll_mode_7'. For 'proportion',
        class-specific names are appended, e.g., 'roll_proportion_7_class_0'.
    fillna : str, float, default None
        Fill missing values in `transform_batch` method. Available 
        methods are: 'mean', 'median', 'ffill', 'bfill', or a float value.

    Returns
    -------
    None

    """

    # stats
    allowed_stats = [
        'proportion', 'mode', 'entropy', 'n_changes', 'n_unique'
    ]

    if not isinstance(stats, (str, list)):
        raise TypeError(
            f"`stats` must be a string or a list of strings. Got {type(stats)}."
        )        
    if isinstance(stats, str):
        stats = [stats]

    for stat in set(stats):
        if stat not in allowed_stats:
            raise ValueError(
                f"Statistic '{stat}' is not allowed. Allowed stats are: {allowed_stats}."
            )
    n_stats = len(stats)

    # window_sizes
    if not isinstance(window_sizes, (int, list)):
        raise TypeError(
            f"`window_sizes` must be an int or a list of ints. Got {type(window_sizes)}."
        )

    if isinstance(window_sizes, list):
        n_window_sizes = len(window_sizes)
        if n_window_sizes != n_stats:
            raise ValueError(
                f"Length of `window_sizes` list ({n_window_sizes}) "
                f"must match length of `stats` list ({n_stats})."
            )

    # Check duplicates (stats, window_sizes)
    if isinstance(window_sizes, int):
        window_sizes = [window_sizes] * n_stats
    if len(set(zip(stats, window_sizes))) != n_stats:
        raise ValueError(
            f"Duplicate (stat, window_size) pairs are not allowed.\n"
            f"    `stats`        : {stats}\n"
            f"    `window_sizes` : {window_sizes}"
        )

    # min_periods
    if not isinstance(min_periods, (int, list, type(None))):
        raise TypeError(
            f"`min_periods` must be an int, list of ints, or None. Got {type(min_periods)}."
        )

    if min_periods is not None:
        if isinstance(min_periods, int):
            min_periods = [min_periods] * n_stats
        elif isinstance(min_periods, list):
            n_min_periods = len(min_periods)
            if n_min_periods != n_stats:
                raise ValueError(
                    f"Length of `min_periods` list ({n_min_periods}) "
                    f"must match length of `stats` list ({n_stats})."
                )

        for i, min_period in enumerate(min_periods):
            if min_period > window_sizes[i]:
                raise ValueError(
                    "Each `min_period` must be less than or equal to its "
                    "corresponding `window_size`."
                )

    # features_names
    if not isinstance(features_names, (list, type(None))):
        raise TypeError(
            f"`features_names` must be a list of strings or None. Got {type(features_names)}."
        )

    if isinstance(features_names, list):
        n_features_names = len(features_names)
        if n_features_names != n_stats:
            raise ValueError(
                f"Length of `features_names` list ({n_features_names}) "
                f"must match length of `stats` list ({n_stats})."
            )

    # TODO: Not used as ForecasterRecursiveClassifier doesn't allow NaNs. Check
    # when creating ForecasterRecursiveMultiSeriesClassifier
    # fillna
    if fillna is not None:
        if not isinstance(fillna, (int, float, str)):
            raise TypeError(
                f"`fillna` must be a float, string, or None. Got {type(fillna)}."
            )

        if isinstance(fillna, str):
            allowed_fill_strategy = ['mean', 'median', 'ffill', 'bfill']
            if fillna not in allowed_fill_strategy:
                raise ValueError(
                    f"'{fillna}' is not allowed. Allowed `fillna` "
                    f"values are: {allowed_fill_strategy} or a float value."
                )

_apply_stat_pandas

_apply_stat_pandas(X, rolling_obj, stat)

Apply the specified statistic to a pandas rolling object.

Parameters:

Name	Type	Description	Default
rolling_obj	pandas Rolling	Rolling object to apply the statistic.	required
stat	str	Statistic to compute.	required

Returns:

Name	Type	Description
stat_series	pandas Series	Series with the computed statistic.

Source code in skforecast/preprocessing/_preprocessing.py

def _apply_stat_pandas(
    self, 
    X: pd.Series,
    rolling_obj: pd.core.window.rolling.Rolling, 
    stat: str
) -> pd.Series:
    """
    Apply the specified statistic to a pandas rolling object.

    Parameters
    ----------
    rolling_obj : pandas Rolling
        Rolling object to apply the statistic.
    stat : str
        Statistic to compute.

    Returns
    -------
    stat_series : pandas Series
        Series with the computed statistic.

    """

    if stat == 'proportion':
        rolling_params = {
            'window': rolling_obj.window, 
            'min_periods': rolling_obj.min_periods, 
            'center': rolling_obj.center,
            'closed': rolling_obj.closed
        }
        dummies = pd.get_dummies(X, prefix='class')
        proportions = dummies.rolling(**rolling_params).sum() / rolling_obj.window

        return proportions

    elif stat == 'mode':
        return rolling_obj.apply(lambda x: scipy_mode(x)[0], raw=True)
    elif stat == 'entropy':
        return rolling_obj.apply(_entropy, raw=True)
    elif stat == 'n_changes':
        return rolling_obj.apply(_n_changes_jit, raw=True)
    elif stat == 'n_unique':
        return rolling_obj.apply(_n_unique_jit, raw=True)
    else:
        raise ValueError(f"Statistic '{stat}' is not implemented.")

transform_batch

transform_batch(X)

Transform an entire pandas Series using rolling windows and compute the specified statistics.

Parameters:

Name	Type	Description	Default
X	pandas Series	The input data series to transform.	required

Returns:

Name	Type	Description
rolling_features	pandas DataFrame	A DataFrame containing the rolling features.

Source code in skforecast/preprocessing/_preprocessing.py

def transform_batch(
    self, 
    X: pd.Series
) -> pd.DataFrame:
    """
    Transform an entire pandas Series using rolling windows and compute the 
    specified statistics.

    Parameters
    ----------
    X : pandas Series
        The input data series to transform.

    Returns
    -------
    rolling_features : pandas DataFrame
        A DataFrame containing the rolling features.

    """

    if self.classes is None:
        self.classes = list(np.sort(X.unique()))

        features_names = []
        for stat, feature_name in zip(self.stats, self.features_names):
            if stat != 'proportion':
                features_names.append(feature_name)
            else:
                for cls in self.classes:
                    feature_name_class = f"{feature_name}_class_{cls}"
                    features_names.append(feature_name_class)

        self.features_names = features_names

    for k in self.unique_rolling_windows.keys():
        rolling_obj = X.rolling(**self.unique_rolling_windows[k]['params'])
        self.unique_rolling_windows[k]['rolling_obj'] = rolling_obj

    rolling_features = []
    for i, stat in enumerate(self.stats):
        window_size = self.window_sizes[i]
        min_periods = self.min_periods[i]

        key = f"{window_size}_{min_periods}"
        rolling_obj = self.unique_rolling_windows[key]['rolling_obj']

        stat_series = self._apply_stat_pandas(X=X, rolling_obj=rolling_obj, stat=stat)     
        rolling_features.append(stat_series)

    rolling_features = pd.concat(rolling_features, axis=1)
    rolling_features.columns = self.features_names
    rolling_features = rolling_features.iloc[self.max_window_size:]

    if self.fillna is not None:
        if self.fillna == 'mean':
            rolling_features = rolling_features.fillna(rolling_features.mean())
        elif self.fillna == 'median':
            rolling_features = rolling_features.fillna(rolling_features.median())
        elif self.fillna == 'ffill':
            rolling_features = rolling_features.ffill()
        elif self.fillna == 'bfill':
            rolling_features = rolling_features.bfill()
        else:
            rolling_features = rolling_features.fillna(self.fillna)

    return rolling_features

_apply_stat_numpy_jit

_apply_stat_numpy_jit(X_window, stat)

Apply the specified statistic to a numpy array using Numba JIT.

Parameters:

Name	Type	Description	Default
X_window	numpy array	Array with the rolling window.	required
stat	str	Statistic to compute.	required

Returns:

Name	Type	Description
stat_value	float	Value of the computed statistic.

Source code in skforecast/preprocessing/_preprocessing.py

def _apply_stat_numpy_jit(
    self, 
    X_window: np.ndarray, 
    stat: str
) -> float:
    """
    Apply the specified statistic to a numpy array using Numba JIT.

    Parameters
    ----------
    X_window : numpy array
        Array with the rolling window.
    stat : str
        Statistic to compute.

    Returns
    -------
    stat_value : float
        Value of the computed statistic.

    """

    if stat == 'proportion':
        # Calculate proportions for each class
        proportions = np.zeros(len(self.classes))
        len_window = len(X_window)
        for i, cls in enumerate(self.classes):
            proportions[i] = np.sum(X_window == cls) / len_window
        return proportions

    elif stat == 'mode':
        return scipy_mode(X_window)[0]
    elif stat == 'entropy':
        return _entropy(X_window)
    elif stat == 'n_changes':
        return _n_changes_jit(X_window)
    elif stat == 'n_unique':
        return _n_unique_jit(X_window)
    else:
        raise ValueError(f"Statistic '{stat}' is not implemented.")

transform

transform(X)

Transform a numpy array using rolling windows and compute the specified statistics. The returned array will have the shape (X.shape[1] if exists, n_stats). For example, if X is a flat array, the output will have shape (n_stats,). If X is a 2D array, the output will have shape (X.shape[1], n_stats).

Parameters:

Name	Type	Description	Default
X	numpy ndarray	The input data array to transform.	required

Returns:

Name	Type	Description
rolling_features	numpy ndarray	An array containing the computed statistics.

Source code in skforecast/preprocessing/_preprocessing.py

def transform(
    self, 
    X: np.ndarray
) -> np.ndarray:
    """
    Transform a numpy array using rolling windows and compute the 
    specified statistics. The returned array will have the shape 
    (X.shape[1] if exists, n_stats). For example, if X is a flat
    array, the output will have shape (n_stats,). If X is a 2D array,
    the output will have shape (X.shape[1], n_stats).

    Parameters
    ----------
    X : numpy ndarray
        The input data array to transform.

    Returns
    -------
    rolling_features : numpy ndarray
        An array containing the computed statistics.

    """

    if self.classes is None:
        raise ValueError(
            "Classes must be specified before calling transform. "
            "Call `transform_batch` first to infer classes from data."
        )

    array_ndim = X.ndim
    if array_ndim == 1:
        X = X[:, np.newaxis]

    # TODO: If more than one columns 2d Array, maybe the classes doesn't come
    # from the same column. Col 1 has classes [0, 1], col 2 has classes [3, 4].
    n_classes = len(self.classes)
    n_output_features = 0
    for stat in self.stats:
        if stat == 'proportion':
            n_output_features += n_classes
        else:
            n_output_features += 1

    rolling_features = np.full(
        shape=(X.shape[1], n_output_features), fill_value=np.nan, dtype=float
    )
    for i in range(X.shape[1]):
        feature_idx = 0
        for j, stat in enumerate(self.stats):
            X_window = X[-self.window_sizes[j]:, i]
            X_window = X_window[~np.isnan(X_window)]

            if len(X_window) >= 0:
                result = self._apply_stat_numpy_jit(X_window, stat)

                if stat == 'proportion':
                    # Result is an array with one value per class
                    rolling_features[i, feature_idx:feature_idx + n_classes] = result
                    feature_idx += n_classes
                else:
                    # Result is a single value
                    rolling_features[i, feature_idx] = result
                    feature_idx += 1
            else:
                if stat == 'proportion':
                    rolling_features[i, feature_idx:feature_idx + n_classes] = np.nan
                    feature_idx += n_classes
                else:
                    rolling_features[i, feature_idx] = np.nan
                    feature_idx += 1

    if array_ndim == 1:
        rolling_features = rolling_features.ravel()

    return rolling_features

skforecast.preprocessing._calendar.CalendarFeatures

CalendarFeatures(
    features=None,
    features_to_encode=None,
    encoding="cyclical",
    max_values=None,
    spline_kwargs=None,
    keep_original_columns=True,
    tol=1e-12,
)

Bases: BaseEstimator, TransformerMixin

A transformer for extracting datetime features from the DateTime index of a pandas DataFrame or Series, or from a pandas DatetimeIndex directly. It can also apply encoding to the extracted features.

The allowed features to extract are: 'year', 'month', 'week', 'day_of_week', 'day_of_month', 'day_of_year', 'weekend', 'hour', 'minute', 'second', and 'quarter'. If not specified, all of these features are extracted.

Parameters:

Name	Type	Description	Default
features	list	List of calendar features (strings) to extract from the index. When None, the following features are extracted: 'year', 'month', 'week', 'day_of_week', 'day_of_month', 'day_of_year', 'weekend', 'hour', 'minute', 'second', 'quarter'.	None
features_to_encode	list	List of calendar features (strings) to encode. When None, all extracted features are encoded. If a feature is not in features, a ValueError is raised at fit/transform time. If the explicit list contains features that cannot be encoded with the chosen encoding (e.g. 'year' or 'weekend', which are never encodable), an IgnoredArgumentWarning is issued and those features are kept as raw integers.	None
encoding	str	Encoding method for the extracted features. Options are None, 'cyclical', 'onehot' or 'spline'. Features that cannot be encoded under the chosen mode are kept as raw integers. By default, 'year' and 'weekend' are never encoded. 'onehot' excludes them via the known-category set, while 'cyclical' and 'spline' exclude them via max_values.	'cyclical'
max_values	dict	Dictionary of maximum values for the cyclical and spline encoding. User-provided values are merged with the defaults: keys passed by the user override the corresponding default, and missing keys fall back to the defaults {'month': 12, 'week': 53, 'day_of_week': 7, 'day_of_month': 31, 'day_of_year': 366, 'hour': 24, 'minute': 60, 'second': 60, 'quarter': 4}. For example, passing max_values={'month': 6} overrides only month; the other features keep their defaults. Features that are not in the defaults (e.g. 'year', 'weekend') are left as raw integers.	None
spline_kwargs	dict	Additional keyword arguments for the spline encoding. Only used when encoding='spline'. When None, defaults to {'degree': 3, 'include_bias': True, 'extrapolation': 'periodic'}; n_knots defaults to max_values[feature] + 1 per feature. Knots are placed uniformly over [min_val, min_val + max_val] (e.g. 1-13 for month, 0-24 for hour), giving a periodic period of exactly max_val for both 0-indexed and 1-indexed features and ensuring consistent encoding across training and prediction. Any keyword argument accepted by sklearn.preprocessing.SplineTransformer is allowed (e.g. n_knots, degree, include_bias, extrapolation, order) except knots (computed internally from max_values) and sparse_output (incompatible with the DataFrame output). Passing either of these or an unknown key raises ValueError at fit/transform time.	None
keep_original_columns	bool	If True, the original columns of X are kept in the output DataFrame. If False, only the extracted datetime features are returned. When True and X is an unnamed pandas Series (X.name is None), a ValueError is raised at fit/transform time; either set X.name to a string, or pass keep_original_columns=False. When X is a pandas DatetimeIndex this argument has no effect, as there are no original columns to keep.	True
tol	(float, None)	Absolute tolerance for clamping near-zero values produced by cyclical encoding to exactly 0. Floating point arithmetic can produce values such as -2.4e-16 instead of 0 (e.g. sin(2*pi*12/12)). When not None, any sin or cos value whose absolute value is smaller than tol is replaced with 0.0. Set to None to disable clamping. Only used when encoding='cyclical'.	1e-12

Attributes:

Name	Type	Description
features	(list, None)	List of calendar features to extract from the index. None means the default features are used.
features_to_encode	(list, None)	List of calendar features to encode. None means all extracted features are encoded.
encoding	str	Encoding method for the extracted features.
max_values	(dict, None)	Dictionary of maximum values for the cyclical and spline encoding of calendar features. None means the default values are used.
spline_kwargs	(dict, None)	Keyword arguments for the spline encoding. None means the default values are used (degree=3, include_bias=True, extrapolation='periodic', n_knots=max_val+1 per feature).
keep_original_columns	bool	Whether to keep original columns from the input.
tol	(float, None)	Absolute tolerance for clamping near-zero cyclical encoding values. None means clamping is disabled.
feature_names_out_	list	Names of the output features. Set after calling fit or transform.

Notes

The default max_values use 53 for 'week' and 366 for 'day_of_year' to accommodate the maximum possible values across all calendar years: ISO week 53 occurs in some years (e.g. 2015, 2020, 2026) and day-of-year 366 occurs in leap years. Because the encoding must be stateless (the same for any year, without prior knowledge of whether it is a leap year or contains ISO week 53), the period is fixed at the maximum-possible value. This implies:

• Onehot: the week_53 and day_of_year_366 columns are always present in the output and equal 0 for rows whose year never reaches those values. This guarantees a consistent column schema across training and prediction.
• Cyclical / spline: in years where the maximum value is reached, the cyclical wrap-around is exact (e.g. sin(2π·366/366) = 0 matches sin(2π·0/366) = 0). In years where it is not, there is a one-step "phantom gap" between the highest observed value and 1. The cyclical distance is two steps instead of one. This residual asymmetry is numerically small (≈ 1.7% for day_of_year, ≈ 12% for week) and is strictly preferable to the alternative (period 52 / 365), which would silently collapse week 53 onto week 1 and day 366 onto day 1 in years where those values occur.

For high-cardinality features such as day_of_year (366 columns) or day_of_month (31 columns), encoding='cyclical' (2 columns per feature) or 'spline' (≈ max_val columns per feature, but dense rather than sparse) are typically more memory-efficient than 'onehot', especially on multi-year hourly or sub-daily data.

Output column order: encoded features appear in the order given by features. Non-encoded extracted features (e.g. 'year' or 'weekend', which are never encoded) appear first in the output, in features-list order; encoded features follow, also in features-list order. For example, with features=['year', 'month', 'weekend', 'hour'] and encoding='cyclical' the output columns are [year, weekend, month_sin, month_cos, hour_sin, hour_cos].

Methods:

Name	Description
fit	Fit the transformer by computing the output feature names.
transform	Create datetime features from the DateTime index of a pandas DataFrame or Series.
get_feature_names_out	Get the names of the output features.

Source code in skforecast/preprocessing/_calendar.py

def __init__(
    self,
    features: list[str] | None = None,
    features_to_encode: list[str] | None = None,
    encoding: str = "cyclical",
    max_values: dict[str, int] | None = None,
    spline_kwargs: dict | None = None,
    keep_original_columns: bool = True,
    tol: float | None = 1e-12,
) -> None:

    allowed_features = [
        "year", "month", "week", "day_of_week", "day_of_month",
        "day_of_year", "weekend", "hour", "minute", "second", "quarter",
    ]
    if features is not None:
        not_supported_features = set(features) - set(allowed_features)
        if not_supported_features:
            raise ValueError(
                f"Calendar features {not_supported_features} are not supported. "
                f"Supported features are {allowed_features}."
            )
    else:
        features = allowed_features

    if encoding not in ["cyclical", "onehot", "spline", None]:
        raise ValueError(
            "Encoding must be one of 'cyclical', 'onehot', 'spline' or None"
        )

    self.features = features
    self.features_to_encode = features_to_encode
    self.encoding = encoding
    self.max_values = max_values
    self.spline_kwargs = spline_kwargs
    self.keep_original_columns = keep_original_columns
    self.tol = tol

Attributes

features instance-attribute

features = features

features_to_encode instance-attribute

features_to_encode = features_to_encode

encoding instance-attribute

encoding = encoding

max_values instance-attribute

max_values = max_values

spline_kwargs instance-attribute

spline_kwargs = spline_kwargs

keep_original_columns instance-attribute

keep_original_columns = keep_original_columns

tol instance-attribute

tol = tol

Methods:

fit

fit(X, y=None)

Fit the transformer by computing the output feature names.

Parameters:

Name	Type	Description	Default
X	pandas Series, pandas DataFrame, pandas DatetimeIndex	Input DataFrame or Series with a datetime index, or a pandas DatetimeIndex directly.	required
y	ignored	Not used, present for API compatibility.	None

Returns:

Name	Type	Description
self	CalendarFeatures	Fitted transformer.

Source code in skforecast/preprocessing/_calendar.py

def fit(self, X, y=None):
    """
    Fit the transformer by computing the output feature names.

    Parameters
    ----------
    X : pandas Series, pandas DataFrame, pandas DatetimeIndex
        Input DataFrame or Series with a datetime index, or a pandas
        DatetimeIndex directly.
    y : ignored
        Not used, present for API compatibility.

    Returns
    -------
    self : CalendarFeatures
        Fitted transformer.

    """

    # Slice to the first 2 rows: the encoding is stateless (column
    # names depend on parameters and the index frequency, not on data
    # values), so any non-empty slice yields the same output schema.
    # Non-pandas inputs are passed through unchanged so that
    # `create_calendar_features` raises the appropriate `TypeError`.
    if isinstance(X, pd.DatetimeIndex):
        X_sample = X[:2]
    elif isinstance(X, (pd.DataFrame, pd.Series)):
        X_sample = X.iloc[:2]
    else:
        X_sample = X

    result = create_calendar_features(
                 X                     = X_sample,
                 features              = self.features,
                 features_to_encode    = self.features_to_encode,
                 encoding              = self.encoding,
                 max_values            = self.max_values,
                 spline_kwargs         = self.spline_kwargs,
                 keep_original_columns = self.keep_original_columns,
                 tol                   = self.tol,
             )
    self.feature_names_out_ = list(result.columns)

    return self

transform

transform(X)

Create datetime features from the DateTime index of a pandas DataFrame or Series.

Parameters:

Name	Type	Description	Default
X	pandas Series, pandas DataFrame, pandas DatetimeIndex	Input DataFrame or Series with a datetime index, or a pandas DatetimeIndex directly.	required

Returns:

Name	Type	Description
X_new	pandas DataFrame	DataFrame with the extracted (and optionally encoded) datetime features.

Source code in skforecast/preprocessing/_calendar.py

def transform(
    self,
    X: pd.Series | pd.DataFrame | pd.DatetimeIndex
) -> pd.DataFrame:
    """
    Create datetime features from the DateTime index of a pandas DataFrame or Series.

    Parameters
    ----------
    X : pandas Series, pandas DataFrame, pandas DatetimeIndex
        Input DataFrame or Series with a datetime index, or a pandas
        DatetimeIndex directly.

    Returns
    -------
    X_new : pandas DataFrame
        DataFrame with the extracted (and optionally encoded) datetime features.

    """

    X_new = create_calendar_features(
                X                     = X,
                features              = self.features,
                features_to_encode    = self.features_to_encode,
                encoding              = self.encoding,
                max_values            = self.max_values,
                spline_kwargs         = self.spline_kwargs,
                keep_original_columns = self.keep_original_columns,
                tol                   = self.tol,
            )

    return X_new

get_feature_names_out

get_feature_names_out(input_features=None)

Get the names of the output features.

Parameters:

Name	Type	Description	Default
input_features	list	Ignored. Present for API compatibility with sklearn.	None

Returns:

Name	Type	Description
feature_names_out_	list	Names of the output features.

Source code in skforecast/preprocessing/_calendar.py

def get_feature_names_out(
    self,
    input_features: list[str] | None = None
) -> list[str]:
    """
    Get the names of the output features.

    Parameters
    ----------
    input_features : list, default None
        Ignored. Present for API compatibility with sklearn.

    Returns
    -------
    feature_names_out_ : list
        Names of the output features.

    """

    check_is_fitted(self, "feature_names_out_")

    return self.feature_names_out_

skforecast.preprocessing._preprocessing.reshape_series_wide_to_long

reshape_series_wide_to_long(data, return_multi_index=True)

Convert a pandas DataFrame where each column represents a different time series into a long format DataFrame with a MultiIndex. The index of the input DataFrame must be a pandas DatetimeIndex with a defined frequency. The function reshapes the DataFrame from wide format to long format, where each row corresponds to a specific time point and series ID. The resulting DataFrame will have a MultiIndex with the series IDs as the first level and a pandas DatetimeIndex as the second level. If return_multi_index is set to False, the returned DataFrame have three columns: 'series_id', 'datetime' and 'value', with a regular index.

Parameters:

Name	Type	Description	Default
data	DataFrame	Wide format series. The index must be a pandas DatetimeIndex with a defined frequency and each column must represent a different time series.	required
return_multi_index	bool	If True, the returned DataFrame will have a MultiIndex with the series IDs as the first level and a pandas DatetimeIndex as the second level. If False, the returned DataFrame will have a regular index.	True

Returns:

Name	Type	Description
data	pandas DataFrame	Long format series with a MultiIndex. The first level contains the series IDs, and the second level contains a pandas DatetimeIndex with the same frequency for each series.

Source code in skforecast/preprocessing/_preprocessing.py

def reshape_series_wide_to_long(
    data: pd.DataFrame,
    return_multi_index: bool = True
) -> pd.DataFrame:
    """
    Convert a pandas DataFrame where each column represents a different time series
    into a long format DataFrame with a MultiIndex. The index of the input DataFrame
    must be a pandas DatetimeIndex with a defined frequency. The function reshapes the
    DataFrame from wide format to long format, where each row corresponds to a
    specific time point and series ID. The resulting DataFrame will have a MultiIndex
    with the series IDs as the first level and a pandas DatetimeIndex as the second
    level. If `return_multi_index` is set to False, the returned DataFrame have three
    columns: 'series_id', 'datetime' and 'value', with a regular index.

    Parameters
    ----------
    data: pandas DataFrame
        Wide format series. The index must be a pandas DatetimeIndex with a 
        defined frequency and each column must represent a different time series.
    return_multi_index: bool, default True
        If True, the returned DataFrame will have a MultiIndex with the series IDs
        as the first level and a pandas DatetimeIndex as the second level. If False,
        the returned DataFrame will have a regular index.

    Returns
    -------
    data: pandas DataFrame
        Long format series with a MultiIndex. The first level contains the series IDs,
        and the second level contains a pandas DatetimeIndex with the same frequency
        for each series.

    """

    if not isinstance(data, pd.DataFrame):
        raise TypeError("`data` must be a pandas DataFrame.")

    if not isinstance(data.index, pd.DatetimeIndex):
        raise TypeError("`data` index must be a pandas DatetimeIndex.")

    freq = data.index.freq
    data.index.name = "datetime"
    data = data.reset_index()
    data = pd.melt(data, id_vars="datetime", var_name="series_id", value_name="value")
    data = data.groupby("series_id", sort=False).apply(
        lambda x: x.set_index("datetime").asfreq(freq), include_groups=False
    )

    if not return_multi_index:
        data = data.reset_index()

    return data

skforecast.preprocessing._preprocessing.reshape_series_long_to_dict

reshape_series_long_to_dict(
    data,
    freq,
    series_id=None,
    index=None,
    values=None,
    fill_value=None,
    suppress_warnings=False,
)

Convert a long-format DataFrame into a dictionary of pandas Series with the specified frequency. Supports two input formats:

• A pandas DataFrame with explicit columns for the series identifier, time index, and values.
• A pandas DataFrame with a MultiIndex, where the first level contains the series IDs, and the second level contains a pandas DatetimeIndex.

Parameters:

Name	Type	Description	Default
data	DataFrame	Long-format series.	required
freq	str	Frequency of the series.	required
series_id	str | None	Column name with the series identifier. Not needed if the input data is a pandas DataFrame with MultiIndex.	None
index	str | None	Column name with the time index. Not needed if the input data is a pandas DataFrame with MultiIndex.	None
values	str | None	Column name with the values. Not needed if the input data is a pandas DataFrame with MultiIndex.	None
fill_value	float | None	Value to use for filling gaps created when setting the frequency with asfreq (note this does not fill NaNs that already were present). If None, gaps will contain NaN values.	None
suppress_warnings	bool	If True, suppress warnings when a series is incomplete after setting the frequency.	False

Returns:

Name	Type	Description
series_dict	dict	Dictionary with the series.

Source code in skforecast/preprocessing/_preprocessing.py

def reshape_series_long_to_dict(
    data: pd.DataFrame,
    freq: str,
    series_id: str | None = None,
    index: str | None = None,
    values: str | None = None,
    fill_value: float | None = None,
    suppress_warnings: bool = False
) -> dict[str, pd.Series]:
    """
    Convert a long-format DataFrame into a dictionary of pandas Series with the 
    specified frequency. Supports two input formats:

    - A pandas DataFrame with explicit columns for the series identifier, time 
    index, and values.
    - A pandas DataFrame with a MultiIndex, where the first level contains the 
    series IDs, and the second level contains a pandas DatetimeIndex.

    Parameters
    ----------
    data: pandas DataFrame
        Long-format series.
    freq: str
        Frequency of the series.
    series_id: str, default None
        Column name with the series identifier. Not needed if the input data
        is a pandas DataFrame with MultiIndex.
    index: str, default None
        Column name with the time index. Not needed if the input data is a pandas
        DataFrame with MultiIndex.
    values: str, default None
        Column name with the values. Not needed if the input data is a pandas
        DataFrame with MultiIndex.
    fill_value: float, default None
        Value to use for filling gaps created when setting the frequency with 
        `asfreq` (note this does not fill NaNs that already were present). If 
        None, gaps will contain NaN values.
    suppress_warnings: bool, default False
        If True, suppress warnings when a series is incomplete after setting the
        frequency.

    Returns
    -------
    series_dict: dict
        Dictionary with the series.

    """

    if not isinstance(data, pd.DataFrame):
        raise TypeError("`data` must be a pandas DataFrame.")

    if isinstance(data.index, pd.MultiIndex):

        data = data.copy()
        first_col = data.columns[0]
        data.index = data.index.set_names([data.index.names[0], None])
        series_dict = {}
        for k, group in data.groupby(level=0, sort=True, observed=True):
            group = group.droplevel(0)
            original_size = len(group)
            series_dict[k] = group[first_col].rename(k).asfreq(freq, fill_value=fill_value)
            if not suppress_warnings and len(series_dict[k]) != original_size:
                fill_msg = (
                    "NaNs have been introduced"
                    if fill_value is None
                    else f"Missing values have been filled with {fill_value}"
                )
                warnings.warn(
                    f"Series '{k}' is incomplete. {fill_msg} after "
                    f"setting the frequency.",
                    MissingValuesWarning
                )

    else:

        for col in [series_id, index, values]:
            if col is None:
                raise ValueError(
                    "Arguments `series_id`, `index`, and `values` must be "
                    "specified when the input DataFrame does not have a MultiIndex. "
                    "Please provide a value for each of these arguments."
                )
            if col not in data.columns:
                raise ValueError(f"Column '{col}' not found in `data`.")

        data_grouped = data.groupby(series_id, observed=True)   
        original_sizes = data_grouped.size()
        series_dict = {}
        for k, v in data_grouped:
            series_dict[k] = v.set_index(index)[values].asfreq(freq, fill_value=fill_value).rename(k)
            series_dict[k].index.name = None
            if not suppress_warnings and len(series_dict[k]) != original_sizes[k]:
                fill_msg = (
                    "NaNs have been introduced"
                    if fill_value is None
                    else f"Missing values have been filled with {fill_value}"
                )
                warnings.warn(
                    f"Series '{k}' is incomplete. {fill_msg} after "
                    f"setting the frequency.",
                    MissingValuesWarning
                )

    return series_dict

skforecast.preprocessing._preprocessing.reshape_exog_long_to_dict

reshape_exog_long_to_dict(
    data,
    freq,
    series_id=None,
    index=None,
    fill_value=None,
    drop_all_nan_cols=False,
    consolidate_dtypes=True,
    suppress_warnings=False,
)

Convert a long-format DataFrame of exogenous variables into a dictionary of pandas DataFrames with the specified frequency. Supports two input formats:

• A pandas DataFrame with explicit columns for the series identifier, time index, and exogenous variables.
• A pandas DataFrame with a MultiIndex, where the first level contains the series IDs, and the second level contains a pandas DatetimeIndex.

Parameters:

Name	Type	Description	Default
data	DataFrame	Long format exogenous variables.	required
freq	str	Frequency of the series.	required
series_id	str | None	Column name with the series identifier. Not needed if the input data is a pandas DataFrame with MultiIndex.	None
index	str | None	Column name with the time index. Not needed if the input data is a pandas DataFrame with MultiIndex.	None
fill_value	float | None	Value to use for filling gaps created when setting the frequency with asfreq (note this does not fill NaNs that already were present). If None, gaps will contain NaN values. Only applied to numeric columns; non-numeric columns (e.g. string, categorical) will still contain NaN in the gaps.	None
drop_all_nan_cols	bool	If True, drop columns with all values as NaN. This is useful when there are series without some exogenous variables.	False
consolidate_dtypes	bool	Consolidate the data types of the exogenous variables across all series. If, after setting the frequency, NaNs are introduced in any series and cause a column's dtype to change to float, that column is also cast to float in every other series so that all series share the same dtypes.	True
suppress_warnings	bool	If True, suppress warnings when exog is incomplete after setting the frequency.	False

Returns:

Name	Type	Description
exog_dict	dict	Dictionary with the exogenous variables.

Source code in skforecast/preprocessing/_preprocessing.py

def reshape_exog_long_to_dict(
    data: pd.DataFrame,
    freq: str,
    series_id: str | None = None,
    index: str | None = None,
    fill_value: float | None = None,
    drop_all_nan_cols: bool = False,
    consolidate_dtypes: bool = True,
    suppress_warnings: bool = False
) -> dict[str, pd.DataFrame]:
    """
    Convert a long-format DataFrame of exogenous variables into a dictionary 
    of pandas DataFrames with the specified frequency. Supports two input formats:

    - A pandas DataFrame with explicit columns for the series identifier, time 
    index, and exogenous variables.
    - A pandas DataFrame with a MultiIndex, where the first level contains the 
    series IDs, and the second level contains a pandas DatetimeIndex.

    Parameters
    ----------
    data: pandas DataFrame
        Long format exogenous variables.
    freq: str
        Frequency of the series.
    series_id: str, default None
        Column name with the series identifier. Not needed if the input data
        is a pandas DataFrame with MultiIndex.
    index: str, default None
        Column name with the time index. Not needed if the input data is a pandas
        DataFrame with MultiIndex.
    fill_value: float, default None
        Value to use for filling gaps created when setting the frequency with 
        `asfreq` (note this does not fill NaNs that already were present). If 
        None, gaps will contain NaN values. Only applied to numeric columns;
        non-numeric columns (e.g. string, categorical) will still contain NaN
        in the gaps.
    drop_all_nan_cols: bool, default False
        If True, drop columns with all values as NaN. This is useful when
        there are series without some exogenous variables.
    consolidate_dtypes: bool, default True
        Consolidate the data types of the exogenous variables across all series.
        If, after setting the frequency, NaNs are introduced in any series and
        cause a column's dtype to change to float, that column is also cast to
        float in every other series so that all series share the same dtypes.
    suppress_warnings: bool, default False
        If True, suppress warnings when exog is incomplete after setting the
        frequency.

    Returns
    -------
    exog_dict: dict
        Dictionary with the exogenous variables.

    """

    if not isinstance(data, pd.DataFrame):
        raise TypeError("`data` must be a pandas DataFrame.")

    if isinstance(data.index, pd.MultiIndex):

        data = data.copy()
        data.index = data.index.set_names([data.index.names[0], None])
        exog_dict = {}
        cols_float_dtype = set()
        nans_introduced = False
        for k, group in data.groupby(level=0, sort=True, observed=True):
            group = group.droplevel(0)
            original_index = group.index
            original_size = len(group)
            exog_dict[k] = group.asfreq(freq)
            if len(exog_dict[k]) != original_size:
                nans_introduced = True
                non_numeric_cols = []
                if fill_value is not None:
                    numeric_cols = exog_dict[k].select_dtypes(include='number').columns
                    non_numeric_cols = exog_dict[k].columns.difference(numeric_cols)
                    new_rows_mask = ~exog_dict[k].index.isin(original_index)
                    if len(numeric_cols) > 0:
                        exog_dict[k].loc[new_rows_mask, numeric_cols] = (
                            exog_dict[k].loc[new_rows_mask, numeric_cols].fillna(fill_value)
                        )
                if not suppress_warnings:
                    if fill_value is None:
                        fill_msg = "NaNs have been introduced"
                    else:
                        fill_msg = (
                            f"Missing values have been filled with {fill_value}"
                        )
                        if len(non_numeric_cols) > 0:
                            fill_msg += (
                                f" in numeric columns only. Non-numeric columns "
                                f"{list(non_numeric_cols)} still contain NaN"
                            )
                    warnings.warn(
                        f"Exogenous variables for series '{k}' are incomplete. "
                        f"{fill_msg} after setting the frequency.",
                        MissingValuesWarning
                    )
                if consolidate_dtypes:
                    cols_float_dtype.update(
                        {
                            col for col in exog_dict[k].columns
                            if pd.api.types.is_float_dtype(exog_dict[k][col])
                        }
                    )

    else:

        for col in [series_id, index]:
            if col is None:
                raise ValueError(
                    "Arguments `series_id`, and `index` must be "
                    "specified when the input DataFrame does not have a MultiIndex. "
                    "Please provide a value for each of these arguments."
                )
            if col not in data.columns:
                raise ValueError(f"Column '{col}' not found in `data`.")

        cols_float_dtype = {
            col for col in data.columns
            if col not in (series_id, index)
            and pd.api.types.is_float_dtype(data[col])
        }

        data_grouped = data.groupby(series_id, observed=True) 
        original_sizes = data_grouped.size()
        exog_dict = dict(tuple(data_grouped))
        original_indices = {
            k: set(v[index]) for k, v in exog_dict.items()
        }
        exog_dict = {
            k: v.set_index(index).drop(columns=series_id).asfreq(freq)
            for k, v in exog_dict.items()
        }

        for k in exog_dict.keys():
            exog_dict[k].index.name = None

        nans_introduced = False
        for k, v in exog_dict.items():
            if len(v) != original_sizes[k]:
                nans_introduced = True
                non_numeric_cols = []
                if fill_value is not None:
                    numeric_cols = v.select_dtypes(include='number').columns
                    non_numeric_cols = v.columns.difference(numeric_cols)
                    new_rows_mask = ~v.index.isin(original_indices[k])
                    if len(numeric_cols) > 0:
                        exog_dict[k].loc[new_rows_mask, numeric_cols] = (
                            v.loc[new_rows_mask, numeric_cols].fillna(fill_value)
                        )
                if not suppress_warnings:
                    if fill_value is None:
                        fill_msg = "NaNs have been introduced"
                    else:
                        fill_msg = (
                            f"Missing values have been filled with {fill_value}"
                        )
                        if len(non_numeric_cols) > 0:
                            fill_msg += (
                                f" in numeric columns only. Non-numeric columns "
                                f"{list(non_numeric_cols)} still contain NaN"
                            )
                    warnings.warn(
                        f"Exogenous variables for series '{k}' are incomplete. "
                        f"{fill_msg} after setting the frequency.",
                        MissingValuesWarning
                    )
                if consolidate_dtypes:
                    cols_float_dtype.update(
                        {
                            col for col in v.columns 
                            if pd.api.types.is_float_dtype(v[col])
                        }
                    )

    if consolidate_dtypes and nans_introduced:
        new_dtypes = {col: float for col in cols_float_dtype}
        exog_dict = {k: v.astype(new_dtypes, copy=False) for k, v in exog_dict.items()}

    if drop_all_nan_cols:
        exog_dict = {k: v.dropna(how="all", axis=1) for k, v in exog_dict.items()}

    return exog_dict

skforecast.preprocessing._preprocessing.reshape_series_exog_dict_to_long

reshape_series_exog_dict_to_long(
    series,
    exog,
    series_col_name="series_value",
    index_names=["series_id", "datetime"],
    merge_how="left",
)

Convert dictionaries of series and exogenous variables to a long-format pandas DataFrame with MultiIndex. The first level of the MultiIndex contains the series identifiers, and the second level contains the temporal index. If both series and exog are provided, they are merged into a single DataFrame.

Parameters:

Name	Type	Description	Default
series	dict[str, Series] | None	Dictionary with multiple time series (expected: dict[str, pd.Series]).	required
exog	dict[str, Series | DataFrame] | None	Dictionary with exogenous variables (expected: dict[str, pd.Series or pd.DataFrame]).	required
series_col_name	str	Column name for the series values in the resulting DataFrame.	'series_value'
index_names	list[str]	Names for the levels of the MultiIndex in the resulting DataFrame. The first name corresponds to the series identifier, and the second name corresponds to the temporal index.	['series_id', 'datetime']
merge_how	str	Type of merge to perform when combining series and exog. Options are: 'left': Keep only indices from series (default) 'right': Keep only indices from exog 'outer': Keep all indices from both series and exog 'inner': Keep only indices present in both	'left'

Returns:

Name	Type	Description
long_df	DataFrame	Long-format DataFrame with a MultiIndex of two levels: - First level: series identifier (named by index_names[0], default 'series_id') - Second level: temporal index (named by index_names[1], default 'datetime') Columns include: - Series values (named by series_col_name, default 'series_value') if series is provided. - Exogenous variable columns (from exog) if exog is provided. If both series and exog are provided, columns from both are present. If only one is provided, only its columns are present.

Source code in skforecast/preprocessing/_preprocessing.py

def reshape_series_exog_dict_to_long(
    series: dict[str, pd.Series] | None,
    exog: dict[str, pd.Series | pd.DataFrame] | None,
    series_col_name: str = 'series_value',
    index_names: list[str] = ['series_id', 'datetime'],
    merge_how: str = 'left'
) -> pd.DataFrame:
    """
    Convert dictionaries of series and exogenous variables to a long-format
    pandas DataFrame with MultiIndex. The first level of the MultiIndex contains the
    series identifiers, and the second level contains the temporal index. If both
    series and exog are provided, they are merged into a single DataFrame.

    Parameters
    ----------
    series: dict, None
        Dictionary with multiple time series (expected: dict[str, pd.Series]).
    exog: dict, None
        Dictionary with exogenous variables (expected: dict[str, pd.Series or pd.DataFrame]).
    series_col_name: str, default 'series_value'
        Column name for the series values in the resulting DataFrame.
    index_names: list[str], default ['series_id', 'datetime']
        Names for the levels of the MultiIndex in the resulting DataFrame. The first
        name corresponds to the series identifier, and the second name corresponds
        to the temporal index.
    merge_how: str, default 'left'
        Type of merge to perform when combining `series` and `exog`. Options are:

        - 'left': Keep only indices from `series` (default)
        - 'right': Keep only indices from `exog`
        - 'outer': Keep all indices from both `series` and `exog`
        - 'inner': Keep only indices present in both

    Returns
    -------
    long_df : pandas.DataFrame
        Long-format DataFrame with a MultiIndex of two levels:
        - First level: series identifier (named by `index_names[0]`, default 'series_id')
        - Second level: temporal index (named by `index_names[1]`, default 'datetime')
        Columns include:
        - Series values (named by `series_col_name`, default 'series_value') if `series` is provided.
        - Exogenous variable columns (from `exog`) if `exog` is provided.
        If both `series` and `exog` are provided, columns from both are present.
        If only one is provided, only its columns are present.

    """

    if series is None and exog is None:
        raise ValueError("Both `series` and `exog` cannot be None.")

    if series is not None:
        if not isinstance(series, dict):
            raise TypeError(f"`series` must be a dictionary. Got {type(series)}.")
        for k, v in series.items():
            if not isinstance(v, pd.Series):
                raise TypeError(f"`series['{k}']` must be a pandas Series.")
        series = pd.concat(series, names=index_names).to_frame(series_col_name)

    if exog is not None:
        if not isinstance(exog, dict):
            raise TypeError(f"`exog` must be a dictionary. Got {type(exog)}.")
        for k, v in exog.items():
            if not isinstance(v, (pd.Series, pd.DataFrame)):
                raise TypeError(
                    f"`exog['{k}']` must be a pandas Series or a pandas DataFrame."
                )
        exog = pd.concat(exog, names=index_names)
        if isinstance(exog, pd.Series):
            exog = exog.to_frame(name='exog_value')

    if series is not None and exog is not None:
        series_idx_type = type(series.index.get_level_values(1))
        exog_idx_type = type(exog.index.get_level_values(1))

        if series_idx_type != exog_idx_type:
            raise TypeError(
                f"Index type mismatch: series has index of type "
                f"{series_idx_type}, but `exog` has {exog_idx_type}. "
                f"Ensure all indices are compatible."
            )

        if series_col_name in exog.columns:
            raise ValueError(
                f"Column name conflict: '{series_col_name}' already exists in `exog`. "
                f"Please choose a different `series_col_name` value."
            )

    if series is None:
        long_df = exog
    elif exog is None:
        long_df = series
    else:
        long_df = pd.merge(
            series, exog, left_index=True, right_index=True, how=merge_how
        )

    return long_df

skforecast.preprocessing._preprocessing.ConformalIntervalCalibrator

ConformalIntervalCalibrator(
    nominal_coverage=0.8, symmetric_calibration=True
)

Transformer that calibrates the prediction interval to achieve the desired coverage based on conformity scores. It uses the conformal split method.

Parameters:

Name	Type	Description	Default
nominal_coverage	float	Desired coverage. This is the desired probability that the true value falls within the calibrated interval.	0.8
symmetric_calibration	bool	If True, the calibration factor is the same for the lower and upper bounds. If False, the calibration factor is different for the lower and upper bounds.	True

Attributes:

Name	Type	Description
nominal_coverage	float	Desired coverage. This is the desired probability that the true value falls within the calibrated interval.
symmetric_calibration	bool, default True	If True, the calibration factor is the same for the lower and upper bounds. If False, the calibration factor is different for the lower and upper bounds.
correction_factor_	dict	Correction factor to achieve the desired coverage. This is the correction factor used when symmetric_calibration is True.
correction_factor_lower_	dict	Correction factor for the lower bound to achieve the desired coverage. It is used when symmetric_calibration is False.
correction_factor_upper_	dict	Correction factor for the upper bound to achieve the desired coverage. It is used when symmetric_calibration is False.
fit_coverage_	dict	Coverage observed in the data used to fit the transformer. This is the empirical coverage from which the correction factor is learned.
fit_input_type_	str	Type of input data used to fit the transformer. Can be 'single' or 'multi'.
fit_series_names_	list	Names of the series used to fit the transformer.

Methods:

Name	Description
fit	Learn the correction factor needed to achieve the desired coverage.
transform	Apply the correction factor to the prediction interval to achieve the desired

Source code in skforecast/preprocessing/_preprocessing.py

def __init__(
    self,
    nominal_coverage: float = 0.8,
    symmetric_calibration: bool = True
) -> None:

    if nominal_coverage < 0 or nominal_coverage > 1:
        raise ValueError(
            f"`nominal_coverage` must be a float between 0 and 1. Got {nominal_coverage}"
        )

    self.nominal_coverage         = nominal_coverage
    self.symmetric_calibration    = symmetric_calibration
    self.correction_factor_       = {}
    self.correction_factor_lower_ = {}
    self.correction_factor_upper_ = {}
    self.fit_coverage_            = {}
    self.fit_input_type_          = None
    self.fit_series_names_        = None
    self.is_fitted                = False

Attributes

nominal_coverage instance-attribute

nominal_coverage = nominal_coverage

symmetric_calibration instance-attribute

symmetric_calibration = symmetric_calibration

correction_factor_ instance-attribute

correction_factor_ = {}

correction_factor_lower_ instance-attribute

correction_factor_lower_ = {}

correction_factor_upper_ instance-attribute

correction_factor_upper_ = {}

fit_coverage_ instance-attribute

fit_coverage_ = {}

fit_input_type_ instance-attribute

fit_input_type_ = None

fit_series_names_ instance-attribute

fit_series_names_ = None

is_fitted instance-attribute

is_fitted = False

Methods:

_repr_html_

_repr_html_()

HTML representation of the object. The "General Information" section is expanded by default.

Source code in skforecast/preprocessing/_preprocessing.py

def _repr_html_(self) -> str:
    """
    HTML representation of the object.
    The "General Information" section is expanded by default.
    """

    style, unique_id = get_style_repr_html(is_fitted=self.is_fitted)

    content = f"""
    <div class="container-{unique_id}">
        <p style="font-size: 1.5em; font-weight: bold; margin-block-start: 0.83em; margin-block-end: 0.83em;">{type(self).__name__}</p>
        <details open>
            <summary>General Information</summary>
            <ul>
                <li><strong>Nominal coverage:</strong> {self.nominal_coverage}</li>
                <li><strong>Coverage in fit data:</strong> {self.fit_coverage_}</li>
                <li><strong>Symmetric interval:</strong> {self.symmetric_calibration}</li>
                <li><strong>Symmetric correction factor:</strong> {self.correction_factor_}</li>
                <li><strong>Asymmetric correction factor lower:</strong> {self.correction_factor_lower_}</li>
                <li><strong>Asymmetric correction factor upper:</strong> {self.correction_factor_upper_}</li>
                <li><strong>Fitted series:</strong> {self.fit_series_names_}</li>
            </ul>
        </details>
        <p>
            <a href="https://skforecast.org/{__version__}/api/preprocessing#skforecast.preprocessing._preprocessing.ConformalIntervalCalibrator">&#128214; <strong>API Reference</strong></a>
            &nbsp;&nbsp;
            <a href="https://skforecast.org/{__version__}/user_guides/probabilistic-forecasting-conformal-calibration.html">&#128221; <strong>User Guide</strong></a>
        </p>
    </div>
    """

    return style + content

fit

fit(y_true, y_pred_interval)

Learn the correction factor needed to achieve the desired coverage.

Parameters:

Name	Type	Description	Default
y_true	pandas Series, pandas DataFrame, dict	True values of the time series. If pandas Series, it is assumed that only one series is available. If pandas DataFrame, it is assumed that each column is a different series which will be calibrated separately. The column names are used as series names. If dict, it is assumed that each key is a series name and the corresponding value is a pandas Series with the true values.	required
y_pred_interval	pandas DataFrame	Prediction interval estimated for the time series. If y_true contains only one series, y_pred_interval must have two columns, 'lower_bound' and 'upper_bound'. If y_true contains multiple series, y_pred_interval must be a long-format DataFrame with three columns: 'level', 'lower_bound', and 'upper_bound'. The 'level' column identifies the series to which each interval belongs.	required

Returns:

Type	Description
None

Source code in skforecast/preprocessing/_preprocessing.py

def fit(
    self,
    y_true: pd.Series | pd.DataFrame | dict[str, pd.Series],
    y_pred_interval: pd.DataFrame,
) -> None:
    """
    Learn the correction factor needed to achieve the desired coverage.

    Parameters
    ----------
    y_true : pandas Series, pandas DataFrame, dict
        True values of the time series.

        - If pandas Series, it is assumed that only one series is available.
        - If pandas DataFrame, it is assumed that each column is a different 
        series which will be calibrated separately. The column names are 
        used as series names.
        - If dict, it is assumed that each key is a series name and the 
        corresponding value is a pandas Series with the true values.
    y_pred_interval : pandas DataFrame
        Prediction interval estimated for the time series. 

        - If `y_true` contains only one series, `y_pred_interval` must have 
        two columns, 'lower_bound' and 'upper_bound'.
        - If `y_true` contains multiple series, `y_pred_interval` must be
        a long-format DataFrame with three columns: 'level', 'lower_bound',
        and 'upper_bound'. The 'level' column identifies the series to which
        each interval belongs.

    Returns
    -------
    None

    """

    self.correction_factor_       = {}
    self.correction_factor_lower_ = {}
    self.correction_factor_upper_ = {}
    self.fit_coverage_            = {}
    self.fit_input_type_          = None
    self.fit_series_names_        = None
    self.is_fitted                = False

    if not isinstance(y_true, (pd.Series, pd.DataFrame, dict)):
        raise TypeError(
            "`y_true` must be a pandas Series, pandas DataFrame, or a dictionary."
        )

    if not isinstance(y_pred_interval, (pd.DataFrame)):
        raise TypeError(
            "`y_pred_interval` must be a pandas DataFrame."
        )

    if not set(["lower_bound", "upper_bound"]).issubset(y_pred_interval.columns):
        raise ValueError(
            "`y_pred_interval` must have columns 'lower_bound' and 'upper_bound'."
        )

    if isinstance(y_true, (pd.DataFrame, dict)) and 'level' not in y_pred_interval.columns:
        raise ValueError(
            "If `y_true` is a pandas DataFrame or a dictionary, `y_pred_interval` "
            "must have an additional column 'level' to identify each series."
        )

    if isinstance(y_true, pd.Series):
        name = y_true.name if y_true.name is not None else 'y'
        self.fit_input_type_ = "single_series"    
        y_true = {name: y_true}

        if "level" not in y_pred_interval.columns:
            y_pred_interval = y_pred_interval.copy()
            y_pred_interval["level"] = name
        else:
            if y_pred_interval["level"].nunique() > 1:
                raise ValueError(
                    "If `y_true` is a pandas Series, `y_pred_interval` must have "
                    "only one series. Found multiple values in column 'level'."
                )
            if y_pred_interval["level"].iat[0] != name:
                raise ValueError(
                    f"Series name in `y_true`, '{name}', does not match the level "
                    f"name in `y_pred_interval`, '{y_pred_interval['level'].iat[0]}'."
                )
    elif isinstance(y_true, pd.DataFrame):
        self.fit_input_type_ = "multiple_series"
        y_true = y_true.to_dict(orient='series')
    else:
        self.fit_input_type_ = "multiple_series"
        for k, v in y_true.items():
            if not isinstance(v, pd.Series):
                raise ValueError(
                    f"When `y_true` is a dict, all its values must be pandas "
                    f"Series. Got {type(v)} for series '{k}'."
                )

    y_pred_interval = {
        k: v[['lower_bound', 'upper_bound']]
        for k, v in y_pred_interval.groupby('level')
    }

    if not y_pred_interval.keys() == y_true.keys():
        raise ValueError(
            f"Series names in `y_true` and `y_pred_interval` do not match.\n"
            f"   `y_true` series names          : {list(y_true.keys())}\n"
            f"   `y_pred_interval` series names : {list(y_pred_interval.keys())}"
        )

    for k in y_true.keys():

        if not y_true[k].index.equals(y_pred_interval[k].index):
            raise IndexError(
                f"Index of `y_true` and `y_pred_interval` must match. Different "
                f"indices found for series '{k}'."
            )

        y_true_ = np.asarray(y_true[k])
        y_pred_interval_ = np.asarray(y_pred_interval[k])

        lower_bound = y_pred_interval_[:, 0]
        upper_bound = y_pred_interval_[:, 1]
        conformity_scores_lower = lower_bound - y_true_
        conformity_scores_upper = y_true_ - upper_bound
        conformity_scores = np.max(
            [
                conformity_scores_lower,
                conformity_scores_upper,
            ],
            axis=0,
        )

        self.correction_factor_[k] = float(np.quantile(conformity_scores, self.nominal_coverage))
        self.correction_factor_lower_[k] = float(
            -1 * np.quantile(-1 * conformity_scores_lower, (1 - self.nominal_coverage) / 2)
        )
        self.correction_factor_upper_[k] = float(
            np.quantile(conformity_scores_upper,  1 - (1 - self.nominal_coverage) / 2)
        )
        coverage_fit_ = calculate_coverage(
                            y_true      = y_true_,
                            lower_bound = lower_bound,
                            upper_bound = upper_bound,
                        )
        self.fit_coverage_[k] = float(coverage_fit_)

    self.is_fitted = True
    self.fit_series_names_ = list(y_true.keys())

transform

transform(y_pred_interval)

Apply the correction factor to the prediction interval to achieve the desired coverage.

Parameters:

Name	Type	Description	Default
y_pred_interval	pandas DataFrame	Prediction interval to be calibrated using conformal method. If only intervals for one series are available, y_pred_interval must have two columns, 'lower_bound' and 'upper_bound'. If multiple series are available, y_pred_interval must be a long-format DataFrame with three columns: 'level', 'lower_bound', and 'upper_bound'. The 'level' column identifies the series to which each interval belongs.	required

Returns:

Name	Type	Description
y_pred_interval_conformal	pandas DataFrame	Prediction interval with the correction factor applied.

Source code in skforecast/preprocessing/_preprocessing.py

def transform(
    self, 
    y_pred_interval: pd.DataFrame
) -> pd.DataFrame:
    """
    Apply the correction factor to the prediction interval to achieve the desired
    coverage.

    Parameters
    ----------
    y_pred_interval : pandas DataFrame
        Prediction interval to be calibrated using conformal method.

        - If only intervals for one series are available, `y_pred_interval` 
        must have two columns, 'lower_bound' and 'upper_bound'.
        - If multiple series are available, `y_pred_interval` must be
        a long-format DataFrame with three columns: 'level', 'lower_bound',
        and 'upper_bound'. The 'level' column identifies the series to which
        each interval belongs.

    Returns
    -------
    y_pred_interval_conformal : pandas DataFrame
        Prediction interval with the correction factor applied.

    """

    if not self.is_fitted:
        raise NotFittedError(
            "ConformalIntervalCalibrator not fitted yet. Call 'fit' with "
            "training data first."
        )
    if not isinstance(y_pred_interval, pd.DataFrame):
        raise TypeError(
            "`y_pred_interval` must be a pandas DataFrame."
        )

    if not set(["lower_bound", "upper_bound"]).issubset(y_pred_interval.columns):
        raise ValueError(
            "`y_pred_interval` must have columns 'lower_bound' and 'upper_bound'."
        )

    if self.fit_input_type_ == "single_series" and 'level' not in y_pred_interval.columns:
        y_pred_interval = y_pred_interval.copy()
        y_pred_interval["level"] = self.fit_series_names_[0]

    if self.fit_input_type_ == "multiple_series" and 'level' not in y_pred_interval.columns:
        raise ValueError(
            "The transformer was fitted with multiple series. `y_pred_interval` "
            "must be a long-format DataFrame with three columns: 'level', "
            "'lower_bound', and 'upper_bound'. The 'level' column identifies "
            "the series to which each interval belongs."
        )

    conformalized_intervals = []
    for k, y_pred_interval_ in y_pred_interval.groupby('level')[['lower_bound', 'upper_bound']]:

        if k not in self.fit_series_names_:
            raise ValueError(
                f"Series '{k}' was not seen during fit. Available series are: "
                f"{self.fit_series_names_}."
            )

        correction_factor = self.correction_factor_[k]   
        correction_factor_lower = self.correction_factor_lower_[k]
        correction_factor_upper = self.correction_factor_upper_[k]

        index = y_pred_interval_.index
        y_pred_interval_ = y_pred_interval_.to_numpy()
        y_pred_interval_conformal = y_pred_interval_.copy()

        if self.symmetric_calibration:
            y_pred_interval_conformal[:, 0] = (
                y_pred_interval_conformal[:, 0] - correction_factor
            )
            y_pred_interval_conformal[:, 1] = (
                y_pred_interval_conformal[:, 1] + correction_factor
            )
        else:
            y_pred_interval_conformal[:, 0] = (
                y_pred_interval_conformal[:, 0] - correction_factor_lower
            )
            y_pred_interval_conformal[:, 1] = (
                y_pred_interval_conformal[:, 1] + correction_factor_upper
            )

        # If upper bound is less than lower bound, swap them
        mask = (
            y_pred_interval_conformal[:, 1]
            < y_pred_interval_conformal[:, 0]
        )

        (
            y_pred_interval_conformal[mask, 0],
            y_pred_interval_conformal[mask, 1],
        ) = (
            y_pred_interval_conformal[mask, 1],
            y_pred_interval_conformal[mask, 0],
        )

        y_pred_interval_conformal = pd.DataFrame(
            data    = y_pred_interval_conformal,
            columns = ['lower_bound', 'upper_bound'],
            index   = index
        )
        y_pred_interval_conformal.insert(0, 'level', k)
        conformalized_intervals.append(y_pred_interval_conformal)

    conformalized_intervals = pd.concat(conformalized_intervals)

    return conformalized_intervals

skforecast.preprocessing._preprocessing.TimeSeriesDifferentiator

TimeSeriesDifferentiator(order=1, window_size=None)

Bases: BaseEstimator, TransformerMixin

Transforms a time series into a differentiated time series of a specified order and provides functionality to revert the differentiation.

When using a direct module Forecaster, the model in step 1 must be used if you want to reverse the differentiation of the training time series with the inverse_transform_training method.

Parameters:

Name	Type	Description	Default
order	int	The order of differentiation to be applied.	1
window_size	int	The window size used by the forecaster. This is required to revert the differentiation for the target variable y or its predicted values.	None

Attributes:

Name	Type	Description
order	int	The order of differentiation.
initial_values	list	List with the first value of the time series before each differentiation. If order = 2, first value correspond with the first value of the original time series and the second value correspond with the first value of the differentiated time series of order 1. These values are necessary to revert the differentiation and reconstruct the original time series.
pre_train_values	list	List with the first training value of the time series before each differentiation. For order = 1, the value correspond with the last value of the window used to create the predictors. For order > 1, the value correspond with the first value of the differentiated time series prior to the next differentiation. These values are necessary to revert the differentiation and reconstruct the training time series.
last_values	list	List with the last value of the time series before each differentiation, used to revert differentiation on subsequent data windows. If order = 2, first value correspond with the last value of the original time series and the second value correspond with the last value of the differentiated time series of order 1. This is essential for correctly transforming a time series that follows immediately after the series used to fit the transformer.

Methods:

Name	Description
fit	Fits the transformer. Stores the values needed to revert the
transform	Transforms a time series into a differentiated time series of order n.
inverse_transform	Reverts the differentiation. To do so, the input array is assumed to be
inverse_transform_training	Reverts the differentiation. To do so, the input array is assumed to be
inverse_transform_next_window	Reverts the differentiation. The input array X is assumed to be a
set_params	Set the parameters of the TimeSeriesDifferentiator.

Source code in skforecast/preprocessing/_preprocessing.py

def __init__(
    self, 
    order: int = 1,
    window_size: int | None = None
) -> None:

    if not isinstance(order, (int, np.integer)):
        raise TypeError(
            f"Parameter `order` must be an integer greater than 0. Found {type(order)}."
        )
    if order < 1:
        raise ValueError(
            f"Parameter `order` must be an integer greater than 0. Found {order}."
        )

    if window_size is not None:
        if not isinstance(window_size, (int, np.integer)):
            raise TypeError(
                f"Parameter `window_size` must be an integer greater than 0. "
                f"Found {type(window_size)}."
            )
        if window_size < 1:
            raise ValueError(
                f"Parameter `window_size` must be an integer greater than 0. "
                f"Found {window_size}."
            )

    self.order = order
    self.window_size = window_size
    self.initial_values = []
    self.pre_train_values = []
    self.last_values = []

Attributes

order instance-attribute

order = order

window_size instance-attribute

window_size = window_size

initial_values instance-attribute

initial_values = []

pre_train_values instance-attribute

pre_train_values = []

last_values instance-attribute

last_values = []

Methods:

fit

fit(X, y=None)

Fits the transformer. Stores the values needed to revert the differentiation of different window of the time series, original time series, training time series, and a time series that follows immediately after the series used to fit the transformer.

Parameters:

Name	Type	Description	Default
X	numpy ndarray	Time series to be differentiated.	required
y	Ignored	Not used, present here for API consistency by convention.	None

Returns:

Name	Type	Description
self	TimeSeriesDifferentiator

Source code in skforecast/preprocessing/_preprocessing.py

@_check_X_numpy_ndarray_1d()
def fit(
    self, 
    X: np.ndarray, 
    y: Any = None
) -> TimeSeriesDifferentiator:
    """
    Fits the transformer. Stores the values needed to revert the 
    differentiation of different window of the time series, original 
    time series, training time series, and a time series that follows
    immediately after the series used to fit the transformer.

    Parameters
    ----------
    X : numpy ndarray
        Time series to be differentiated.
    y : Ignored
        Not used, present here for API consistency by convention.

    Returns
    -------
    self : TimeSeriesDifferentiator

    """

    self.initial_values = []
    self.pre_train_values = []
    self.last_values = []

    for i in range(self.order):
        if i == 0:
            self.initial_values.append(X[0])
            if self.window_size is not None:
                self.pre_train_values.append(X[self.window_size - self.order])
            self.last_values.append(X[-1])
            X_diff = np.diff(X, n=1)
        else:
            self.initial_values.append(X_diff[0])
            if self.window_size is not None:
                self.pre_train_values.append(X_diff[self.window_size - self.order])
            self.last_values.append(X_diff[-1])
            X_diff = np.diff(X_diff, n=1)

    return self

transform

transform(X, y=None)

Transforms a time series into a differentiated time series of order n.

Parameters:

Name	Type	Description	Default
X	numpy ndarray	Time series to be differentiated.	required
y	Ignored	Not used, present here for API consistency by convention.	None

Returns:

Name	Type	Description
X_diff	numpy ndarray	Differentiated time series. The length of the array is the same as the original time series but the first n order values are nan.

Source code in skforecast/preprocessing/_preprocessing.py

@_check_X_numpy_ndarray_1d()
def transform(
    self, 
    X: np.ndarray, 
    y: Any = None
) -> np.ndarray:
    """
    Transforms a time series into a differentiated time series of order n.

    Parameters
    ----------
    X : numpy ndarray
        Time series to be differentiated.
    y : Ignored
        Not used, present here for API consistency by convention.

    Returns
    -------
    X_diff : numpy ndarray
        Differentiated time series. The length of the array is the same as
        the original time series but the first n `order` values are nan.

    """

    X_diff = np.diff(X, n=self.order)
    X_diff = np.append((np.full(shape=self.order, fill_value=np.nan)), X_diff)

    return X_diff

inverse_transform

inverse_transform(X, y=None)

Reverts the differentiation. To do so, the input array is assumed to be the same time series used to fit the transformer but differentiated.

Parameters:

Name	Type	Description	Default
X	numpy ndarray	Differentiated time series.	required
y	Ignored	Not used, present here for API consistency by convention.	None

Returns:

Name	Type	Description
X_diff	numpy ndarray	Reverted differentiated time series.

Source code in skforecast/preprocessing/_preprocessing.py

@_check_X_numpy_ndarray_1d()
def inverse_transform(
    self, 
    X: np.ndarray, 
    y: Any = None
) -> np.ndarray:
    """
    Reverts the differentiation. To do so, the input array is assumed to be
    the same time series used to fit the transformer but differentiated.

    Parameters
    ----------
    X : numpy ndarray
        Differentiated time series.
    y : Ignored
        Not used, present here for API consistency by convention.

    Returns
    -------
    X_diff : numpy ndarray
        Reverted differentiated time series.

    """

    # Remove initial nan values if present
    X = X[np.argmax(~np.isnan(X)):]
    for i in range(self.order):
        if i == 0:
            X_undiff = np.insert(X, 0, self.initial_values[-1])
            X_undiff = np.cumsum(X_undiff, dtype=float)
        else:
            X_undiff = np.insert(X_undiff, 0, self.initial_values[-(i + 1)])
            X_undiff = np.cumsum(X_undiff, dtype=float)

    return X_undiff

inverse_transform_training

inverse_transform_training(X, y=None)

Reverts the differentiation. To do so, the input array is assumed to be the differentiated training time series generated with the original time series used to fit the transformer.

When using a direct module Forecaster, the model in step 1 must be used if you want to reverse the differentiation of the training time series with the inverse_transform_training method.

Parameters:

Name	Type	Description	Default
X	numpy ndarray	Differentiated time series.	required
y	Ignored	Not used, present here for API consistency by convention.	None

Returns:

Name	Type	Description
X_diff	numpy ndarray	Reverted differentiated time series.

Source code in skforecast/preprocessing/_preprocessing.py

@_check_X_numpy_ndarray_1d()
def inverse_transform_training(
    self, 
    X: np.ndarray, 
    y: Any = None
) -> np.ndarray:
    """
    Reverts the differentiation. To do so, the input array is assumed to be
    the differentiated training time series generated with the original 
    time series used to fit the transformer.

    When using a `direct` module Forecaster, the model in step 1 must be 
    used if you want to reverse the differentiation of the training time 
    series with the `inverse_transform_training` method.

    Parameters
    ----------
    X : numpy ndarray
        Differentiated time series.
    y : Ignored
        Not used, present here for API consistency by convention.

    Returns
    -------
    X_diff : numpy ndarray
        Reverted differentiated time series.

    """

    if not self.pre_train_values:
        raise ValueError(
            "The `window_size` parameter must be set before fitting the "
            "transformer to revert the differentiation of the training "
            "time series."
        )

    # Remove initial nan values if present
    X = X[np.argmax(~np.isnan(X)):]
    for i in range(self.order):
        if i == 0:
            X_undiff = np.insert(X, 0, self.pre_train_values[-1])
            X_undiff = np.cumsum(X_undiff, dtype=float)
        else:
            X_undiff = np.insert(X_undiff, 0, self.pre_train_values[-(i + 1)])
            X_undiff = np.cumsum(X_undiff, dtype=float)

    # Remove initial values as they are not part of the training time series
    X_undiff = X_undiff[self.order:]

    return X_undiff

inverse_transform_next_window

inverse_transform_next_window(X, y=None)

Reverts the differentiation. The input array X is assumed to be a differentiated time series of order n that starts right after the the time series used to fit the transformer.

Parameters:

Name	Type	Description	Default
X	numpy ndarray	Differentiated time series. It is assumed o start right after the time series used to fit the transformer.	required
y	Ignored	Not used, present here for API consistency by convention.	None

Returns:

Name	Type	Description
X_undiff	numpy ndarray	Reverted differentiated time series.

Source code in skforecast/preprocessing/_preprocessing.py

@_check_X_numpy_ndarray_1d(ensure_1d=False)
def inverse_transform_next_window(
    self,
    X: np.ndarray,
    y: Any = None
) -> np.ndarray:
    """
    Reverts the differentiation. The input array `X` is assumed to be a 
    differentiated time series of order n that starts right after the
    the time series used to fit the transformer.

    Parameters
    ----------
    X : numpy ndarray
        Differentiated time series. It is assumed o start right after
        the time series used to fit the transformer.
    y : Ignored
        Not used, present here for API consistency by convention.

    Returns
    -------
    X_undiff : numpy ndarray
        Reverted differentiated time series.

    """

    array_ndim = X.ndim
    if array_ndim == 1:
        X = X[:, np.newaxis]

    # Remove initial rows with nan values if present
    X = X[~np.isnan(X).any(axis=1)]

    for i in range(self.order):
        if i == 0:
            X_undiff = np.cumsum(X, axis=0, dtype=float) + self.last_values[-1]
        else:
            X_undiff = np.cumsum(X_undiff, axis=0, dtype=float) + self.last_values[-(i + 1)]

    if array_ndim == 1:
        X_undiff = X_undiff.ravel()

    return X_undiff

set_params

set_params(**params)

Set the parameters of the TimeSeriesDifferentiator.

Parameters:

Name	Type	Description	Default
params	dict	A dictionary of the parameters to set.	{}

Returns:

Type	Description
None

Source code in skforecast/preprocessing/_preprocessing.py

def set_params(self, **params):
    """
    Set the parameters of the TimeSeriesDifferentiator.

    Parameters
    ----------
    params : dict
        A dictionary of the parameters to set.

    Returns
    -------
    None

    """

    for param, value in params.items():
        setattr(self, param, value)

skforecast.preprocessing._preprocessing.QuantileBinner

QuantileBinner(
    n_bins,
    method="linear",
    subsample=200000,
    dtype=np.float64,
    random_state=789654,
)

QuantileBinner class to bin data into quantile-based bins using numpy.percentile. This class is similar to KBinsDiscretizer but optimized for performance using numpy.searchsorted for fast bin assignment. Bin intervals are defined following the convention: bins[i-1] <= x < bins[i]. Values outside the range are clipped to the first or last bin. See more information in numpy.percentile and numpy.searchsorted.

Parameters:

Name	Type	Description	Default
n_bins	int	The number of quantile-based bins to create.	required
method	str	The method used to compute the quantiles. This parameter is passed to numpy.percentile. Default is 'linear'. Valid values are "inverse_cdf", "averaged_inverse_cdf", "closest_observation", "interpolated_inverse_cdf", "hazen", "weibull", "linear", "median_unbiased", "normal_unbiased".	'linear'
subsample	int	The number of samples to use for computing quantiles. If the dataset has more samples than subsample, a random subset will be used.	200000
dtype	data type	The data type to use for the bin indices. Default is numpy.float64.	numpy.float64
random_state	int	The random seed to use for generating a random subset of the data.	789654

Attributes:

Name	Type	Description
n_bins	int	The number of quantile-based bins to create.
method	str	The method used to compute the quantiles. This parameter is passed to numpy.percentile. Default is 'linear'. Valid values are 'linear', 'lower', 'higher', 'midpoint', 'nearest'.
subsample	int	The number of samples to use for computing quantiles. If the dataset has more samples than subsample, a random subset will be used.
dtype	data type	The data type to use for the bin indices. Default is numpy.float64.
random_state	int	The random seed to use for generating a random subset of the data.
n_bins_	int	The number of bins learned during fitting. This may be less than n_bins if there are duplicate bin edges due to repeated predicted values.
bin_edges_	numpy ndarray	The edges of the bins learned during fitting.
internal_edges_	numpy ndarray	The internal edges used for optimized bin assignment using numpy.searchsorted.
intervals_	dict	A dictionary with the bin indices as keys and the corresponding bin intervals as values.

Methods:

Name	Description
fit	Learn the bin edges based on quantiles from the training data.
transform	Assign new data to the learned bins.
fit_transform	Fit the model to the data and return the bin indices for the same data.
get_params	Get the parameters of the quantile binner.
set_params	Set the parameters of the QuantileBinner.

Source code in skforecast/preprocessing/_preprocessing.py

def __init__(
    self,
    n_bins: int,
    method: str = "linear",
    subsample: int = 200000,
    dtype: type = np.float64,
    random_state: int = 789654
) -> None:

    self._validate_params(
        n_bins,
        method,
        subsample,
        dtype,
        random_state
    )

    self.n_bins          = n_bins
    self.method          = method
    self.subsample       = subsample
    self.dtype           = dtype
    self.random_state    = random_state
    self.n_bins_         = None
    self.bin_edges_      = None
    self.internal_edges_ = None
    self.intervals_      = None

Attributes

n_bins instance-attribute

n_bins = n_bins

method instance-attribute

method = method

subsample instance-attribute

subsample = subsample

dtype instance-attribute

dtype = dtype

random_state instance-attribute

random_state = random_state

n_bins_ instance-attribute

n_bins_ = None

bin_edges_ instance-attribute

bin_edges_ = None

internal_edges_ instance-attribute

internal_edges_ = None

intervals_ instance-attribute

intervals_ = None

Methods:

_validate_params

_validate_params(
    n_bins, method, subsample, dtype, random_state
)

Validate the parameters passed to the class initializer.

Source code in skforecast/preprocessing/_preprocessing.py

def _validate_params(
    self,
    n_bins: int,
    method: str,
    subsample: int,
    dtype: type,
    random_state: int
):
    """
    Validate the parameters passed to the class initializer.
    """

    if not isinstance(n_bins, int) or n_bins < 2:
        raise ValueError(
            f"`n_bins` must be an int greater than 1. Got {n_bins}."
        )

    valid_methods = [
        "inverse_cdf",
        "averaged_inverse_cdf",
        "closest_observation",
        "interpolated_inverse_cdf",
        "hazen",
        "weibull",
        "linear",
        "median_unbiased",
        "normal_unbiased",
    ]
    if method not in valid_methods:
        raise ValueError(
            f"`method` must be one of {valid_methods}. Got {method}."
        )
    if not isinstance(subsample, int) or subsample < 1:
        raise ValueError(
            f"`subsample` must be an integer greater than or equal to 1. "
            f"Got {subsample}."
        )
    if not isinstance(random_state, int) or random_state < 0:
        raise ValueError(
            f"`random_state` must be an integer greater than or equal to 0. "
            f"Got {random_state}."
        )
    if not isinstance(dtype, type):
        raise ValueError(
            f"`dtype` must be a valid numpy dtype. Got {dtype}."
        )

fit

fit(X)

Learn the bin edges based on quantiles from the training data.

Parameters:

Name	Type	Description	Default
X	numpy ndarray	The training data used to compute the quantiles.	required

Returns:

Type	Description
None

Source code in skforecast/preprocessing/_preprocessing.py

def fit(self, X: np.ndarray):
    """
    Learn the bin edges based on quantiles from the training data.

    Parameters
    ----------
    X : numpy ndarray
        The training data used to compute the quantiles.

    Returns
    -------
    None

    """

    if X.size == 0:
        raise ValueError("Input data `X` cannot be empty.")
    if len(X) > self.subsample:
        rng = np.random.default_rng(self.random_state)
        X = X[rng.integers(0, len(X), self.subsample)]

    bin_edges = np.percentile(
        a      = X,
        q      = np.linspace(0, 100, self.n_bins + 1),
        method = self.method
    )

    # Remove duplicate edges (can happen when data has many repeated values)
    # to ensure bins are always numbered 0 to n_bins_-1
    self.bin_edges_ = np.unique(bin_edges)

    # Ensure at least 1 bin when all values are identical
    if len(self.bin_edges_) == 1:
        # Create artificial edges around the single value
        self.bin_edges_ = np.array([self.bin_edges_.item(), self.bin_edges_.item()])

    self.n_bins_ = len(self.bin_edges_) - 1

    if self.n_bins_ != self.n_bins:
        warnings.warn(
            f"The number of bins has been reduced from {self.n_bins} to "
            f"{self.n_bins_} due to duplicated edges caused by repeated predicted "
            f"values.",
            IgnoredArgumentWarning
        )

    # Internal edges for optimized transform with searchsorted
    self.internal_edges_ = self.bin_edges_[1:-1]
    self.intervals_ = {
        int(i): (float(self.bin_edges_[i]), float(self.bin_edges_[i + 1]))
        for i in range(self.n_bins_)
    }

transform

transform(X)

Assign new data to the learned bins.

Parameters:

Name	Type	Description	Default
X	numpy ndarray	The data to assign to the bins.	required

Returns:

Name	Type	Description
bin_indices	numpy ndarray	The indices of the bins each value belongs to. Values less than the smallest bin edge are assigned to the first bin, and values greater than the largest bin edge are assigned to the last bin.

Source code in skforecast/preprocessing/_preprocessing.py

def transform(self, X: np.ndarray):
    """
    Assign new data to the learned bins.

    Parameters
    ----------
    X : numpy ndarray
        The data to assign to the bins.

    Returns
    -------
    bin_indices : numpy ndarray 
        The indices of the bins each value belongs to.
        Values less than the smallest bin edge are assigned to the first bin,
        and values greater than the largest bin edge are assigned to the last bin.

    """

    if self.bin_edges_ is None:
        raise NotFittedError(
            "The model has not been fitted yet. Call 'fit' with training data first."
        )

    bin_indices = np.searchsorted(
        self.internal_edges_, X, side='right'
    ).astype(self.dtype, copy=False)

    return bin_indices

fit_transform

fit_transform(X)

Fit the model to the data and return the bin indices for the same data.

Parameters:

Name	Type	Description	Default
X	ndarray	The data to fit and transform.	required

Returns:

Name	Type	Description
bin_indices	ndarray	The indices of the bins each value belongs to. Values less than the smallest bin edge are assigned to the first bin, and values greater than the largest bin edge are assigned to the last bin.

Source code in skforecast/preprocessing/_preprocessing.py

def fit_transform(self, X):
    """
    Fit the model to the data and return the bin indices for the same data.

    Parameters
    ----------
    X : numpy.ndarray
        The data to fit and transform.

    Returns
    -------
    bin_indices : numpy.ndarray
        The indices of the bins each value belongs to.
        Values less than the smallest bin edge are assigned to the first bin,
        and values greater than the largest bin edge are assigned to the last bin.

    """

    self.fit(X)

    return self.transform(X)

get_params

get_params()

Get the parameters of the quantile binner.

Parameters:

Name	Type	Description	Default
self			required

Returns:

Name	Type	Description
params	dict	A dictionary of the parameters of the quantile binner.

Source code in skforecast/preprocessing/_preprocessing.py

def get_params(self):
    """
    Get the parameters of the quantile binner.

    Parameters
    ----------
    self

    Returns
    -------
    params : dict
        A dictionary of the parameters of the quantile binner.

    """

    return {
        "n_bins": self.n_bins,
        "method": self.method,
        "subsample": self.subsample,
        "dtype": self.dtype,
        "random_state": self.random_state,
    }

set_params

set_params(**params)

Set the parameters of the QuantileBinner.

Parameters:

Name	Type	Description	Default
params	dict	A dictionary of the parameters to set.	{}

Returns:

Type	Description
None

Source code in skforecast/preprocessing/_preprocessing.py

def set_params(self, **params):
    """
    Set the parameters of the QuantileBinner.

    Parameters
    ----------
    params : dict
        A dictionary of the parameters to set.

    Returns
    -------
    None

    """

    for param, value in params.items():
        setattr(self, param, value)

skforecast.preprocessing._calendar.create_calendar_features

create_calendar_features(
    X,
    features=None,
    features_to_encode=None,
    encoding="cyclical",
    max_values=None,
    spline_kwargs=None,
    keep_original_columns=True,
    tol=1e-12,
)

Extract datetime features from the DateTime index of a pandas DataFrame or Series.

Parameters:

Name	Type	Description	Default
X	pandas Series, pandas DataFrame, pandas DatetimeIndex	Input DataFrame or Series with a datetime index, or a pandas DatetimeIndex directly. When a DatetimeIndex is passed, it is used as the datetime source and there are no original columns to keep, so keep_original_columns has no effect.	required
features	list	List of calendar features (strings) to extract from the index. When None, the following features are extracted: 'year', 'month', 'week', 'day_of_week', 'day_of_month', 'day_of_year', 'weekend', 'hour', 'minute', 'second', 'quarter'.	None
features_to_encode	list	List of calendar features (strings) to encode. When None, all extracted features are encoded. If a feature is not in features, a ValueError is raised. If the explicit list contains features that cannot be encoded with the chosen encoding (e.g. 'year' or 'weekend', which are never encodable), an IgnoredArgumentWarning is issued and those features are kept as raw integers.	None
encoding	str	Encoding method for the extracted features. Options are None, 'cyclical', 'onehot' or 'spline'. Features that cannot be encoded under the chosen mode are kept as raw integers. By default, 'year' and 'weekend' are never encoded. 'onehot' excludes them via the known-category set, while 'cyclical' and 'spline' exclude them via max_values.	'cyclical'
max_values	dict	Dictionary of maximum values for the cyclical and spline encoding. User-provided values are merged with the defaults: keys passed by the user override the corresponding default, and missing keys fall back to the defaults {'month': 12, 'week': 53, 'day_of_week': 7, 'day_of_month': 31, 'day_of_year': 366, 'hour': 24, 'minute': 60, 'second': 60, 'quarter': 4}. For example, passing max_values={'month': 6} overrides only month; the other features keep their defaults. Features that are not in the defaults (e.g. 'year', 'weekend') are left as raw integers.	None
spline_kwargs	dict	Additional keyword arguments for the spline encoding. Only used when encoding='spline'. When None, defaults to {'degree': 3, 'include_bias': True, 'extrapolation': 'periodic'}; n_knots defaults to max_values[feature] + 1 per feature, which produces one spline column per distinct period value (analogous to a smooth one-hot encoding). Knots are placed uniformly over [min_val, min_val + max_val] (e.g. 1-13 for month, 0-24 for hour), giving a periodic period of exactly max_val for both 0-indexed and 1-indexed features and keeping the encoding stateless and consistent between training and prediction. Any keyword argument accepted by sklearn.preprocessing.SplineTransformer is allowed (e.g. n_knots, degree, include_bias, extrapolation, order) except knots (computed internally from max_values) and sparse_output (incompatible with the DataFrame output). Passing either of these or an unknown key raises ValueError.	None
keep_original_columns	bool	If True, the original columns of X are kept in the output DataFrame. If False, only the extracted datetime features are returned. When True and X is an unnamed pandas Series (X.name is None), a ValueError is raised; either set X.name to a string, or pass keep_original_columns=False. When X is a pandas DatetimeIndex this argument has no effect, as there are no original columns to keep.	True
tol	(float, None)	Absolute tolerance for clamping near-zero values produced by cyclical encoding to exactly 0. Floating point arithmetic can produce values such as -2.4e-16 instead of 0 (e.g. sin(2*pi*12/12)). When not None, any sin or cos value whose absolute value is smaller than tol is replaced with 0.0. Set to None to disable clamping. Only used when encoding='cyclical'.	1e-12

Returns:

Name	Type	Description
X_new	pandas DataFrame	DataFrame with the extracted (and optionally encoded) datetime features.

Notes

The default max_values use 53 for 'week' and 366 for 'day_of_year' to accommodate the maximum possible values across all calendar years: ISO week 53 occurs in some years (e.g. 2015, 2020, 2026) and day-of-year 366 occurs in leap years. Because the encoding must be stateless (the same for any year, without prior knowledge of whether it is a leap year or contains ISO week 53), the period is fixed at the maximum-possible value. This implies:

• Onehot: the week_53 and day_of_year_366 columns are always present in the output and equal 0 for rows whose year never reaches those values. This guarantees a consistent column schema across training and prediction.
• Cyclical / spline: in years where the maximum value is reached, the cyclical wrap-around is exact (e.g. sin(2π·366/366) = 0 matches sin(2π·0/366) = 0). In years where it is not, there is a one-step "phantom gap" between the highest observed value and 1. The cyclical distance is two steps instead of one. This residual asymmetry is numerically small (≈ 1.7% for day_of_year, ≈ 12% for week) and is strictly preferable to the alternative (period 52 / 365), which would silently collapse week 53 onto week 1 and day 366 onto day 1 in years where those values occur.

For high-cardinality features such as day_of_year (366 columns) or day_of_month (31 columns), encoding='cyclical' (2 columns per feature) or 'spline' (≈ max_val columns per feature, but dense rather than sparse) are typically more memory-efficient than 'onehot', especially on multi-year hourly or sub-daily data.

Output column order: encoded features appear in the order given by features. Non-encoded extracted features (e.g. 'year' or 'weekend', which are never encoded) appear first in the output, in features-list order; encoded features follow, also in features-list order. For example, with features=['year', 'month', 'weekend', 'hour'] and encoding='cyclical' the output columns are [year, weekend, month_sin, month_cos, hour_sin, hour_cos].

Source code in skforecast/preprocessing/_calendar.py

def create_calendar_features(
    X: pd.Series | pd.DataFrame | pd.DatetimeIndex,
    features: list[str] | None = None,
    features_to_encode: list[str] | None = None,
    encoding: str = "cyclical",
    max_values: dict[str, int] | None = None,
    spline_kwargs: dict | None = None,
    keep_original_columns: bool = True,
    tol: float | None = 1e-12,
) -> pd.DataFrame:
    """
    Extract datetime features from the DateTime index of a pandas DataFrame or Series.

    Parameters
    ----------
    X : pandas Series, pandas DataFrame, pandas DatetimeIndex
        Input DataFrame or Series with a datetime index, or a pandas
        DatetimeIndex directly. When a DatetimeIndex is passed, it is used as
        the datetime source and there are no original columns to keep, so
        `keep_original_columns` has no effect.
    features : list, default None
        List of calendar features (strings) to extract from the index. When
        `None`, the following features are extracted: `'year'`, `'month'`,
        `'week'`, `'day_of_week'`, `'day_of_month'`, `'day_of_year'`,
        `'weekend'`, `'hour'`, `'minute'`, `'second'`, `'quarter'`.
    features_to_encode : list, default None
        List of calendar features (strings) to encode. When `None`, all
        extracted features are encoded. If a feature is not in `features`, a
        `ValueError` is raised. If the explicit list contains features that
        cannot be encoded with the chosen `encoding` (e.g. `'year'` or
        `'weekend'`, which are never encodable), an `IgnoredArgumentWarning`
        is issued and those features are kept as raw integers.
    encoding : str, default 'cyclical'
        Encoding method for the extracted features. Options are `None`,
        `'cyclical'`, `'onehot'` or `'spline'`. Features that cannot be
        encoded under the chosen mode are kept as raw integers. By default,
        `'year'` and `'weekend'` are never encoded. `'onehot'` excludes them
        via the known-category set, while `'cyclical'` and `'spline'` exclude
        them via `max_values`.
    max_values : dict, default None
        Dictionary of maximum values for the cyclical and spline encoding.
        User-provided values are merged with the defaults: keys passed by
        the user override the corresponding default, and missing keys fall
        back to the defaults `{'month': 12, 'week': 53, 'day_of_week': 7,
        'day_of_month': 31, 'day_of_year': 366, 'hour': 24, 'minute': 60,
        'second': 60, 'quarter': 4}`. For example, passing
        `max_values={'month': 6}` overrides only `month`; the other features
        keep their defaults. Features that are not in the defaults (e.g.
        `'year'`, `'weekend'`) are left as raw integers.
    spline_kwargs : dict, default None
        Additional keyword arguments for the spline encoding. Only used when
        `encoding='spline'`. When `None`, defaults to `{'degree': 3,
        'include_bias': True, 'extrapolation': 'periodic'}`; `n_knots`
        defaults to `max_values[feature] + 1` per feature, which produces one
        spline column per distinct period value (analogous to a smooth
        one-hot encoding). Knots are placed uniformly over `[min_val,
        min_val + max_val]` (e.g. 1-13 for month, 0-24 for hour), giving a
        periodic period of exactly `max_val` for both 0-indexed and
        1-indexed features and keeping the encoding stateless and consistent
        between training and prediction. Any keyword argument accepted by
        `sklearn.preprocessing.SplineTransformer` is allowed (e.g. `n_knots`,
        `degree`, `include_bias`, `extrapolation`, `order`) except
        `knots` (computed internally from `max_values`) and `sparse_output`
        (incompatible with the DataFrame output). Passing either of these or
        an unknown key raises `ValueError`.
    keep_original_columns : bool, default True
        If True, the original columns of `X` are kept in the output
        DataFrame. If False, only the extracted datetime features are
        returned. When `True` and `X` is an unnamed pandas Series
        (`X.name is None`), a `ValueError` is raised; either set `X.name` to
        a string, or pass `keep_original_columns=False`. When `X` is a pandas
        DatetimeIndex this argument has no effect, as there are no original
        columns to keep.
    tol : float, None, default 1e-12
        Absolute tolerance for clamping near-zero values produced by cyclical
        encoding to exactly 0. Floating point arithmetic can produce values
        such as -2.4e-16 instead of 0 (e.g. sin(2*pi*12/12)). When not None,
        any sin or cos value whose absolute value is smaller than `tol` is
        replaced with 0.0. Set to None to disable clamping. Only used when
        `encoding='cyclical'`.

    Returns
    -------
    X_new : pandas DataFrame
        DataFrame with the extracted (and optionally encoded) datetime features.

    Notes
    -----
    The default `max_values` use 53 for `'week'` and 366 for `'day_of_year'`
    to accommodate the maximum possible values across all calendar years:
    ISO week 53 occurs in some years (e.g. 2015, 2020, 2026) and
    day-of-year 366 occurs in leap years. Because the encoding must be
    stateless (the same for any year, without prior knowledge of whether it
    is a leap year or contains ISO week 53), the period is fixed at the
    maximum-possible value. This implies:

    - Onehot: the `week_53` and `day_of_year_366` columns are always
      present in the output and equal 0 for rows whose year never reaches
      those values. This guarantees a consistent column schema across
      training and prediction.
    - Cyclical / spline: in years where the maximum value is reached,
      the cyclical wrap-around is exact (e.g. `sin(2π·366/366) = 0` matches
      `sin(2π·0/366) = 0`). In years where it is not, there is a one-step
      "phantom gap" between the highest observed value and 1. The
      cyclical distance is two steps instead of one. This residual
      asymmetry is numerically small (≈ 1.7% for `day_of_year`, ≈ 12% for
      `week`) and is strictly preferable to the alternative (period
      52 / 365), which would silently collapse week 53 onto week 1 and day
      366 onto day 1 in years where those values occur.

    For high-cardinality features such as `day_of_year` (366 columns) or
    `day_of_month` (31 columns), `encoding='cyclical'` (2 columns per
    feature) or `'spline'` (≈ `max_val` columns per feature, but dense
    rather than sparse) are typically more memory-efficient than
    `'onehot'`, especially on multi-year hourly or sub-daily data.

    Output column order: encoded features appear in the order given by
    `features`. Non-encoded extracted features (e.g. `'year'` or
    `'weekend'`, which are never encoded) appear first in the output, in
    `features`-list order; encoded features follow, also in
    `features`-list order. For example, with
    `features=['year', 'month', 'weekend', 'hour']` and
    `encoding='cyclical'` the output columns are
    `[year, weekend, month_sin, month_cos, hour_sin, hour_cos]`.

    """

    if not isinstance(X, (pd.DataFrame, pd.Series, pd.DatetimeIndex)):
        raise TypeError(
            "Input `X` must be a pandas Series, DataFrame or DatetimeIndex"
        )

    if isinstance(X, pd.DatetimeIndex):
        datetime_index = X
    else:
        if not isinstance(X.index, pd.DatetimeIndex):
            raise TypeError("Input `X` must have a pandas DatetimeIndex")
        datetime_index = X.index

    if len(X) == 0:
        raise ValueError("Cannot fit on empty input.")

    if encoding not in ["cyclical", "onehot", "spline", None]:
        raise ValueError(
            "Encoding must be one of 'cyclical', 'onehot', 'spline' or None"
        )

    if isinstance(X, pd.Series) and X.name is None and keep_original_columns:
        raise ValueError(
            "When `keep_original_columns=True`, the input Series must have a "
            "name (`X.name`). Either set `X.name` to a string, or pass "
            "`keep_original_columns=False`."
        )

    datetime_attrs = {
        "year": "year",
        "month": "month",
        "week": lambda idx: idx.isocalendar().week.astype(int),
        "day_of_week": "dayofweek",
        "day_of_month": "day",
        "day_of_year": "dayofyear",
        "weekend": lambda idx: (idx.dayofweek >= 5).astype(int),
        "hour": "hour",
        "minute": "minute",
        "second": "second",
        "quarter": "quarter",
    }
    if features is None:
        features = list(datetime_attrs.keys())

    resolved_max_values = _DEFAULT_MAX_VALUES.copy()
    if max_values is not None:
        unknown = set(max_values) - set(_DEFAULT_MAX_VALUES)
        if unknown:
            warnings.warn(
                f"Unknown keys in `max_values`: {sorted(unknown)}. "
                f"Valid keys: {sorted(_DEFAULT_MAX_VALUES)}. Unknown keys "
                f"are ignored.",
                IgnoredArgumentWarning,
            )
            max_values = {k: v for k, v in max_values.items() if k not in unknown}
        if max_values and encoding == "onehot":
            warnings.warn(
                "`max_values` is ignored when `encoding='onehot'`; onehot "
                "uses the fixed known-category set. Pass "
                "`encoding='cyclical'` or `encoding='spline'` for "
                "`max_values` to take effect.",
                IgnoredArgumentWarning,
            )
        resolved_max_values.update(max_values)
    max_values = resolved_max_values

    X_new = pd.DataFrame(index=datetime_index)

    not_supported_features = set(features) - set(datetime_attrs.keys())
    if not_supported_features:
        raise ValueError(
            f"Features {not_supported_features} are not supported. "
            f"Supported features are {list(datetime_attrs.keys())}."
        )

    for feature in features:
        attr = datetime_attrs[feature]
        X_new[feature] = (
            attr(datetime_index)
            if callable(attr)
            else getattr(datetime_index, attr).astype(int)
        )

    if features_to_encode is not None:
        not_supported_features_to_encode = set(features_to_encode) - set(features)
        if not_supported_features_to_encode:
            raise ValueError(
                f"Features {not_supported_features_to_encode} are not present in `features`."
            )

        if encoding is not None:
            if encoding == "onehot":
                encodable = set(_FEATURE_KNOWN_CATEGORIES.keys())
            else:  # encoding in ("cyclical", "spline")
                encodable = set(max_values.keys())
            not_encodable = [f for f in features_to_encode if f not in encodable]
            if not_encodable:
                warnings.warn(
                    f"Features {not_encodable} cannot be encoded with "
                    f"encoding={encoding!r}. Encodable features for this encoding "
                    f"are: {sorted(encodable)}. These features will be kept as "
                    f"raw integers.",
                    IgnoredArgumentWarning,
                )
    else:
        features_to_encode = features

    if encoding == "cyclical":
        cols_to_drop = []
        for feature in features:
            if feature in features_to_encode and feature in max_values:
                max_val = max_values[feature]
                sin_vals = np.sin(X_new[feature] * (2.0 * np.pi / max_val))
                cos_vals = np.cos(X_new[feature] * (2.0 * np.pi / max_val))
                if tol is not None:
                    sin_vals = np.where(np.abs(sin_vals) < tol, 0.0, sin_vals)
                    cos_vals = np.where(np.abs(cos_vals) < tol, 0.0, cos_vals)
                X_new[f"{feature}_sin"] = sin_vals
                X_new[f"{feature}_cos"] = cos_vals
                cols_to_drop.append(feature)
        X_new = X_new.drop(columns=cols_to_drop)
    elif encoding == "onehot":
        effective_encode = [
            f for f in features
            if f in features_to_encode and f in _FEATURE_KNOWN_CATEGORIES
        ]
        for feature in effective_encode:
            X_new[feature] = pd.Categorical(
                X_new[feature],
                categories=_FEATURE_KNOWN_CATEGORIES[feature],
            )
        if effective_encode:
            X_new = pd.get_dummies(
                X_new, columns=effective_encode, drop_first=False, sparse=False, dtype=int
            )
            # Match the cyclical / spline ordering: non-encoded features
            # first (in `features` order), then encoded dummies grouped per
            # feature (also in `features` order). `pd.get_dummies` keeps
            # non-encoded columns at their original position, so the dummies
            # are otherwise interleaved.
            non_encoded = [f for f in features if f not in effective_encode]
            encoded_cols = [
                f"{feature}_{cat}"
                for feature in effective_encode
                for cat in _FEATURE_KNOWN_CATEGORIES[feature]
            ]
            X_new = X_new[non_encoded + encoded_cols]
    elif encoding == "spline":
        if spline_kwargs is not None:
            invalid = set(spline_kwargs) - _SPLINE_ALLOWED_KWARGS
            if invalid:
                blocked_passed = invalid & _SPLINE_BLOCKED_KWARGS
                unknown = invalid - _SPLINE_BLOCKED_KWARGS
                msgs = []
                if blocked_passed:
                    msgs.append(
                        f"Keys {sorted(blocked_passed)} are not allowed in "
                        f"`spline_kwargs`: `knots` is computed internally from "
                        f"`max_values`, and `sparse_output` is incompatible "
                        f"with the DataFrame output."
                    )
                if unknown:
                    msgs.append(
                        f"Unknown keys in `spline_kwargs`: {sorted(unknown)}. "
                        f"Allowed keys: {sorted(_SPLINE_ALLOWED_KWARGS)}."
                    )
                raise ValueError(" ".join(msgs))

        resolved_spline_kwargs = {
            "degree": 3,
            "include_bias": True,
            "extrapolation": "periodic",
        }
        if spline_kwargs is not None:
            resolved_spline_kwargs.update(spline_kwargs)

        n_knots_global = resolved_spline_kwargs.pop("n_knots", None)
        cols_to_drop = []
        spline_cols = {}
        for feature in features:
            if feature in features_to_encode and feature in max_values:
                max_val = max_values[feature]
                n_knots = n_knots_global if n_knots_global is not None else max_val + 1
                min_val = _DEFAULT_MIN_VALUES.get(feature, 0)
                # Knots span [min_val, min_val + max_val] so that the periodic
                # period is exactly `max_val` for both 0-indexed (e.g. hour:
                # [0..24]) and 1-indexed features (e.g. month: [1..13]). With
                # the alternative `linspace(min_val, max_val, ...)`, 1-indexed
                # features would have period `max_val - min_val`, collapsing
                # the last value onto the first (e.g. month 12 = month 1).
                knots = np.linspace(min_val, min_val + max_val, n_knots).reshape(-1, 1)
                spt = SplineTransformer(
                          knots = knots,
                          **resolved_spline_kwargs,
                      )
                values = X_new[feature].to_numpy().reshape(-1, 1)
                spline_out = spt.fit_transform(values)
                col_names = spt.get_feature_names_out([feature])
                for col_name, col_values in zip(col_names, spline_out.T):
                    spline_cols[col_name] = col_values
                cols_to_drop.append(feature)

        X_new = X_new.drop(columns=cols_to_drop)
        X_new = pd.concat(
            [X_new, pd.DataFrame(spline_cols, index=X_new.index)], axis=1
        )

    if keep_original_columns and not isinstance(X, pd.DatetimeIndex):
        X_df = X.to_frame() if isinstance(X, pd.Series) else X
        overlapping_cols = set(X_df.columns).intersection(set(X_new.columns))
        if overlapping_cols:
            container = "Series" if isinstance(X, pd.Series) else "DataFrame"
            rename_target = "Series" if isinstance(X, pd.Series) else "columns"
            raise ValueError(
                f"The following extracted feature names already exist in the input "
                f"{container}: {list(overlapping_cols)}. To avoid duplicate columns, "
                f"rename the original {rename_target} or avoid extracting these features."
            )
        X_new = pd.concat([X_df, X_new], axis=1)

    return X_new

skforecast.preprocessing._calendar.calculate_distance_from_holiday

calculate_distance_from_holiday(
    X, holiday_column=None, date_column=None, fill_na=0
)

Calculate the number of periods to the next and since the last holiday.

The time unit used for the calculation (days, hours, minutes, …) is inferred from the frequency of the index when date_column=None, and is always days when a date column is used instead. Output columns are always named time_to_holiday and time_since_holiday regardless of the unit.

Parameters:

Name	Type	Description	Default
X	pandas Series, pandas DataFrame	Input data containing the holiday indicator. When a Series is passed, its values are used directly as the holiday indicator (boolean or 0/1) and holiday_column is ignored. When a DataFrame is passed, holiday_column must be specified.	required
holiday_column	(str, None)	Name of the boolean column indicating holidays (True or 1 on holiday dates, False or 0 otherwise). Required when X is a pandas DataFrame. Ignored when X is a pandas Series.	None
date_column	(str, None)	Name of the column containing dates to use as reference. When None, the index is used and must be a pandas DatetimeIndex.	None
fill_na	(int, float)	Value used to fill rows where no previous or next holiday exists (i.e. before the first holiday or after the last). Must be an int, a numpy.integer, or numpy.nan. Booleans and other floats are rejected because the output columns have Int64 dtype. Pass numpy.nan to keep those entries as pd.NA.	0

Returns:

Name	Type	Description
result	pandas DataFrame	DataFrame with two new columns and the same index as X: time_to_holiday: periods until the next holiday. time_since_holiday: periods since the last holiday.

Notes

When date_column is specified, the unit is always days regardless of the data frequency, because no index frequency information is available.

When date_column=None, the time unit is inferred from the index frequency:

• Daily or coarser (weekly, monthly, …): days
• Hourly: hours
• Minute: minutes
• Second: seconds
• Millisecond: milliseconds
• Microsecond: microseconds
• Nanosecond: nanoseconds

When the index has no frequency set, pd.infer_freq is attempted. If the frequency still cannot be determined (e.g. irregular spacing or fewer than three observations), the unit defaults to hours and a UserWarning is issued.

In date_column mode, fractional days (from arbitrary intra-day timestamps) are truncated, not rounded e.g. a distance of 0.99 days is reported as 0. To preserve sub-day precision, use the index-based mode with an appropriate frequency ('h', 'min', etc.).

When a row corresponds to a holiday, both time_to_holiday and time_since_holiday are 0, the date is at distance 0 from itself.

Source code in skforecast/preprocessing/_calendar.py

def calculate_distance_from_holiday(
    X: pd.DataFrame | pd.Series,
    holiday_column: str | None = None,
    date_column: str | None = None,
    fill_na: int | float = 0,
) -> pd.DataFrame:
    """
    Calculate the number of periods to the next and since the last holiday.

    The time unit used for the calculation (days, hours, minutes, …) is inferred
    from the frequency of the index when `date_column=None`, and is always days
    when a date column is used instead. Output columns are always named
    `time_to_holiday` and `time_since_holiday` regardless of the unit.

    Parameters
    ----------
    X : pandas Series, pandas DataFrame
        Input data containing the holiday indicator. When a Series is passed,
        its values are used directly as the holiday indicator (boolean or 0/1)
        and `holiday_column` is ignored. When a DataFrame is passed,
        `holiday_column` must be specified.
    holiday_column : str, None, default None
        Name of the boolean column indicating holidays (`True` or `1` on holiday
        dates, `False` or `0` otherwise). Required when `X` is a pandas
        DataFrame. Ignored when `X` is a pandas Series.
    date_column : str, None, default None
        Name of the column containing dates to use as reference. When `None`,
        the index is used and must be a pandas DatetimeIndex.
    fill_na : int, float, default 0
        Value used to fill rows where no previous or next holiday exists (i.e.
        before the first holiday or after the last). Must be an `int`, a
        `numpy.integer`, or `numpy.nan`. Booleans and other floats are
        rejected because the output columns have `Int64` dtype. Pass
        `numpy.nan` to keep those entries as `pd.NA`.

    Returns
    -------
    result : pandas DataFrame
        DataFrame with two new columns and the same index as `X`:

        - `time_to_holiday`: periods until the next holiday.
        - `time_since_holiday`: periods since the last holiday.

    Notes
    -----
    When `date_column` is specified, the unit is always days regardless of the
    data frequency, because no index frequency information is available.

    When `date_column=None`, the time unit is inferred from the index frequency:

    - Daily or coarser (weekly, monthly, …): days
    - Hourly: hours
    - Minute: minutes
    - Second: seconds
    - Millisecond: milliseconds
    - Microsecond: microseconds
    - Nanosecond: nanoseconds

    When the index has no frequency set, `pd.infer_freq` is attempted. If the
    frequency still cannot be determined (e.g. irregular spacing or fewer than
    three observations), the unit defaults to hours and a `UserWarning` is
    issued.

    In `date_column` mode, fractional days (from arbitrary intra-day
    timestamps) are truncated, not rounded e.g. a distance of 0.99
    days is reported as 0. To preserve sub-day precision, use the
    index-based mode with an appropriate frequency (`'h'`, `'min'`, etc.).

    When a row corresponds to a holiday, both `time_to_holiday` and
    `time_since_holiday` are 0, the date is at distance 0 from itself.

    """

    if not isinstance(X, (pd.DataFrame, pd.Series)):
        raise TypeError(
            "Input `X` must be a pandas Series or pandas DataFrame."
        )

    if isinstance(fill_na, bool) or not (
        isinstance(fill_na, (int, np.integer))
        or (isinstance(fill_na, (float, np.floating)) and np.isnan(fill_na))
    ):
        raise TypeError(
            "`fill_na` must be an int, np.integer, or numpy.nan, "
            f"got {type(fill_na).__name__}={fill_na!r}. The output columns "
            "have `Int64` dtype, so floats other than NaN cannot be used."
        )

    if isinstance(X, pd.Series):
        if holiday_column is not None:
            warnings.warn(
                "`holiday_column` is ignored when `X` is a pandas Series. "
                "The Series values are used directly as the holiday indicator.",
                IgnoredArgumentWarning,
                stacklevel=2,
            )
        col_name = X.name if X.name is not None else "is_holiday"
        X = X.rename(col_name).to_frame()
        holiday_column = col_name
    else:
        if holiday_column is None:
            raise ValueError(
                "`holiday_column` must be specified when `X` is a pandas DataFrame."
            )
        if holiday_column not in X.columns:
            raise ValueError(
                f"`holiday_column='{holiday_column}'` is not a column of `X`. "
                f"Available columns: {list(X.columns)}."
            )

    if date_column is not None:
        if date_column not in X.columns:
            raise ValueError(
                f"`date_column='{date_column}'` is not a column of `X`. "
                f"Available columns: {list(X.columns)}."
            )
        if pd.to_datetime(X[date_column], errors="coerce").isna().any():
            raise ValueError(
                f"`date_column='{date_column}'` contains NaN or unparseable "
                f"values. All entries must be valid datetimes."
            )

    if X[holiday_column].isna().any():
        warnings.warn(
            f"`{holiday_column}` contains NaN values. "
            f"They are filled with `False` (treated as non-holidays) "
            f"before computing distances.",
            UserWarning,
            stacklevel=2,
        )
        X = X.copy()
        with pd.option_context("future.no_silent_downcasting", True):
            X[holiday_column] = X[holiday_column].fillna(False).astype(bool)

    if date_column is None:
        if not isinstance(X.index, pd.DatetimeIndex):
            raise TypeError(
                "When `date_column=None`, the index must be a pandas DatetimeIndex."
            )
        dates = X.index.to_numpy()
        holiday_dates = X.index[X[holiday_column].astype(bool)].to_numpy()

        freq_str = X.index.freqstr if X.index.freq is not None else None
        if freq_str is None:
            freq_str = pd.infer_freq(X.index)
        if freq_str is None:
            warnings.warn(
                "Could not determine the frequency of the index. "
                "The output column unit defaults to 'hours'. To avoid this "
                "warning, either set `X.index.freq = pd.infer_freq(X.index)` "
                "or pass an index with a known frequency.",
                UserWarning,
                stacklevel=2,
            )
            freq_str = 'h'
        unit = _freq_to_timedelta_unit(freq_str)
    else:
        dates = pd.to_datetime(X[date_column]).to_numpy()
        holiday_dates = pd.to_datetime(
            X.loc[X[holiday_column].astype(bool), date_column]
        ).to_numpy()
        unit = 'D'

    holiday_dates_sorted = np.sort(holiday_dates)

    # Periods until the next holiday
    next_idx = np.searchsorted(holiday_dates_sorted, dates, side='left')
    has_next = next_idx < len(holiday_dates_sorted)
    to_holiday = np.full(len(dates), np.nan)
    to_holiday[has_next] = (
        holiday_dates_sorted[next_idx[has_next]] - dates[has_next]
    ).astype(f'timedelta64[{unit}]').astype(int)

    # Periods since the last holiday
    prev_idx = np.searchsorted(holiday_dates_sorted, dates, side='right') - 1
    has_prev = prev_idx >= 0
    since_holiday = np.full(len(dates), np.nan)
    since_holiday[has_prev] = (
        dates[has_prev] - holiday_dates_sorted[prev_idx[has_prev]]
    ).astype(f'timedelta64[{unit}]').astype(int)

    to_col = pd.Series(to_holiday, index=X.index, dtype="Int64").fillna(fill_na)
    since_col = pd.Series(since_holiday, index=X.index, dtype="Int64").fillna(fill_na)

    return pd.DataFrame(
        {"time_to_holiday": to_col, "time_since_holiday": since_col}
    )