window_futures¶

def window_futures(
    X_future: pl.DataFrame,
    observation_times: pl.Series,
    forecasting_horizon: int,
    interval: str | timedelta,
    *,
    time_col: str = "time",
) -> pl.DataFrame:
    """Window known-future features into step-indexed columns.

    For each observation time T and forecast horizon H, extracts values at
    ``T + 1*interval`` through ``T + H*interval`` from ``X_future``, producing
    step-indexed columns ``<col>_step_1`` through ``<col>_step_H``.

    The output has one row per observation time and uses ``"time"`` as the
    time column (set to the observation time).

    Parameters
    ----------
    X_future : pl.DataFrame
        Known-future data with a ``time_col`` column and one or more value
        columns. Values are deterministic (e.g., holidays, day-of-week).
    observation_times : pl.Series
        Series of observation timestamps to window from.
    forecasting_horizon : int
        Number of forward steps (H) to extract per observation time.
    interval : str or timedelta
        Time frequency between steps (e.g., ``"1d"``, ``"1h"``).
    time_col : str, default="time"
        Name of the time column in ``X_future``.

    Returns
    -------
    pl.DataFrame
        Wide DataFrame with ``[time, <col>_step_1, ..., <col>_step_H]``.
        One row per unique observation time. Duplicate timestamps in
        ``observation_times`` are deduplicated, so the output contains at
        most one row per distinct observation time. This diverges from
        ``window_forecasts``, which preserves duplicate observation times.

    Raises
    ------
    ValueError
        If ``time_col`` is not in ``X_future``.
    ValueError
        If ``forecasting_horizon`` is not positive.
    ValueError
        If no value columns remain after removing ``time_col``.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> holidays = pl.DataFrame({
    ...     "time": [
    ...         datetime(2020, 1, 1),
    ...         datetime(2020, 1, 2),
    ...         datetime(2020, 1, 3),
    ...         datetime(2020, 1, 4),
    ...         datetime(2020, 1, 5),
    ...     ],
    ...     "is_holiday": [1, 0, 0, 1, 0],
    ... })
    >>> obs_times = pl.Series([datetime(2020, 1, 1), datetime(2020, 1, 2)])
    >>> window_futures(holidays, obs_times, forecasting_horizon=3, interval="1d")
    shape: (2, 4)
    ┌─────────────────────┬───────────────────┬───────────────────┬───────────────────┐
    │ time                ┆ is_holiday_step_1 ┆ is_holiday_step_2 ┆ is_holiday_step_3 │
    │ ---                 ┆ ---               ┆ ---               ┆ ---               │
    │ datetime[μs]        ┆ i64               ┆ i64               ┆ i64               │
    ╞═════════════════════╪═══════════════════╪═══════════════════╪═══════════════════╡
    │ 2020-01-01 00:00:00 ┆ 0                 ┆ 0                 ┆ 1                 │
    │ 2020-01-02 00:00:00 ┆ 0                 ┆ 1                 ┆ 0                 │
    └─────────────────────┴───────────────────┴───────────────────┴───────────────────┘

    """
    if time_col not in X_future.columns:
        msg = f"Column '{time_col}' not found in DataFrame. Available columns: {X_future.columns}"
        raise ValueError(msg)

    if forecasting_horizon < 1:
        msg = f"forecasting_horizon must be positive, got {forecasting_horizon}."
        raise ValueError(msg)

    value_cols = [c for c in X_future.columns if c != time_col]
    if not value_cols:
        msg = f"No value columns found. DataFrame has only '{time_col}'. At least one value column is required."
        raise ValueError(msg)

    if observation_times.is_empty():
        # No observation times: return empty frame with correct schema
        cols = {"time": pl.Series([], dtype=observation_times.dtype)}
        for col in value_cols:
            for step in range(1, forecasting_horizon + 1):
                cols[f"{col}_step_{step}"] = pl.Series([], dtype=X_future[col].dtype)
        return pl.DataFrame(cols)

    # Build a lookup table as the cross product of observation times and steps,
    # then compute each target time with a vectorized calendar-aware expression.
    steps = pl.DataFrame({"_step": range(1, forecasting_horizon + 1)})
    lookup = (
        pl
        .DataFrame({"vintage_time": observation_times})
        .join(steps, how="cross")
        .with_columns(_target_time_expr("vintage_time", "_step", interval).alias("time"))
    )

    # Join lookup with X_future on time to get values at each target time
    joined = lookup.join(
        X_future.select([time_col] + value_cols),
        left_on="time",
        right_on=time_col,
        how="left",
    )

    # Pivot to wide format
    pivot_exprs: list[pl.Expr] = []
    for step in range(1, forecasting_horizon + 1):
        for col in value_cols:
            pivot_exprs.append(
                pl.when(pl.col("_step") == step).then(pl.col(col)).otherwise(None).max().alias(f"{col}_step_{step}")
            )

    result = joined.group_by("vintage_time", maintain_order=True).agg(pivot_exprs)
    result = result.rename({"vintage_time": "time"})

    return result