feat(forecaster): quantile-LightGBM probabilistic head factory (#3)

_quarto.yml

@@ -696,6 +696,8 @@ quartodoc:
         - multitask.predict.PredictTask
         - multitask.clean.CleanTask
         - multitask.factories.default_lgbm_forecaster_factory
+        - multitask.factories.quantile_lgbm_forecaster_factory
+        - multitask.factories.predict_quantile_band
         - multitask.strategies.TrainingStrategy
         - multitask.strategies.LazyStrategy
         - multitask.strategies.DefaultsStrategy

docs/reference/index.qmd

@@ -182,6 +182,8 @@ strategies, and a one-call runner.
 | [multitask.predict.PredictTask](multitask.predict.PredictTask.qmd#spotforecast2_safe.multitask.predict.PredictTask) | Task 5 — Predict-only using previously saved models. |
 | [multitask.clean.CleanTask](multitask.clean.CleanTask.qmd#spotforecast2_safe.multitask.clean.CleanTask) | Cache-cleaning task — removes all cached data from the pipeline cache. |
 | [multitask.factories.default_lgbm_forecaster_factory](multitask.factories.default_lgbm_forecaster_factory.qmd#spotforecast2_safe.multitask.factories.default_lgbm_forecaster_factory) | Return a fresh, unfitted LightGBM ``ForecasterRecursive``. |
+| [multitask.factories.quantile_lgbm_forecaster_factory](multitask.factories.quantile_lgbm_forecaster_factory.qmd#spotforecast2_safe.multitask.factories.quantile_lgbm_forecaster_factory) | Return one quantile-regression LightGBM ``ForecasterRecursive`` per quantile. |
+| [multitask.factories.predict_quantile_band](multitask.factories.predict_quantile_band.qmd#spotforecast2_safe.multitask.factories.predict_quantile_band) | Assemble per-quantile forecasts into one non-crossing band. |
 | [multitask.strategies.TrainingStrategy](multitask.strategies.TrainingStrategy.qmd#spotforecast2_safe.multitask.strategies.TrainingStrategy) | Strategy interface for preparing a forecaster before the final fit. |
 | [multitask.strategies.LazyStrategy](multitask.strategies.LazyStrategy.qmd#spotforecast2_safe.multitask.strategies.LazyStrategy) | Approach 1 — Lazy fitting with optional cached tuning. |
 | [multitask.strategies.DefaultsStrategy](multitask.strategies.DefaultsStrategy.qmd#spotforecast2_safe.multitask.strategies.DefaultsStrategy) | Approach 2 — Train with defaults, no tuning, no cached params. |

docs/reference/multitask.factories.predict_quantile_band.qmd

@@ -0,0 +1,73 @@
+# multitask.factories.predict_quantile_band { #spotforecast2_safe.multitask.factories.predict_quantile_band }
+
+```python
+multitask.factories.predict_quantile_band(
+    forecasters,
+    steps,
+    *,
+    last_window=None,
+    exog=None,
+    enforce_monotonic=True,
+)
+```
+
+Assemble per-quantile forecasts into one non-crossing band.
+
+Calls ``predict`` on each fitted forecaster from
+:func:`quantile_lgbm_forecaster_factory` and stacks the results into columns
+named ``q_<level>`` (e.g. ``q_0.1``), in ascending quantile order. Because the
+quantile heads are fitted independently they can *cross* (a higher quantile
+predicting below a lower one); with ``enforce_monotonic`` the rows are sorted
+ascending (the Chernozhukov rearrangement), a deterministic post-hoc fix that
+restores monotonicity without changing the marginal quantile levels.
+
+## Parameters {.doc-section .doc-section-parameters}
+
+| Name              | Type                                                                                                                                  | Description                                                                                                                           | Default    |
+|-------------------|---------------------------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------|------------|
+| forecasters       | [Mapping](`typing.Mapping`)\[[float](`float`), [ForecasterRecursive](`spotforecast2_safe.forecaster.recursive.ForecasterRecursive`)\] | Map from quantile level to a **fitted** ``ForecasterRecursive`` (as returned by :func:`quantile_lgbm_forecaster_factory`, after fit). | _required_ |
+| steps             | [int](`int`)                                                                                                                          | Forecast horizon passed to each ``predict``.                                                                                          | _required_ |
+| last_window       | [Optional](`typing.Optional`)\[[Any](`typing.Any`)\]                                                                                  | Optional last-window override forwarded to ``predict``.                                                                               | `None`     |
+| exog              | [Optional](`typing.Optional`)\[[Any](`typing.Any`)\]                                                                                  | Optional exogenous frame forwarded to ``predict``.                                                                                    | `None`     |
+| enforce_monotonic | [bool](`bool`)                                                                                                                        | When ``True`` (default), sort each row ascending so the band never crosses.                                                           | `True`     |
+
+## Returns {.doc-section .doc-section-returns}
+
+| Name   | Type                                           | Description                                                           |
+|--------|------------------------------------------------|-----------------------------------------------------------------------|
+|        | [pd](`pandas`).[DataFrame](`pandas.DataFrame`) | pd.DataFrame: One column per quantile (``q_<level>``), indexed by the |
+|        | [pd](`pandas`).[DataFrame](`pandas.DataFrame`) | forecast horizon.                                                     |
+
+## Raises {.doc-section .doc-section-raises}
+
+| Name   | Type                       | Description                                          |
+|--------|----------------------------|------------------------------------------------------|
+|        | [ValueError](`ValueError`) | If *forecasters* is empty or its levels are invalid. |
+
+## Examples {.doc-section .doc-section-examples}
+
+```{python}
+import numpy as np
+import pandas as pd
+import types
+from spotforecast2_safe.multitask.factories import (
+    quantile_lgbm_forecaster_factory,
+    predict_quantile_band,
+)
+
+idx = pd.date_range("2023-01-01", periods=300, freq="h")
+y = pd.Series(
+    50 + 10 * np.sin(np.arange(300) * 2 * np.pi / 24), index=idx, name="y"
+)
+config = types.SimpleNamespace(
+    random_state=0, lags_consider=[1, 24], window_size=24
+)
+heads = quantile_lgbm_forecaster_factory(config, quantiles=[0.1, 0.5, 0.9])
+for fc in heads.values():
+    fc.fit(y=y)
+band = predict_quantile_band(heads, steps=6)
+print(band.columns.tolist())
+# Non-crossing after rearrangement.
+assert (band["q_0.1"] <= band["q_0.5"] + 1e-9).all()
+assert (band["q_0.5"] <= band["q_0.9"] + 1e-9).all()
+```

docs/reference/multitask.factories.quantile_lgbm_forecaster_factory.qmd

@@ -0,0 +1,58 @@
+# multitask.factories.quantile_lgbm_forecaster_factory { #spotforecast2_safe.multitask.factories.quantile_lgbm_forecaster_factory }
+
+```python
+multitask.factories.quantile_lgbm_forecaster_factory(
+    config,
+    *,
+    quantiles=DEFAULT_QUANTILES,
+    weight_func=None,
+    target=None,
+)
+```
+
+Return one quantile-regression LightGBM ``ForecasterRecursive`` per quantile.
+
+Each forecaster uses ``LGBMRegressor(objective="quantile", alpha=q)`` and the
+same lag/rolling configuration as :func:`default_lgbm_forecaster_factory`, so a
+caller can fit the lower/median/upper heads independently and assemble a band
+with :func:`predict_quantile_band`. Deterministic given ``config.random_state``;
+sf2-safe (LightGBM only, no torch/optuna). Refs ``hong16b``, ``roma19a``.
+
+## Parameters {.doc-section .doc-section-parameters}
+
+| Name        | Type                                                 | Description                                                                                                    | Default             |
+|-------------|------------------------------------------------------|----------------------------------------------------------------------------------------------------------------|---------------------|
+| config      | [Any](`typing.Any`)                                  | Object satisfying the ``PipelineConfig`` protocol; reads ``random_state``, ``lags_consider``, ``window_size``. | _required_          |
+| quantiles   | [Sequence](`typing.Sequence`)\[[float](`float`)\]    | Quantile levels in the open interval ``(0, 1)``. Defaults to ``(0.1, 0.5, 0.9)``.                              | `DEFAULT_QUANTILES` |
+| weight_func | [Optional](`typing.Optional`)\[[Any](`typing.Any`)\] | Optional per-sample weight function.                                                                           | `None`              |
+| target      | [Optional](`typing.Optional`)\[[str](`str`)\]        | Accepted and ignored (parity with the default factory).                                                        | `None`              |
+
+## Returns {.doc-section .doc-section-returns}
+
+| Name   | Type                                                                                                                            | Description                                                           |
+|--------|---------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------|
+|        | [Dict](`typing.Dict`)\[[float](`float`), [ForecasterRecursive](`spotforecast2_safe.forecaster.recursive.ForecasterRecursive`)\] | Dict[float, ForecasterRecursive]: Map from quantile level to a fresh, |
+|        | [Dict](`typing.Dict`)\[[float](`float`), [ForecasterRecursive](`spotforecast2_safe.forecaster.recursive.ForecasterRecursive`)\] | unfitted forecaster, in ascending quantile order.                     |
+
+## Raises {.doc-section .doc-section-raises}
+
+| Name   | Type                       | Description                                                    |
+|--------|----------------------------|----------------------------------------------------------------|
+|        | [ValueError](`ValueError`) | If *quantiles* is empty, out of ``(0, 1)``, or has duplicates. |
+
+## Examples {.doc-section .doc-section-examples}
+
+```{python}
+import types
+from spotforecast2_safe.multitask.factories import (
+    quantile_lgbm_forecaster_factory,
+)
+
+config = types.SimpleNamespace(
+    random_state=42, lags_consider=[1, 2, 3], window_size=3
+)
+heads = quantile_lgbm_forecaster_factory(config, quantiles=[0.1, 0.5, 0.9])
+print(sorted(heads))
+print(heads[0.1].regressor.get_params()["objective"])
+print(heads[0.1].regressor.get_params()["alpha"])
+```

src/spotforecast2_safe/multitask/factories.py

@@ -8,15 +8,28 @@
 a standalone function lets the upcoming ENTSO-E integration (and any future
 single-target task) supply its own factory via ``config.forecaster_factory``
 without subclassing ``BaseTask``.
+
+A second factory, :func:`quantile_lgbm_forecaster_factory`, builds a *probabilistic
+head*: one LightGBM ``ForecasterRecursive`` per quantile (``objective="quantile"``),
+giving native quantile regression for calibrated bands. It complements — and is
+distinct from — the residual-based ``predict_interval`` / ``predict_quantiles``
+already on ``ForecasterRecursive`` (which derive intervals by bootstrapping or
+split-conformal from a single point model). :func:`predict_quantile_band` assembles
+the per-quantile forecasts into one non-crossing band.
 """
 
-from typing import Any, Optional
+from typing import Any, Dict, Mapping, Optional, Sequence
 
+import numpy as np
+import pandas as pd
 from lightgbm import LGBMRegressor
 
 from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
 from spotforecast2_safe.preprocessing import RollingFeatures as RollingFeaturesUnified
 
+# Default lower/median/upper quantiles for the probabilistic head.
+DEFAULT_QUANTILES = (0.1, 0.5, 0.9)
+
 
 def default_lgbm_forecaster_factory(
     config: Any,
@@ -73,3 +86,158 @@ def default_lgbm_forecaster_factory(
         ),
         weight_func=weight_func,
     )
+
+
+def _validate_quantiles(quantiles: Sequence[float]) -> list:
+    """Return *quantiles* as a sorted list of floats in (0, 1), fail-safe."""
+    qs = [float(q) for q in quantiles]
+    if not qs:
+        raise ValueError("quantiles must be non-empty.")
+    if any(not (0.0 < q < 1.0) for q in qs):
+        raise ValueError(f"quantiles must lie in the open interval (0, 1); got {qs}.")
+    if len(set(qs)) != len(qs):
+        raise ValueError(f"quantiles must be unique; got {qs}.")
+    return sorted(qs)
+
+
+def quantile_lgbm_forecaster_factory(
+    config: Any,
+    *,
+    quantiles: Sequence[float] = DEFAULT_QUANTILES,
+    weight_func: Optional[Any] = None,
+    target: Optional[str] = None,
+) -> Dict[float, ForecasterRecursive]:
+    """Return one quantile-regression LightGBM ``ForecasterRecursive`` per quantile.
+
+    Each forecaster uses ``LGBMRegressor(objective="quantile", alpha=q)`` and the
+    same lag/rolling configuration as :func:`default_lgbm_forecaster_factory`, so a
+    caller can fit the lower/median/upper heads independently and assemble a band
+    with :func:`predict_quantile_band`. Deterministic given ``config.random_state``;
+    sf2-safe (LightGBM only, no torch/optuna). Refs ``hong16b``, ``roma19a``.
+
+    Args:
+        config: Object satisfying the ``PipelineConfig`` protocol; reads
+            ``random_state``, ``lags_consider``, ``window_size``.
+        quantiles: Quantile levels in the open interval ``(0, 1)``. Defaults to
+            ``(0.1, 0.5, 0.9)``.
+        weight_func: Optional per-sample weight function.
+        target: Accepted and ignored (parity with the default factory).
+
+    Returns:
+        Dict[float, ForecasterRecursive]: Map from quantile level to a fresh,
+        unfitted forecaster, in ascending quantile order.
+
+    Raises:
+        ValueError: If *quantiles* is empty, out of ``(0, 1)``, or has duplicates.
+
+    Examples:
+        ```{python}
+        import types
+        from spotforecast2_safe.multitask.factories import (
+            quantile_lgbm_forecaster_factory,
+        )
+
+        config = types.SimpleNamespace(
+            random_state=42, lags_consider=[1, 2, 3], window_size=3
+        )
+        heads = quantile_lgbm_forecaster_factory(config, quantiles=[0.1, 0.5, 0.9])
+        print(sorted(heads))
+        print(heads[0.1].regressor.get_params()["objective"])
+        print(heads[0.1].regressor.get_params()["alpha"])
+        ```
+    """
+    del target  # parity with default factory; not specialised per target
+    qs = _validate_quantiles(quantiles)
+    return {
+        q: ForecasterRecursive(
+            estimator=LGBMRegressor(
+                objective="quantile",
+                alpha=q,
+                random_state=config.random_state,
+                verbose=-1,
+            ),
+            lags=config.lags_consider[-1],
+            window_features=RollingFeaturesUnified(
+                stats=["mean"], window_sizes=config.window_size
+            ),
+            weight_func=weight_func,
+        )
+        for q in qs
+    }
+
+
+def predict_quantile_band(
+    forecasters: Mapping[float, ForecasterRecursive],
+    steps: int,
+    *,
+    last_window: Optional[Any] = None,
+    exog: Optional[Any] = None,
+    enforce_monotonic: bool = True,
+) -> pd.DataFrame:
+    """Assemble per-quantile forecasts into one non-crossing band.
+
+    Calls ``predict`` on each fitted forecaster from
+    :func:`quantile_lgbm_forecaster_factory` and stacks the results into columns
+    named ``q_<level>`` (e.g. ``q_0.1``), in ascending quantile order. Because the
+    quantile heads are fitted independently they can *cross* (a higher quantile
+    predicting below a lower one); with ``enforce_monotonic`` the rows are sorted
+    ascending (the Chernozhukov rearrangement), a deterministic post-hoc fix that
+    restores monotonicity without changing the marginal quantile levels.
+
+    Args:
+        forecasters: Map from quantile level to a **fitted** ``ForecasterRecursive``
+            (as returned by :func:`quantile_lgbm_forecaster_factory`, after fit).
+        steps: Forecast horizon passed to each ``predict``.
+        last_window: Optional last-window override forwarded to ``predict``.
+        exog: Optional exogenous frame forwarded to ``predict``.
+        enforce_monotonic: When ``True`` (default), sort each row ascending so the
+            band never crosses.
+
+    Returns:
+        pd.DataFrame: One column per quantile (``q_<level>``), indexed by the
+        forecast horizon.
+
+    Raises:
+        ValueError: If *forecasters* is empty or its levels are invalid.
+
+    Examples:
+        ```{python}
+        import numpy as np
+        import pandas as pd
+        import types
+        from spotforecast2_safe.multitask.factories import (
+            quantile_lgbm_forecaster_factory,
+            predict_quantile_band,
+        )
+
+        idx = pd.date_range("2023-01-01", periods=300, freq="h")
+        y = pd.Series(
+            50 + 10 * np.sin(np.arange(300) * 2 * np.pi / 24), index=idx, name="y"
+        )
+        config = types.SimpleNamespace(
+            random_state=0, lags_consider=[1, 24], window_size=24
+        )
+        heads = quantile_lgbm_forecaster_factory(config, quantiles=[0.1, 0.5, 0.9])
+        for fc in heads.values():
+            fc.fit(y=y)
+        band = predict_quantile_band(heads, steps=6)
+        print(band.columns.tolist())
+        # Non-crossing after rearrangement.
+        assert (band["q_0.1"] <= band["q_0.5"] + 1e-9).all()
+        assert (band["q_0.5"] <= band["q_0.9"] + 1e-9).all()
+        ```
+    """
+    qs = _validate_quantiles(list(forecasters.keys()))
+    columns = [f"q_{q}" for q in qs]
+    preds = []
+    index = None
+    for q in qs:
+        pred = forecasters[q].predict(steps=steps, last_window=last_window, exog=exog)
+        if index is None:
+            index = pred.index
+        preds.append(np.asarray(pred, dtype="float64"))
+
+    values = np.column_stack(preds)
+    if enforce_monotonic:
+        values = np.sort(values, axis=1)
+    return pd.DataFrame(values, index=index, columns=columns)