[ENH] Add online regressor support and River adapter

docs/source/api_reference/tags.rst

@@ -90,6 +90,7 @@ probabilistic regressors.
     capability__multioutput
     capability__missing
     capability__update
+    capability__pred_int
     X_inner_mtype
     y_inner_mtype
     C_inner_mtype

skpro/benchmarking/evaluate.py

@@ -202,6 +202,8 @@ def evaluate(
 
     scoring = _check_scores(scoring)
 
+    _check_scoring_capability(estimator, scoring)
+
     _evaluate_fold_kwargs = {
         "estimator": estimator,
         "scoring": scoring,
@@ -433,3 +435,63 @@ def _check_scores(metrics):
         else:
             metrics_type[scitype].append(metric)
     return metrics_type
+
+
+def _check_scoring_capability(estimator, scoring):
+    """Raise if the metrics need predictions the estimator cannot make.
+
+    Some regressors only predict a single number, for example online
+    regressors adapted from ``river``. These have the ``capability:pred_int``
+    tag set to ``False``, and their ``predict_proba``, ``predict_interval``,
+    ``predict_quantiles`` and ``predict_var`` methods raise an exception.
+
+    Most metrics score a probabilistic prediction, so they cannot be used
+    with such a regressor. This is checked here, before any model is fitted,
+    because ``_evaluate_fold`` catches every exception and turns it into
+    ``error_score``. Without this check, ``evaluate`` would fit all folds
+    and then quietly return a table of ``NaN``.
+
+    Parameters
+    ----------
+    estimator : skpro regressor, BaseProbaRegressor descendant
+        regressor passed to ``evaluate``
+    scoring : dict
+        return of ``_check_scores``, metrics keyed by the ``scitype:y_pred``
+        tag, i.e., by the prediction method used to score them
+
+    Raises
+    ------
+    ValueError
+        if ``estimator`` has the ``capability:pred_int`` tag set to ``False``,
+        and any metric in ``scoring`` scores a probabilistic prediction
+    """
+    # metric scitypes which are scored against a probabilistic prediction
+    PROBA_SCITYPES = ["pred_proba", "pred_interval", "pred_quantiles"]
+
+    # non-skpro estimators carry no tags, their capability is not known here
+    if not hasattr(estimator, "get_tag"):
+        return
+
+    if estimator.get_tag("capability:pred_int", True, raise_error=False):
+        return
+
+    offenders = [
+        getattr(metric, "name", str(metric))
+        for scitype in PROBA_SCITYPES
+        for metric in scoring.get(scitype, [])
+    ]
+
+    # a point prediction only regressor can still be scored by point metrics
+    if not offenders:
+        return
+
+    raise ValueError(
+        f"Error in evaluate: the estimator {type(estimator).__name__} is "
+        "point prediction only, i.e., has the capability:pred_int tag set to "
+        "False, but the following metrics passed in scoring score "
+        f"probabilistic predictions: {', '.join(offenders)}. "
+        "To evaluate a point prediction regressor, wrap it in a regressor "
+        "which adds a distributional prediction, e.g., BaggingRegressor, "
+        "BootstrapRegressor, ResidualDouble, or EnbpiRegressor, and evaluate "
+        "the wrapped regressor."
+    )

skpro/model_selection/_tuning.py

@@ -80,6 +80,9 @@ def __init__(
             "capability:multioutput",
             "capability:missing",
             "capability:survival",
+            # prediction is delegated to best_estimator_, a clone of estimator,
+            # so the tuner is point prediction only iff estimator is
+            "capability:pred_int",
         ]
         self.clone_tags(estimator, tags_to_clone)
         self._set_update_capability_tag(estimator)
@@ -99,6 +102,33 @@ def _set_update_capability_tag(self, estimator):
                 "'no_update', 'inner_only', 'full_refit'."
             )
 
+    def _check_scoring_capability(self, scoring):
+        """Raise if ``scoring`` needs predictions the tuned estimator cannot make.
+
+        All ``skpro`` metrics score a probabilistic prediction, i.e., the return
+        of ``predict_proba``, ``predict_interval``, or ``predict_quantiles``.
+        Tuning an estimator which is point prediction only, i.e., which has the
+        ``capability:pred_int`` tag set to ``False``, is therefore not possible.
+        """
+        if self.get_tag("capability:pred_int"):
+            return
+
+        pred_scitype = None
+        if hasattr(scoring, "get_tag"):
+            pred_scitype = scoring.get_tag("scitype:y_pred", None, raise_error=False)
+
+        raise ValueError(
+            f"Error in {type(self).__name__}: the estimator to tune, "
+            f"{type(self.estimator).__name__}, is point prediction only, "
+            "i.e., has the capability:pred_int tag set to False, but the "
+            f"scoring metric {getattr(scoring, 'name', scoring)} scores "
+            f"probabilistic predictions of scitype {pred_scitype}. "
+            "To tune a point prediction regressor, wrap it in a regressor "
+            "which adds a distributional prediction, e.g., BaggingRegressor, "
+            "BootstrapRegressor, ResidualDouble, or EnbpiRegressor, and tune "
+            "the wrapped regressor."
+        )
+
     # attribute for _DelegatedProbaRegressor, which then delegates
     #     all non-overridden methods are same as of getattr(self, _delegate_name)
     #     see further details in _DelegatedProbaRegressor docstring
@@ -158,6 +188,8 @@ def _fit(self, X, y, C=None):
         scoring = self.scoring
         scoring_name = f"test_{scoring.name}"
 
+        self._check_scoring_capability(scoring)
+
         backend = self.backend
         backend_params = self.backend_params if self.backend_params else {}
 

skpro/registry/_tags.py

@@ -38,7 +38,6 @@
 check_tag_is_valid(tag_name, tag_value) - checks whether tag_value is valid for tag_name
 """
 
-
 import inspect
 import sys
 
@@ -275,6 +274,18 @@ class capability__update(_BaseTag):
     }
 
 
+class capability__pred_int(_BaseTag):
+    """Support for probabilistic prediction methods."""
+
+    _tags = {
+        "tag_name": "capability:pred_int",
+        "parent_type": "regressor_proba",
+        "tag_type": "bool",
+        "short_descr": "whether predict_proba, predict_interval, "
+        "and predict_quantiles are available",
+    }
+
+
 class X_inner_mtype(_BaseTag):
     """Internal X machine type."""
 

skpro/regression/adapters/__init__.py

@@ -1,2 +1,6 @@
-"""Adapters for probabilistic regressors."""
+"""Adapters for probabilistic and online regressors."""
 # copyright: skpro developers, BSD-3-Clause License (see LICENSE file)
+
+from skpro.regression.adapters._coerce import coerce_to_skpro_regressor
+
+__all__ = ["coerce_to_skpro_regressor"]

skpro/regression/adapters/_coerce.py

@@ -0,0 +1,33 @@
+"""Coerce foreign regressors to skpro online regressor adapters."""
+# copyright: skpro developers, BSD-3-Clause License (see LICENSE file)
+
+__author__ = ["patelchaitany"]
+
+from skpro.regression.adapters.river._utils import is_river_estimator
+from skpro.regression.base import BaseProbaRegressor
+
+
+def coerce_to_skpro_regressor(estimator):
+    """Wrap foreign online regressors for use in skpro meta-estimators.
+
+    Parameters
+    ----------
+    estimator : object
+        Estimator passed by the user to a meta-estimator such as
+        ``BaggingRegressor``.
+
+    Returns
+    -------
+    estimator : skpro regressor
+        ``estimator`` unchanged if already an skpro regressor; otherwise a
+        suitable adapter instance.
+    """
+    if isinstance(estimator, BaseProbaRegressor):
+        return estimator
+
+    if is_river_estimator(estimator):
+        from skpro.regression.adapters.river import RiverRegressor
+
+        return RiverRegressor(estimator)
+
+    return estimator

skpro/regression/adapters/river/__init__.py

@@ -0,0 +1,6 @@
+"""River adapters for online regressors."""
+# copyright: skpro developers, BSD-3-Clause License (see LICENSE file)
+
+from skpro.regression.adapters.river._river import RiverRegressor
+
+__all__ = ["RiverRegressor"]

skpro/regression/adapters/river/_clone.py

@@ -0,0 +1,24 @@
+"""Clone plugin for River estimators."""
+# copyright: skpro developers, BSD-3-Clause License (see LICENSE file)
+
+__author__ = ["patelchaitany"]
+
+from copy import deepcopy
+
+from skbase.base._clone_plugins import BaseCloner
+
+from skpro.regression.adapters.river._utils import is_river_estimator
+
+
+class _RiverDeepcopyCloner(BaseCloner):
+    """Clone River estimators via deepcopy.
+
+    River models do not implement sklearn-style ``get_params``; this plugin
+    allows ``RiverRegressor.clone()`` to produce independent River model copies.
+    """
+
+    def _check(self, obj):
+        return is_river_estimator(obj)
+
+    def _clone(self, obj):
+        return deepcopy(obj)

skpro/regression/adapters/river/_river.py

@@ -0,0 +1,121 @@
+"""Adapter for River online regressors."""
+# copyright: skpro developers, BSD-3-Clause License (see LICENSE file)
+
+__author__ = ["patelchaitany"]
+__all__ = ["RiverRegressor"]
+
+from copy import deepcopy
+
+import pandas as pd
+
+from skpro.regression.adapters.river._clone import _RiverDeepcopyCloner
+from skpro.regression.adapters.river._utils import (
+    _learn_batch,
+    _predict_batch,
+    is_river_estimator,
+)
+from skpro.regression.base import BaseProbaRegressor
+
+
+class RiverRegressor(BaseProbaRegressor):
+    """Adapter for River online regressors to the skpro point-prediction API.
+
+    Wraps a River regressor and exposes ``fit``, ``update``, and ``predict``
+    with skpro datatype conversion on the public interface. Probabilistic
+    prediction is intentionally not provided; use a meta-estimator such as
+    ``BaggingRegressor`` for distributional output.
+
+    On ``fit`` and ``update``, the River model is trained incrementally via
+    ``learn_many`` when available, otherwise ``learn_one``.
+
+    Parameters
+    ----------
+    estimator : river.base.Estimator
+        River regressor instance to wrap.
+
+    Attributes
+    ----------
+    estimator_ : river estimator
+        Deep copy of the River model after fitting.
+
+    Examples
+    --------
+    >>> from skpro.regression.adapters.river import RiverRegressor
+    >>> from river import linear_model
+    >>> from sklearn.datasets import load_diabetes
+    >>> from sklearn.model_selection import train_test_split
+    >>>
+    >>> X, y = load_diabetes(return_X_y=True, as_frame=True)
+    >>> X_train, X_test, y_train, y_test = train_test_split(X, y)
+    >>> y_train = y_train.to_frame("target")
+    >>>
+    >>> reg = RiverRegressor(linear_model.LinearRegression())
+    >>> reg.fit(X_train, y_train)
+    RiverRegressor(...)
+    >>> reg.update(X_test[:10], y_test[:10].to_frame("target"))
+    RiverRegressor(...)
+    >>> y_pred = reg.predict(X_test[10:])
+    """
+
+    _tags = {
+        "authors": ["patelchaitany"],
+        "maintainers": ["patelchaitany", "fkiraly"],
+        "object_type": "regressor_proba",
+        "estimator_type": "regressor",
+        "python_dependencies": "river",
+        "capability:update": True,
+        "capability:pred_int": False,
+        "capability:multioutput": False,
+        "capability:missing": False,
+        "tests:vm": True,
+    }
+
+    def __init__(self, estimator):
+        if not is_river_estimator(estimator):
+            raise TypeError(
+                "estimator must be a River estimator instance, "
+                f"but found type {type(estimator)}"
+            )
+        self.estimator = estimator
+        super().__init__()
+
+    @classmethod
+    def _get_clone_plugins(cls):
+        parent_plugins = super()._get_clone_plugins()
+        if parent_plugins is None:
+            parent_plugins = []
+        return [_RiverDeepcopyCloner] + list(parent_plugins)
+
+    def _fit(self, X, y, C=None):
+        if len(y.columns) != 1:
+            raise ValueError(
+                "RiverRegressor supports single-output regression only, "
+                f"but y has {len(y.columns)} columns."
+            )
+
+        self.estimator_ = deepcopy(self.estimator)
+        _learn_batch(self.estimator_, X, y)
+        return self
+
+    def _update(self, X, y, C=None):
+        if len(y.columns) != 1:
+            raise ValueError(
+                "RiverRegressor supports single-output regression only, "
+                f"but y has {len(y.columns)} columns."
+            )
+
+        _learn_batch(self.estimator_, X, y)
+        return self
+
+    def _predict(self, X):
+        preds = _predict_batch(self.estimator_, X)
+        columns = self._get_columns(method="predict")
+        return pd.DataFrame(preds, index=X.index, columns=columns)
+
+    @classmethod
+    def get_test_params(cls, parameter_set="default"):
+        """Return testing parameter settings for the estimator."""
+        from river import linear_model
+
+        params1 = {"estimator": linear_model.LinearRegression()}
+        return [params1]